peter
b636ef38bc
A read from before a write no longer brings a row back: a write calls off the read on its way, what lands while one is out stays off the screen, and the last write out does the refresh. A failure puts back its own rows, an accepted reorder brings back none, and an Undo reads past the cache. A list the cache has dropped is read before a removal, so its Undo has places. undo-move refuses the same list twice, positions cannot repeat, and undo-move and the Plex key bodies refuse a body without their videos. Focus goes to the header's box when the last shown row or Save this order goes.
641 lines
21 KiB
Python
641 lines
21 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 queries, service
|
|
from app.reader.models import Playlist, PlaylistItem
|
|
from app.sync.playlists import current_fingerprint
|
|
|
|
|
|
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": ["b", "d"]}
|
|
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"] == []
|
|
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_keeps_when_each_video_was_saved(db, make_user, as_user):
|
|
user = make_user()
|
|
_seed_videos(db, ["a"])
|
|
src = _playlist(db, user, "Source", [])
|
|
dst = _playlist(db, user, "Target", ["a"])
|
|
saved = datetime(2020, 1, 2, tzinfo=timezone.utc)
|
|
row = db.scalar(select(PlaylistItem).where(PlaylistItem.playlist_id == dst.id))
|
|
row.added_at = saved
|
|
db.commit()
|
|
|
|
r = as_user(user).post(
|
|
f"/api/playlists/{src.id}/items/undo-move",
|
|
json={"from_playlist_id": dst.id, "moved": ["a"], "positions": [0]},
|
|
)
|
|
|
|
assert r.json()["added"] == 1
|
|
db.expire_all()
|
|
back = db.scalar(select(PlaylistItem).where(PlaylistItem.playlist_id == src.id))
|
|
assert back.added_at == saved
|
|
|
|
|
|
def test_undo_move_leaves_a_video_that_moved_on_where_it_is(db, make_user, as_user):
|
|
"""A to B, then B to C, then the first Undo: the video is not in the list the Undo takes it from,
|
|
and coming back would leave it in two lists the reader never put it in."""
|
|
user = make_user()
|
|
_seed_videos(db, ["a", "b"])
|
|
src = _playlist(db, user, "Source", [])
|
|
dst = _playlist(db, user, "Target", ["b"])
|
|
third = _playlist(db, user, "Third", ["a"])
|
|
|
|
r = as_user(user).post(
|
|
f"/api/playlists/{src.id}/items/undo-move",
|
|
json={"from_playlist_id": dst.id, "moved": ["a", "b"], "positions": [0, 1]},
|
|
)
|
|
|
|
assert r.json()["added"] == 1
|
|
assert _ids(db, src.id) == ["b"]
|
|
assert _ids(db, dst.id) == []
|
|
assert _ids(db, third.id) == ["a"]
|
|
|
|
|
|
def test_an_undo_closes_up_the_places_of_a_video_that_stayed_out(db, make_user, as_user):
|
|
"""Each index counts the rows put back before it, so one that stays out moves the rest up."""
|
|
user = make_user()
|
|
_seed_videos(db, ["a", "b", "c", "d"])
|
|
src = _playlist(db, user, "Source", ["c", "d"])
|
|
dst = _playlist(db, user, "Target", ["b"])
|
|
|
|
r = as_user(user).post(
|
|
f"/api/playlists/{src.id}/items/undo-move",
|
|
json={"from_playlist_id": dst.id, "moved": ["a", "b"], "positions": [0, 1]},
|
|
)
|
|
|
|
assert r.json()["added"] == 1
|
|
assert _ids(db, src.id) == ["b", "c", "d"]
|
|
|
|
|
|
def test_a_restore_closes_up_the_places_of_a_video_the_catalog_lost(db, make_user, as_user):
|
|
user = make_user()
|
|
_seed_videos(db, ["b", "c", "d"])
|
|
pl = _playlist(db, user, "L", ["c", "d"])
|
|
|
|
r = as_user(user).post(
|
|
f"/api/playlists/{pl.id}/items/restore",
|
|
json={"video_ids": ["lost", "b"], "positions": [0, 1]},
|
|
)
|
|
|
|
assert r.json()["added"] == 1
|
|
assert _ids(db, pl.id) == ["b", "c", "d"]
|
|
|
|
|
|
def test_undo_move_takes_a_video_out_of_the_target_when_the_source_has_it_again(
|
|
db, make_user, as_user
|
|
):
|
|
user = make_user()
|
|
_seed_videos(db, ["a"])
|
|
src = _playlist(db, user, "Source", ["a"])
|
|
dst = _playlist(db, user, "Target", ["a"])
|
|
|
|
r = as_user(user).post(
|
|
f"/api/playlists/{src.id}/items/undo-move",
|
|
json={"from_playlist_id": dst.id, "moved": ["a"], "positions": [0]},
|
|
)
|
|
|
|
assert r.json()["added"] == 0
|
|
assert _ids(db, src.id) == ["a"]
|
|
assert _ids(db, dst.id) == []
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("path", "body", "gone", "reason"),
|
|
[
|
|
("move-bulk", {"video_ids": ["a"], "target_playlist_id": "dst"}, "dst", "move_target"),
|
|
("move-bulk", {"video_ids": ["a"], "target_playlist_id": "dst"}, "src", "playlist"),
|
|
(
|
|
"undo-move",
|
|
{"from_playlist_id": "dst", "moved": ["a"], "positions": [0]},
|
|
"src",
|
|
"playlist",
|
|
),
|
|
(
|
|
"undo-move",
|
|
{"from_playlist_id": "dst", "moved": ["a"], "positions": [0]},
|
|
"dst",
|
|
"move_target",
|
|
),
|
|
],
|
|
)
|
|
def test_a_list_deleted_before_the_lock_is_not_found(
|
|
db, make_user, as_user, monkeypatch, path, body, gone, reason
|
|
):
|
|
user = make_user()
|
|
_seed_videos(db, ["a"])
|
|
src = _playlist(db, user, "Source", ["a"])
|
|
dst = _playlist(db, user, "Target", ["a"])
|
|
ids = {"src": src.id, "dst": dst.id}
|
|
body = {k: (dst.id if v == "dst" else v) for k, v in body.items()}
|
|
# Another tab's delete committed between the route's read and this lock.
|
|
monkeypatch.setattr(service, "lock_playlists", lambda _db, *got: set(got) - {ids[gone]})
|
|
|
|
r = as_user(user).post(f"/api/playlists/{src.id}/items/{path}", json=body)
|
|
|
|
assert r.status_code == 404
|
|
assert reason in str(r.json()["detail"])
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("method", "path", "body"),
|
|
[
|
|
("post", "items", {"video_id": "a"}),
|
|
("delete", "items/a", None),
|
|
("post", "items/remove-bulk", {"video_ids": ["a"]}),
|
|
("post", "items/restore", {"video_ids": ["a"], "positions": [0]}),
|
|
],
|
|
)
|
|
def test_a_one_list_write_finds_its_list_under_the_lock(
|
|
db, make_user, as_user, monkeypatch, method, path, body
|
|
):
|
|
"""Read with its lock, so a delete that won the row answers 404 rather than a foreign key's 500."""
|
|
user = make_user()
|
|
_seed_videos(db, ["a"])
|
|
pl = _playlist(db, user, "L", [])
|
|
real = queries.own_playlist
|
|
|
|
def locked_read_finds_nothing(db_, user_, playlist_id, *, lock=False):
|
|
return None if lock else real(db_, user_, playlist_id)
|
|
|
|
monkeypatch.setattr(queries, "own_playlist", locked_read_finds_nothing)
|
|
kw = {"json": body} if body is not None else {}
|
|
|
|
r = getattr(as_user(user), method)(f"/api/playlists/{pl.id}/{path}", **kw)
|
|
|
|
assert r.status_code == 404
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("path", "body"),
|
|
[
|
|
("items/remove-bulk", {"ids": ["a"]}),
|
|
("items/move-bulk", {"ids": ["a"], "target_playlist_id": 1}),
|
|
("items/undo-move", {"ids": ["a"], "from_playlist_id": 1}),
|
|
],
|
|
)
|
|
def test_a_body_without_its_videos_is_refused_rather_than_acting_on_none(
|
|
db, make_user, as_user, path, body
|
|
):
|
|
user = make_user()
|
|
pl = _playlist(db, user, "L", [])
|
|
|
|
assert as_user(user).post(f"/api/playlists/{pl.id}/{path}", json=body).status_code == 422
|
|
|
|
|
|
def test_an_undo_of_only_videos_the_target_held_names_no_moved_ones(db, make_user, as_user):
|
|
"""An empty `moved` is a real body: every video of the move was already in the target."""
|
|
user = make_user()
|
|
_seed_videos(db, ["a"])
|
|
src = _playlist(db, user, "Source", [])
|
|
dst = _playlist(db, user, "Target", ["a"])
|
|
|
|
r = as_user(user).post(
|
|
f"/api/playlists/{src.id}/items/undo-move",
|
|
json={"from_playlist_id": dst.id, "moved": [], "already_there": ["a"], "positions": [0]},
|
|
)
|
|
|
|
assert r.json()["added"] == 1
|
|
assert _ids(db, src.id) == ["a"]
|
|
assert _ids(db, dst.id) == ["a"]
|
|
|
|
|
|
def test_an_undo_within_one_list_is_refused_and_keeps_its_videos(db, make_user, as_user):
|
|
"""A move refuses the same list twice, so no Undo of one exists; a row must not go for asking."""
|
|
user = make_user()
|
|
_seed_videos(db, ["a", "b"])
|
|
pl = _playlist(db, user, "L", ["a", "b"])
|
|
|
|
r = as_user(user).post(
|
|
f"/api/playlists/{pl.id}/items/undo-move",
|
|
json={"from_playlist_id": pl.id, "moved": ["a"], "positions": [0]},
|
|
)
|
|
|
|
assert r.status_code == 400
|
|
assert r.json()["detail"] == {"code": "playlist_move", "reason": "same_playlist"}
|
|
assert _ids(db, pl.id) == ["a", "b"]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("path", "key", "extra"),
|
|
[
|
|
("items/restore", "video_ids", {}),
|
|
("items/undo-move", "moved", {"from_playlist_id": None}),
|
|
],
|
|
)
|
|
def test_two_videos_cannot_ask_for_one_place(db, make_user, as_user, path, key, extra):
|
|
user = make_user()
|
|
_seed_videos(db, ["a", "b", "c"])
|
|
pl = _playlist(db, user, "L", ["c"])
|
|
other = _playlist(db, user, "Other", ["a", "b"])
|
|
body = {key: ["a", "b"], "positions": [0, 0]}
|
|
body.update({k: other.id for k in extra})
|
|
|
|
r = as_user(user).post(f"/api/playlists/{pl.id}/{path}", json=body)
|
|
|
|
assert r.status_code == 422
|
|
assert _ids(db, pl.id) == ["c"]
|
|
|
|
|
|
def test_a_reorder_without_its_videos_is_refused(db, make_user, as_user):
|
|
user = make_user()
|
|
pl = _playlist(db, user, "L", [])
|
|
|
|
assert as_user(user).put(f"/api/playlists/{pl.id}/order", json={"ids": []}).status_code == 422
|
|
|
|
|
|
def _synced(db, pl):
|
|
pl.synced_fingerprint = current_fingerprint(db, pl)
|
|
pl.dirty = False
|
|
db.commit()
|
|
|
|
|
|
def test_a_bulk_remove_and_its_undo_leave_a_linked_list_as_it_was(db, make_user, as_user):
|
|
user = make_user()
|
|
_seed_videos(db, ["a", "b"])
|
|
pl = _playlist(db, user, "L", ["a", "b"], yt_playlist_id="PLx")
|
|
_synced(db, pl)
|
|
client = as_user(user)
|
|
|
|
r = client.post(f"/api/playlists/{pl.id}/items/remove-bulk", json={"video_ids": ["a"]})
|
|
assert r.status_code == 200
|
|
db.expire_all()
|
|
assert db.get(Playlist, pl.id).dirty is True
|
|
|
|
r = client.post(
|
|
f"/api/playlists/{pl.id}/items/restore", json={"video_ids": ["a"], "positions": [0]}
|
|
)
|
|
assert r.status_code == 200
|
|
db.expire_all()
|
|
assert db.get(Playlist, pl.id).dirty is False
|
|
|
|
|
|
def test_a_move_and_its_undo_leave_both_linked_lists_as_they_were(db, make_user, as_user):
|
|
user = make_user()
|
|
_seed_videos(db, ["a", "b"])
|
|
src = _playlist(db, user, "Source", ["a", "b"], yt_playlist_id="PLs")
|
|
dst = _playlist(db, user, "Target", ["b"], yt_playlist_id="PLd")
|
|
_synced(db, src)
|
|
_synced(db, dst)
|
|
client = as_user(user)
|
|
|
|
moved = client.post(
|
|
f"/api/playlists/{src.id}/items/move-bulk",
|
|
json={"video_ids": ["a"], "target_playlist_id": dst.id},
|
|
).json()
|
|
assert moved["moved"] == ["a"]
|
|
db.expire_all()
|
|
assert db.get(Playlist, src.id).dirty is True
|
|
assert db.get(Playlist, dst.id).dirty is True
|
|
|
|
r = client.post(
|
|
f"/api/playlists/{src.id}/items/undo-move",
|
|
json={"from_playlist_id": dst.id, "moved": ["a"], "positions": [0]},
|
|
)
|
|
assert r.status_code == 200
|
|
db.expire_all()
|
|
assert db.get(Playlist, src.id).dirty is False
|
|
assert db.get(Playlist, dst.id).dirty is False
|
|
|
|
|
|
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"]
|