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.
135 lines
4.7 KiB
Python
135 lines
4.7 KiB
Python
"""Putting removed Plex playlist rows back WHERE THEY WERE: the list page's "Undo".
|
|
|
|
A removal can take one row, a season or a whole show, and a season's episodes need not sit together
|
|
(a reorder inside the show interleaves them), so each key carries its own index — over the ORDERED
|
|
rows, since positions have gaps after a removal. One request: an append plus a second reorder call
|
|
leaves the rows at the end whenever that call fails, and would overwrite another tab's changes.
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from app.plex.models import PlexItem, PlexLibrary, PlexPlaylist, PlexPlaylistItem
|
|
|
|
|
|
def _seed(db, user, rks, *, positions=None, extra=("x", "y")):
|
|
lib = PlexLibrary(plex_key="p1", title="Movies", kind="movie", enabled=True)
|
|
db.add(lib)
|
|
db.flush()
|
|
items = {}
|
|
for rk in [*rks, *extra]:
|
|
it = PlexItem(rating_key=rk, library_id=lib.id, kind="movie", title=rk.upper())
|
|
db.add(it)
|
|
items[rk] = it
|
|
pl = PlexPlaylist(user_id=user.id, title="L")
|
|
db.add(pl)
|
|
db.flush()
|
|
for i, rk in enumerate(rks):
|
|
pos = (positions or range(99))[i]
|
|
db.add(PlexPlaylistItem(playlist_id=pl.id, item_id=items[rk].id, position=pos))
|
|
db.commit()
|
|
return pl
|
|
|
|
|
|
def _order(db, pid):
|
|
db.expire_all()
|
|
rows = db.execute(
|
|
PlexPlaylistItem.__table__.select()
|
|
.where(PlexPlaylistItem.playlist_id == pid)
|
|
.order_by(PlexPlaylistItem.position, PlexPlaylistItem.id)
|
|
).all()
|
|
by_id = {it.id: it.rating_key for it in db.query(PlexItem).all()}
|
|
return [by_id[r.item_id] for r in rows], [r.position for r in rows]
|
|
|
|
|
|
def _add(client, pid, keys, positions=None):
|
|
body = {"item_rating_keys": keys}
|
|
if positions is not None:
|
|
body["positions"] = positions
|
|
return client.post(f"/api/plex/playlists/{pid}/items/bulk", json=body)
|
|
|
|
|
|
def test_one_row_goes_back_to_its_index(db, make_user, as_user):
|
|
user = make_user()
|
|
pl = _seed(db, user, ["a", "b", "c"])
|
|
r = _add(as_user(user), pl.id, ["x"], [1])
|
|
assert r.status_code == 200
|
|
assert r.json()["added"] == 1
|
|
assert _order(db, pl.id) == (["a", "x", "b", "c"], [0, 1, 2, 3])
|
|
|
|
|
|
def test_scattered_rows_go_back_to_each_of_their_indices(db, make_user, as_user):
|
|
# A season whose episodes were interleaved with another season's: removed rows at 1 and 3 of
|
|
# a, x, b, y, c. Given in any order, each lands where it stood.
|
|
user = make_user()
|
|
pl = _seed(db, user, ["a", "b", "c"])
|
|
_add(as_user(user), pl.id, ["y", "x"], [3, 1])
|
|
assert _order(db, pl.id)[0] == ["a", "x", "b", "y", "c"]
|
|
|
|
|
|
def test_indices_count_rows_not_stored_numbers(db, make_user, as_user):
|
|
user = make_user()
|
|
pl = _seed(db, user, ["a", "b", "c"], positions=[0, 7, 20])
|
|
_add(as_user(user), pl.id, ["x"], [2])
|
|
assert _order(db, pl.id) == (["a", "b", "x", "c"], [0, 1, 2, 3])
|
|
|
|
|
|
def test_an_index_past_the_end_appends(db, make_user, as_user):
|
|
user = make_user()
|
|
pl = _seed(db, user, ["a"])
|
|
_add(as_user(user), pl.id, ["x"], [50])
|
|
assert _order(db, pl.id)[0] == ["a", "x"]
|
|
|
|
|
|
def test_a_row_already_in_the_list_does_not_move(db, make_user, as_user):
|
|
user = make_user()
|
|
pl = _seed(db, user, ["a", "b", "c"])
|
|
r = _add(as_user(user), pl.id, ["c"], [0])
|
|
assert r.json()["added"] == 0
|
|
assert _order(db, pl.id)[0] == ["a", "b", "c"]
|
|
|
|
|
|
def test_without_positions_it_still_appends(db, make_user, as_user):
|
|
user = make_user()
|
|
pl = _seed(db, user, ["a", "b"])
|
|
_add(as_user(user), pl.id, ["y", "x"])
|
|
assert _order(db, pl.id)[0] == ["a", "b", "y", "x"]
|
|
|
|
|
|
def test_positions_must_pair_with_the_keys(db, make_user, as_user):
|
|
user = make_user()
|
|
pl = _seed(db, user, ["a"])
|
|
assert _add(as_user(user), pl.id, ["x", "y"], [0]).status_code == 422
|
|
assert _add(as_user(user), pl.id, ["x"], [-1]).status_code == 422
|
|
assert _order(db, pl.id)[0] == ["a"]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("method", "path"),
|
|
[("post", "items/bulk"), ("post", "items/remove-bulk"), ("put", "order")],
|
|
)
|
|
def test_a_body_without_its_keys_is_refused_rather_than_acting_on_none(
|
|
db, make_user, as_user, method, path
|
|
):
|
|
user = make_user()
|
|
pl = _seed(db, user, ["a"], extra=())
|
|
|
|
r = getattr(as_user(user), method)(f"/api/plex/playlists/{pl.id}/{path}", json={"keys": ["a"]})
|
|
|
|
assert r.status_code == 422
|
|
assert _order(db, pl.id)[0] == ["a"]
|
|
|
|
|
|
def test_remove_then_put_back_is_no_change(db, make_user, as_user):
|
|
user = make_user()
|
|
pl = _seed(db, user, ["a", "b", "c", "d"], extra=())
|
|
client = as_user(user)
|
|
assert (
|
|
client.post(
|
|
f"/api/plex/playlists/{pl.id}/items/remove-bulk", json={"item_rating_keys": ["b", "d"]}
|
|
).status_code
|
|
== 200
|
|
)
|
|
assert _order(db, pl.id)[0] == ["a", "c"]
|
|
_add(client, pl.id, ["b", "d"], [1, 3])
|
|
assert _order(db, pl.id)[0] == ["a", "b", "c", "d"]
|