peter
c5f121139b
A season marked watched, a pull from Plex and a YouTube resume point made their first row by a read then an add, and failed on the unique key when a first checkpoint or the watched mark made it in between. They go through the same insert-then-read as the checkpoint (the resume point is one upsert); a pull that meets a local change still waiting for its push leaves it. Folding duplicate Watch later lists moved their videos but not a Home pin, which the cascade then deleted; the first pin of a dropped copy moves onto the kept list. Marking one notice read trimmed before its own flag was flushed, so the trim left one past the cap. The route test now crosses the cap and reads the result from a session of its own.
91 lines
3.6 KiB
Python
91 lines
3.6 KiB
Python
"""Migration 0112's data step, DB-lane: a user who already has two Watch later lists keeps the oldest,
|
|
with the other's videos moved in after its own (one it already held is not added twice) and a Home
|
|
pin of the other moved onto it, and the one-per-user index then refuses a second list.
|
|
|
|
The schema half is checked by test_migrations.py; a video or a pin lost on the way is invisible there.
|
|
"""
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
from alembic import command
|
|
from alembic.config import Config
|
|
from sqlalchemy import text
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
from app.catalog.models import Channel, Video
|
|
from app.db import engine
|
|
|
|
|
|
def _lists(conn, user_id: int, n: int) -> list[int]:
|
|
values = ", ".join(f"(:u, 'Watch later', 'watch_later', {i})" for i in range(n))
|
|
return list(
|
|
conn.execute(
|
|
text(
|
|
f"INSERT INTO playlists (user_id, name, kind, position) VALUES {values} RETURNING id"
|
|
),
|
|
{"u": user_id},
|
|
).scalars()
|
|
)
|
|
|
|
|
|
def _pin(conn, user_id: int, playlist_id: int, position: int) -> None:
|
|
conn.execute(
|
|
text("INSERT INTO comfort_pins (user_id, playlist_id, position) VALUES (:u, :p, :pos)"),
|
|
{"u": user_id, "p": playlist_id, "pos": position},
|
|
)
|
|
|
|
|
|
def test_a_second_watch_later_is_folded_into_the_first(db, make_user):
|
|
user, other = make_user(), make_user()
|
|
db.add(Channel(id="UCm", title="M"))
|
|
now = datetime.now(timezone.utc)
|
|
for v in ("v1", "v2", "v3"):
|
|
db.add(Video(id=v, channel_id="UCm", published_at=now, duration_seconds=60))
|
|
db.commit()
|
|
cfg = Config("alembic.ini")
|
|
command.downgrade(cfg, "0111_audit_request_id")
|
|
try:
|
|
with engine.begin() as conn:
|
|
first, second, third = _lists(conn, user.id, 3)
|
|
theirs, their_second = _lists(conn, other.id, 2)
|
|
conn.execute(
|
|
text(
|
|
"INSERT INTO playlist_items (playlist_id, video_id, position) VALUES"
|
|
" (:a, 'v1', 0), (:b, 'v2', 0), (:b, 'v1', 1), (:c, 'v3', 0)"
|
|
),
|
|
{"a": first, "b": second, "c": theirs},
|
|
)
|
|
# The user pinned two of the copies, the kept one neither: the first on the shelf moves.
|
|
_pin(conn, user.id, third, 0)
|
|
_pin(conn, user.id, second, 1)
|
|
# The other pinned both: the kept one's pin stays, and is the only one.
|
|
_pin(conn, other.id, their_second, 0)
|
|
_pin(conn, other.id, theirs, 1)
|
|
finally:
|
|
command.upgrade(cfg, "head")
|
|
|
|
with engine.connect() as conn:
|
|
lists = conn.execute(
|
|
text("SELECT id, user_id FROM playlists WHERE kind = 'watch_later' ORDER BY id")
|
|
).all()
|
|
assert [tuple(r) for r in lists] == [(first, user.id), (theirs, other.id)]
|
|
items = conn.execute(
|
|
text(
|
|
"SELECT playlist_id, video_id, position FROM playlist_items ORDER BY playlist_id, position"
|
|
)
|
|
).all()
|
|
assert [tuple(r) for r in items] == [(first, "v1", 0), (first, "v2", 1), (theirs, "v3", 0)]
|
|
pins = conn.execute(
|
|
text("SELECT user_id, playlist_id, position FROM comfort_pins ORDER BY user_id")
|
|
).all()
|
|
assert [tuple(r) for r in pins] == [(user.id, first, 0), (other.id, theirs, 1)]
|
|
with pytest.raises(IntegrityError), engine.begin() as conn:
|
|
conn.execute(
|
|
text(
|
|
"INSERT INTO playlists (user_id, name, kind, position)"
|
|
" VALUES (:u, 'Watch later', 'watch_later', 2)"
|
|
),
|
|
{"u": user.id},
|
|
)
|