353 lines
11 KiB
Python
353 lines
11 KiB
Python
"""Many videos at once: remove, move, and the Undo of each. Every test reads both lists after the call,
|
|
as `test_playlist_move` does: a half-applied move is the failure these routes exist to rule out."""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from app.catalog.models import Channel, Video
|
|
from app.reader import service
|
|
from app.reader.models import Playlist, PlaylistItem
|
|
|
|
|
|
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):
|
|
db.expire_all()
|
|
return list(
|
|
db.execute(
|
|
select(PlaylistItem.video_id)
|
|
.where(PlaylistItem.playlist_id == pl_id)
|
|
.order_by(PlaylistItem.position)
|
|
).scalars()
|
|
)
|
|
|
|
|
|
def _positions(db, pl_id):
|
|
db.expire_all()
|
|
return list(
|
|
db.execute(
|
|
select(PlaylistItem.position)
|
|
.where(PlaylistItem.playlist_id == pl_id)
|
|
.order_by(PlaylistItem.position)
|
|
).scalars()
|
|
)
|
|
|
|
|
|
def test_remove_bulk_takes_out_the_named_videos_and_counts_what_was_there(db, make_user, as_user):
|
|
user = make_user()
|
|
_seed_videos(db, ["a", "b", "c", "d"])
|
|
pl = _playlist(db, user, "L", ["a", "b", "c", "d"])
|
|
|
|
r = as_user(user).post(
|
|
f"/api/playlists/{pl.id}/items/remove-bulk", json={"video_ids": ["b", "d", "gone"]}
|
|
)
|
|
|
|
assert r.status_code == 200
|
|
assert r.json() == {"ok": True, "playlist_id": pl.id, "removed": 2}
|
|
assert _ids(db, pl.id) == ["a", "c"]
|
|
|
|
|
|
def test_remove_bulk_of_nothing_leaves_the_list_untouched(db, make_user, as_user):
|
|
user = make_user()
|
|
_seed_videos(db, ["a"])
|
|
pl = _playlist(db, user, "L", ["a"])
|
|
stamp = datetime.now(timezone.utc) - timedelta(days=3)
|
|
pl.updated_at = stamp
|
|
db.commit()
|
|
|
|
r = as_user(user).post(
|
|
f"/api/playlists/{pl.id}/items/remove-bulk", json={"video_ids": ["gone"]}
|
|
)
|
|
|
|
assert r.json()["removed"] == 0
|
|
db.expire_all()
|
|
assert db.get(Playlist, pl.id).updated_at == stamp
|
|
|
|
|
|
def test_restore_puts_the_videos_back_at_their_places(db, make_user, as_user):
|
|
user = make_user()
|
|
_seed_videos(db, ["a", "b", "c", "d"])
|
|
pl = _playlist(db, user, "L", ["b", "d"])
|
|
|
|
r = as_user(user).post(
|
|
f"/api/playlists/{pl.id}/items/restore",
|
|
json={"video_ids": ["c", "a"], "positions": [2, 0]},
|
|
)
|
|
|
|
assert r.json() == {"ok": True, "playlist_id": pl.id, "added": 2}
|
|
assert _ids(db, pl.id) == ["a", "b", "c", "d"]
|
|
assert _positions(db, pl.id) == [0, 1, 2, 3]
|
|
|
|
|
|
def test_restore_leaves_a_video_already_back_where_it_is_and_skips_one_the_catalog_lost(
|
|
db, make_user, as_user
|
|
):
|
|
user = make_user()
|
|
_seed_videos(db, ["a", "b"])
|
|
pl = _playlist(db, user, "L", ["b", "a"])
|
|
|
|
r = as_user(user).post(
|
|
f"/api/playlists/{pl.id}/items/restore",
|
|
json={"video_ids": ["a", "lost"], "positions": [0, 1]},
|
|
)
|
|
|
|
assert r.json()["added"] == 0
|
|
assert _ids(db, pl.id) == ["b", "a"]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"body",
|
|
[
|
|
{"video_ids": ["a"], "positions": []},
|
|
{"video_ids": ["a"], "positions": [0, 1]},
|
|
{"video_ids": ["a"], "positions": [-1]},
|
|
],
|
|
)
|
|
def test_restore_refuses_positions_that_do_not_pair(db, make_user, as_user, body):
|
|
user = make_user()
|
|
pl = _playlist(db, user, "L", [])
|
|
|
|
assert as_user(user).post(f"/api/playlists/{pl.id}/items/restore", json=body).status_code == 422
|
|
|
|
|
|
def test_move_bulk_lands_in_the_sources_order_whatever_order_they_were_picked_in(
|
|
db, make_user, as_user
|
|
):
|
|
user = make_user()
|
|
_seed_videos(db, ["a", "b", "c", "x"])
|
|
src = _playlist(db, user, "Source", ["a", "b", "c"])
|
|
dst = _playlist(db, user, "Target", ["x"])
|
|
|
|
r = as_user(user).post(
|
|
f"/api/playlists/{src.id}/items/move-bulk",
|
|
json={"video_ids": ["c", "a"], "target_playlist_id": dst.id},
|
|
)
|
|
|
|
assert r.status_code == 200
|
|
assert r.json() == {
|
|
"ok": True,
|
|
"from_playlist_id": src.id,
|
|
"to_playlist_id": dst.id,
|
|
"moved": ["a", "c"],
|
|
"already_there": [],
|
|
}
|
|
assert _ids(db, src.id) == ["b"]
|
|
assert _ids(db, dst.id) == ["x", "a", "c"]
|
|
|
|
|
|
def test_move_bulk_says_which_the_target_already_held_and_skips_one_already_gone(
|
|
db, make_user, as_user
|
|
):
|
|
user = make_user()
|
|
_seed_videos(db, ["a", "b", "c"])
|
|
src = _playlist(db, user, "Source", ["a", "b"])
|
|
dst = _playlist(db, user, "Target", ["b"])
|
|
|
|
r = as_user(user).post(
|
|
f"/api/playlists/{src.id}/items/move-bulk",
|
|
json={"video_ids": ["a", "b", "c"], "target_playlist_id": dst.id},
|
|
)
|
|
|
|
assert r.json()["moved"] == ["a"]
|
|
assert r.json()["already_there"] == ["b"]
|
|
assert _ids(db, src.id) == []
|
|
assert _ids(db, dst.id) == ["b", "a"]
|
|
|
|
|
|
def test_move_bulk_keeps_when_each_video_was_saved(db, make_user, as_user):
|
|
user = make_user()
|
|
_seed_videos(db, ["a"])
|
|
src = _playlist(db, user, "Source", ["a"])
|
|
dst = _playlist(db, user, "Target", [])
|
|
saved = datetime(2020, 1, 2, tzinfo=timezone.utc)
|
|
row = db.scalar(select(PlaylistItem).where(PlaylistItem.playlist_id == src.id))
|
|
row.added_at = saved
|
|
db.commit()
|
|
|
|
r = as_user(user).post(
|
|
f"/api/playlists/{src.id}/items/move-bulk",
|
|
json={"video_ids": ["a"], "target_playlist_id": dst.id},
|
|
)
|
|
assert r.status_code == 200
|
|
|
|
db.expire_all()
|
|
moved = db.scalar(select(PlaylistItem).where(PlaylistItem.playlist_id == dst.id))
|
|
assert moved.added_at == saved
|
|
|
|
|
|
def test_a_target_that_gained_nothing_does_not_rise_in_the_recently_changed_sort(
|
|
db, make_user, as_user
|
|
):
|
|
user = make_user()
|
|
_seed_videos(db, ["a"])
|
|
src = _playlist(db, user, "Source", ["a"])
|
|
dst = _playlist(db, user, "Target", ["a"])
|
|
stamp = datetime.now(timezone.utc) - timedelta(days=3)
|
|
dst.updated_at = stamp
|
|
db.commit()
|
|
|
|
r = as_user(user).post(
|
|
f"/api/playlists/{src.id}/items/move-bulk",
|
|
json={"video_ids": ["a"], "target_playlist_id": dst.id},
|
|
)
|
|
assert r.status_code == 200
|
|
|
|
db.expire_all()
|
|
assert db.get(Playlist, dst.id).updated_at == stamp
|
|
|
|
|
|
def test_move_bulk_into_the_same_list_is_refused(db, make_user, as_user):
|
|
user = make_user()
|
|
_seed_videos(db, ["a"])
|
|
pl = _playlist(db, user, "L", ["a"])
|
|
|
|
r = as_user(user).post(
|
|
f"/api/playlists/{pl.id}/items/move-bulk",
|
|
json={"video_ids": ["a"], "target_playlist_id": pl.id},
|
|
)
|
|
|
|
assert r.status_code == 400
|
|
assert r.json()["detail"] == {"code": "playlist_move", "reason": "same_playlist"}
|
|
assert _ids(db, pl.id) == ["a"]
|
|
|
|
|
|
def test_move_bulk_marks_a_linked_list_on_both_sides_as_unsynced(db, make_user, as_user):
|
|
user = make_user()
|
|
_seed_videos(db, ["a"])
|
|
src = _playlist(db, user, "Source", ["a"], yt_playlist_id="PLs")
|
|
dst = _playlist(db, user, "Target", [], yt_playlist_id="PLd")
|
|
|
|
r = as_user(user).post(
|
|
f"/api/playlists/{src.id}/items/move-bulk",
|
|
json={"video_ids": ["a"], "target_playlist_id": dst.id},
|
|
)
|
|
assert r.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_undo_move_restores_both_lists_as_they_stood(db, make_user, as_user):
|
|
user = make_user()
|
|
_seed_videos(db, ["a", "b", "c", "d"])
|
|
src = _playlist(db, user, "Source", ["a", "b", "c", "d"])
|
|
dst = _playlist(db, user, "Target", ["c"])
|
|
client = as_user(user)
|
|
|
|
moved = client.post(
|
|
f"/api/playlists/{src.id}/items/move-bulk",
|
|
json={"video_ids": ["a", "c"], "target_playlist_id": dst.id},
|
|
).json()
|
|
assert _ids(db, src.id) == ["b", "d"]
|
|
assert _ids(db, dst.id) == ["c", "a"]
|
|
|
|
r = client.post(
|
|
f"/api/playlists/{src.id}/items/undo-move",
|
|
json={
|
|
"from_playlist_id": dst.id,
|
|
"moved": moved["moved"],
|
|
"already_there": moved["already_there"],
|
|
"positions": [0, 2],
|
|
},
|
|
)
|
|
|
|
assert r.json() == {"ok": True, "playlist_id": src.id, "added": 2}
|
|
assert _ids(db, src.id) == ["a", "b", "c", "d"]
|
|
# The one the target held before the move stays there.
|
|
assert _ids(db, dst.id) == ["c"]
|
|
|
|
|
|
def test_undo_move_is_one_transaction(db, make_user, as_user, monkeypatch):
|
|
user = make_user()
|
|
_seed_videos(db, ["a"])
|
|
src = _playlist(db, user, "Source", [])
|
|
dst = _playlist(db, user, "Target", ["a"])
|
|
|
|
def boom(*a, **kw):
|
|
raise RuntimeError("second step failed")
|
|
|
|
monkeypatch.setattr(service, "_insert_at", boom)
|
|
client = as_user(user)
|
|
body = {"from_playlist_id": dst.id, "moved": ["a"], "already_there": [], "positions": [0]}
|
|
with pytest.raises(RuntimeError):
|
|
_ = client.post(f"/api/playlists/{src.id}/items/undo-move", json=body)
|
|
|
|
db.rollback()
|
|
assert _ids(db, dst.id) == ["a"]
|
|
assert _ids(db, src.id) == []
|
|
|
|
|
|
def test_undo_move_refuses_positions_that_do_not_pair(db, make_user, as_user):
|
|
user = make_user()
|
|
src = _playlist(db, user, "Source", [])
|
|
dst = _playlist(db, user, "Target", [])
|
|
|
|
r = as_user(user).post(
|
|
f"/api/playlists/{src.id}/items/undo-move",
|
|
json={"from_playlist_id": dst.id, "moved": ["a"], "already_there": ["b"], "positions": [0]},
|
|
)
|
|
|
|
assert r.status_code == 422
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("path", "body"),
|
|
[
|
|
("remove-bulk", {"video_ids": ["a"]}),
|
|
("restore", {"video_ids": ["a"], "positions": [0]}),
|
|
("move-bulk", {"video_ids": ["a"], "target_playlist_id": 0}),
|
|
("undo-move", {"from_playlist_id": 0, "moved": ["a"], "positions": [0]}),
|
|
],
|
|
)
|
|
def test_another_accounts_list_is_not_found(db, make_user, as_user, path, body):
|
|
owner, other = make_user(), make_user()
|
|
_seed_videos(db, ["a"])
|
|
theirs = _playlist(db, owner, "Theirs", ["a"])
|
|
mine = _playlist(db, other, "Mine", [])
|
|
body = {
|
|
k: (mine.id if k in ("target_playlist_id", "from_playlist_id") else v)
|
|
for k, v in body.items()
|
|
}
|
|
|
|
r = as_user(other).post(f"/api/playlists/{theirs.id}/items/{path}", json=body)
|
|
|
|
assert r.status_code == 404
|
|
assert _ids(db, theirs.id) == ["a"]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("path", "key"), [("move-bulk", "target_playlist_id"), ("undo-move", "from_playlist_id")]
|
|
)
|
|
def test_the_other_list_of_a_move_must_be_the_callers_too(db, make_user, as_user, path, key):
|
|
owner, other = make_user(), make_user()
|
|
_seed_videos(db, ["a"])
|
|
theirs = _playlist(db, owner, "Theirs", ["a"])
|
|
mine = _playlist(db, other, "Mine", ["a"])
|
|
body = {"video_ids": ["a"], "moved": ["a"], "positions": [0], key: theirs.id}
|
|
|
|
r = as_user(other).post(f"/api/playlists/{mine.id}/items/{path}", json=body)
|
|
|
|
assert r.status_code == 404
|
|
assert "move_target" in str(r.json()["detail"])
|
|
assert _ids(db, theirs.id) == ["a"]
|
|
assert _ids(db, mine.id) == ["a"]
|