peter
5095047c81
The stop-URL contract passed with the two ids swapped and read the backend from the working directory; nothing called a real beacon's onSettled or its library pass-through; the kept-point rule was held by its constants only.
136 lines
4.6 KiB
Python
136 lines
4.6 KiB
Python
"""The library player's resume point (`library_states`).
|
|
|
|
A checkpoint stores where the caller stopped in a library item they can play — their own download,
|
|
or one shared with them — and the two library listings hand it back per item, so the player opens
|
|
there. Too early or too close to the end clears it. Nobody else's checkpoint, and no item the caller
|
|
cannot reach, is ever read or written. DB-backed (real Postgres).
|
|
"""
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from app.models import DownloadJob, DownloadShare, LibraryState
|
|
from app.routes._common import ProgressIn, kept_position
|
|
|
|
|
|
def _done_job(db, user) -> DownloadJob:
|
|
job = DownloadJob(
|
|
user_id=user.id,
|
|
source_kind="url",
|
|
source_ref="https://example.com/a.mp4",
|
|
status="done",
|
|
profile_snapshot={},
|
|
)
|
|
db.add(job)
|
|
db.commit()
|
|
return job
|
|
|
|
|
|
def _checkpoint(client, job_id: int, position: float, duration: float = 3000) -> int:
|
|
"""Send a checkpoint the caller may make; the point the server kept (0 = cleared)."""
|
|
r = client.post(
|
|
f"/api/downloads/{job_id}/progress",
|
|
json={"position_seconds": position, "duration_seconds": duration},
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
return r.json()["position_seconds"]
|
|
|
|
|
|
def _states(db) -> list[tuple[int, int, int]]:
|
|
db.expire_all()
|
|
rows = db.execute(
|
|
select(LibraryState.user_id, LibraryState.job_id, LibraryState.position_seconds)
|
|
).all()
|
|
return [tuple(r) for r in rows]
|
|
|
|
|
|
def test_a_checkpoint_is_listed_back_on_the_item(db, make_user, as_user):
|
|
me = make_user()
|
|
job = _done_job(db, me)
|
|
c = as_user(me)
|
|
|
|
r = c.post(
|
|
f"/api/downloads/{job.id}/progress",
|
|
json={"position_seconds": 754.9, "duration_seconds": 3000},
|
|
)
|
|
assert r.status_code == 200 and r.json() == {
|
|
"ok": True,
|
|
"job_id": job.id,
|
|
"position_seconds": 754,
|
|
}
|
|
|
|
[item] = c.get("/api/downloads").json()
|
|
assert item["position_seconds"] == 754
|
|
# A second checkpoint moves it rather than adding a row.
|
|
assert _checkpoint(c, job.id, 900) == 900
|
|
assert _states(db) == [(me.id, job.id, 900)]
|
|
|
|
|
|
def test_an_item_never_played_lists_zero(db, make_user, as_user):
|
|
me = make_user()
|
|
_done_job(db, me)
|
|
[item] = as_user(me).get("/api/downloads").json()
|
|
assert item["position_seconds"] == 0
|
|
|
|
|
|
def test_too_early_or_near_the_end_clears_it(db, make_user, as_user):
|
|
me = make_user()
|
|
job = _done_job(db, me)
|
|
c = as_user(me)
|
|
|
|
assert _checkpoint(c, job.id, 600) == 600
|
|
assert _checkpoint(c, job.id, 2995) == 0
|
|
assert _states(db) == []
|
|
|
|
assert _checkpoint(c, job.id, 600) == 600
|
|
assert _checkpoint(c, job.id, 3) == 0
|
|
assert _states(db) == []
|
|
# Clearing what is not there is not an error (the beacon and the last checkpoint of one stop).
|
|
assert _checkpoint(c, job.id, 0) == 0
|
|
|
|
|
|
# The edges themselves, one second either side — the same rows as the client's
|
|
# frontend/src/lib/resume.dom.test.ts: the rule must be applied alike on both sides, and the shared
|
|
# constants alone (resume.contract.test.ts) would not see a `>=` that became `>` on one side.
|
|
@pytest.mark.parametrize(
|
|
("position", "duration", "kept"),
|
|
[(4, 3000, 0), (5, 3000, 5), (2990, 3000, 2990), (2991, 3000, 0)],
|
|
)
|
|
def test_the_kept_point_at_its_exact_edges(position, duration, kept):
|
|
body = ProgressIn(position_seconds=position, duration_seconds=duration)
|
|
assert kept_position(body) == kept
|
|
|
|
|
|
def test_a_recipient_keeps_their_own_place_in_a_shared_item(db, make_user, as_user):
|
|
owner, friend = make_user(), make_user()
|
|
job = _done_job(db, owner)
|
|
db.add(DownloadShare(job_id=job.id, from_user_id=owner.id, to_user_id=friend.id))
|
|
db.commit()
|
|
|
|
assert _checkpoint(as_user(owner), job.id, 100) == 100
|
|
assert _checkpoint(as_user(friend), job.id, 2000) == 2000
|
|
|
|
assert as_user(owner).get("/api/downloads").json()[0]["position_seconds"] == 100
|
|
assert as_user(friend).get("/api/downloads/shared").json()[0]["position_seconds"] == 2000
|
|
|
|
|
|
def test_an_item_the_caller_cannot_play_is_a_404_and_writes_nothing(db, make_user, as_user):
|
|
owner, stranger = make_user(), make_user()
|
|
job = _done_job(db, owner)
|
|
|
|
r = as_user(stranger).post(
|
|
f"/api/downloads/{job.id}/progress",
|
|
json={"position_seconds": 600, "duration_seconds": 3000},
|
|
)
|
|
assert r.status_code == 404
|
|
assert _states(db) == []
|
|
|
|
|
|
def test_deleting_the_job_deletes_its_places(db, make_user, as_user):
|
|
me = make_user()
|
|
job = _done_job(db, me)
|
|
assert _checkpoint(as_user(me), job.id, 600) == 600
|
|
db.delete(db.get(DownloadJob, job.id))
|
|
db.commit()
|
|
assert _states(db) == []
|