feat(e10): make "move a video to another list" one atomic request

The one-step move S2 needs has no honest client-side shape: "add to the target, then
remove from the source" leaves the video in BOTH lists whenever the second call fails,
and the reverse order loses it entirely. So it is one endpoint in one transaction —
POST /api/playlists/{id}/items/{video_id}/move — which re-parents the existing item row
(so `added_at` survives), recomputes `dirty` on both sides, and is idempotent when the
target already holds the video.

Also publishes `updated_at` on the playlist summary, and bumps it explicitly for the
edits that only write `playlist_items` (add / remove / reorder / move). UpdatedAtMixin's
`onupdate` fires on an UPDATE of the playlist row, so without `_touch` the rail's coming
"recently changed" sort would have seen renames and nothing else.
This commit is contained in:
2026-09-05 19:59:41 +02:00
parent 7e97559615
commit b28be682fe
4 changed files with 282 additions and 0 deletions
+190
View File
@@ -0,0 +1,190 @@
"""E10 S2 — the one-step "move this video to another list" endpoint, and the `updated_at` touch
the rail's "recently changed" sort reads.
Two properties are load-bearing and neither is visible from the happy path alone:
1. **The move is one transaction.** The client-side shape it replaces (add to the target, then
remove from the source) leaves the video in BOTH lists when the second call fails. So the
tests assert the *pair* of lists after every call, never just the destination.
2. **`updated_at` tracks item edits.** `UpdatedAtMixin.onupdate` only fires on an UPDATE of the
playlist row, and add/remove/reorder/move write `playlist_items` — so the timestamp is bumped
explicitly, and a regression there would silently degrade the sort to "renames only".
"""
from datetime import datetime, timedelta, timezone
from app.models import Channel, Playlist, PlaylistItem, Video
def _seed_videos(db, ids):
db.add(Channel(id="UCtest", title="Test"))
now = datetime.now(timezone.utc)
for v in ids:
db.add(Video(id=v, channel_id="UCtest", published_at=now))
db.commit()
def _playlist(db, user, name, video_ids, **kw):
pl = Playlist(user_id=user.id, name=name, **kw)
db.add(pl)
db.flush()
for i, vid in enumerate(video_ids):
db.add(PlaylistItem(playlist_id=pl.id, video_id=vid, position=i))
db.commit()
return pl
def _ids(db, pl_id):
return [
it.video_id
for it in db.execute(
PlaylistItem.__table__.select()
.where(PlaylistItem.playlist_id == pl_id)
.order_by(PlaylistItem.position)
).all()
]
def _move(client, src, video_id, dst):
return client.post(
f"/api/playlists/{src}/items/{video_id}/move", json={"target_playlist_id": dst}
)
def test_move_leaves_the_video_in_exactly_one_list(db, make_user, as_user):
user = make_user()
_seed_videos(db, ["v1", "v2"])
src = _playlist(db, user, "Source", ["v1", "v2"])
dst = _playlist(db, user, "Target", [])
r = _move(as_user(user), src.id, "v1", dst.id)
assert r.status_code == 200
assert r.json() == {
"ok": True,
"from_playlist_id": src.id,
"to_playlist_id": dst.id,
"video_id": "v1",
}
db.expire_all()
assert _ids(db, src.id) == ["v2"]
assert _ids(db, dst.id) == ["v1"]
def test_move_appends_to_the_end_of_the_target(db, make_user, as_user):
user = make_user()
_seed_videos(db, ["v1", "v2", "v3"])
src = _playlist(db, user, "Source", ["v3"])
dst = _playlist(db, user, "Target", ["v1", "v2"])
assert _move(as_user(user), src.id, "v3", dst.id).status_code == 200
db.expire_all()
assert _ids(db, dst.id) == ["v1", "v2", "v3"]
def test_move_into_a_list_that_already_has_it_is_a_removal_not_a_duplicate(db, make_user, as_user):
user = make_user()
_seed_videos(db, ["v1"])
src = _playlist(db, user, "Source", ["v1"])
dst = _playlist(db, user, "Target", ["v1"])
assert _move(as_user(user), src.id, "v1", dst.id).status_code == 200
db.expire_all()
assert _ids(db, src.id) == []
assert _ids(db, dst.id) == ["v1"] # exactly one row, not two
def test_move_to_the_same_list_is_refused(db, make_user, as_user):
user = make_user()
_seed_videos(db, ["v1"])
src = _playlist(db, user, "Source", ["v1"])
r = _move(as_user(user), src.id, "v1", src.id)
assert r.status_code == 400
assert r.json()["detail"] == {"code": "playlist_move", "reason": "same_playlist"}
db.expire_all()
assert _ids(db, src.id) == ["v1"]
def test_move_of_a_video_that_left_the_source_reports_it(db, make_user, as_user):
user = make_user()
_seed_videos(db, ["v1"])
src = _playlist(db, user, "Source", [])
dst = _playlist(db, user, "Target", [])
r = _move(as_user(user), src.id, "v1", dst.id)
assert r.status_code == 404
assert r.json()["detail"] == {"code": "playlist_move", "reason": "not_in_playlist"}
def test_move_into_someone_elses_list_404s_and_changes_nothing(db, make_user, as_user):
user, other = make_user(), make_user()
_seed_videos(db, ["v1"])
src = _playlist(db, user, "Mine", ["v1"])
theirs = _playlist(db, other, "Theirs", [])
assert _move(as_user(user), src.id, "v1", theirs.id).status_code == 404
db.expire_all()
assert _ids(db, src.id) == ["v1"]
assert _ids(db, theirs.id) == []
def test_move_marks_both_linked_sides_dirty(db, make_user, as_user):
"""A move edits two YouTube mirrors at once — the source lost an item, the target gained one.
Marking only the source is the asymmetric fix this repo keeps re-learning."""
user = make_user()
_seed_videos(db, ["v1"])
src = _playlist(db, user, "Src", ["v1"], yt_playlist_id="PLsrc", synced_video_ids='["v1"]')
dst = _playlist(db, user, "Dst", [], yt_playlist_id="PLdst", synced_video_ids="[]")
# Baseline: both are in sync with YouTube before the move.
src.dirty = False
dst.dirty = False
db.commit()
assert _move(as_user(user), src.id, "v1", dst.id).status_code == 200
db.expire_all()
assert db.get(Playlist, src.id).dirty is True
assert db.get(Playlist, dst.id).dirty is True
def test_item_edits_bump_updated_at(db, make_user, as_user):
"""The rail's "recently changed first" sort reads this. Adding, moving and reordering all
write `playlist_items` only, so nothing here is covered by the ORM's `onupdate`."""
user = make_user()
_seed_videos(db, ["v1", "v2"])
src = _playlist(db, user, "Source", ["v1", "v2"])
dst = _playlist(db, user, "Target", [])
old = datetime.now(timezone.utc) - timedelta(days=3)
src.updated_at = old
dst.updated_at = old
db.commit()
client = as_user(user)
assert client.post(f"/api/playlists/{dst.id}/items", json={"video_id": "v1"}).status_code == 200
db.expire_all()
assert db.get(Playlist, dst.id).updated_at > old
dst.updated_at = old
db.commit()
assert (
client.put(f"/api/playlists/{src.id}/order", json={"video_ids": ["v2", "v1"]}).status_code
== 200
)
db.expire_all()
assert db.get(Playlist, src.id).updated_at > old
src.updated_at = old
dst.updated_at = old
db.commit()
assert _move(client, src.id, "v2", dst.id).status_code == 200
db.expire_all()
# BOTH sides of a move are "recently changed" — the target is where the reader will look.
assert db.get(Playlist, src.id).updated_at > old
assert db.get(Playlist, dst.id).updated_at > old
def test_the_list_endpoint_publishes_updated_at(db, make_user, as_user):
user = make_user()
_playlist(db, user, "Solo", [])
rows = as_user(user).get("/api/playlists").json()
assert len(rows) == 1
assert rows[0]["updated_at"] # the rail cannot sort by a field the API never sends