75 lines
3.2 KiB
Python
75 lines
3.2 KiB
Python
"""The channel manager's local writes, through HTTP: each one lands and is committed.
|
|
|
|
The service tests (`test_catalog_service.py`) hold what each write does; these hold that the route
|
|
calls it and commits — the `db` fixture fails a test whose handler left a write uncommitted — and
|
|
what a client sees back. None of these reach YouTube. Needs the DB.
|
|
"""
|
|
|
|
from sqlalchemy import select
|
|
|
|
from app.catalog.models import BlockedChannel, Channel, Subscription
|
|
from app.reader.models import ChannelTag, Tag
|
|
|
|
|
|
def _seed(db, user):
|
|
db.add(Channel(id="UCr", title="R"))
|
|
db.flush()
|
|
db.add(Subscription(user_id=user.id, channel_id="UCr"))
|
|
tag = Tag(user_id=user.id, name="Mine")
|
|
db.add(tag)
|
|
db.commit()
|
|
return tag
|
|
|
|
|
|
def test_a_subscriptions_settings_change_and_list_back(db, make_user, as_user):
|
|
user = make_user()
|
|
_seed(db, user)
|
|
client = as_user(user)
|
|
r = client.patch("/api/channels/UCr", json={"priority": 2, "hidden": True})
|
|
assert r.status_code == 200, r.text
|
|
assert (r.json()["priority"], r.json()["hidden"]) == (2, True)
|
|
listed = client.get("/api/channels").json()
|
|
assert [(c["id"], c["priority"], c["hidden"]) for c in listed] == [("UCr", 2, True)]
|
|
r = client.post("/api/sync/deep-all?on=false")
|
|
assert r.json() == {"ok": True, "updated": 1, "deep_requested": False}
|
|
|
|
|
|
def test_a_block_and_an_unblock(db, make_user, as_user):
|
|
user = make_user()
|
|
_seed(db, user)
|
|
client = as_user(user)
|
|
assert client.post("/api/channels/UCr/block").json() == {"ok": True, "blocked": "UCr"}
|
|
assert client.post("/api/channels/UCr/block").status_code == 200 # idempotent
|
|
assert [c["id"] for c in client.get("/api/channels/blocked/list").json()] == ["UCr"]
|
|
assert client.get("/api/channels/UCr").json()["blocked"] is True
|
|
assert client.delete("/api/channels/UCr/block").json() == {"ok": True, "unblocked": "UCr"}
|
|
db.expire_all()
|
|
assert db.execute(select(BlockedChannel)).first() is None
|
|
|
|
|
|
def test_a_tag_attached_to_a_channel_and_detached(db, make_user, as_user):
|
|
user, stranger = make_user(), make_user()
|
|
tag = _seed(db, user)
|
|
client = as_user(user)
|
|
r = client.post("/api/channels/UCr/tags", json={"tag_id": tag.id})
|
|
assert r.status_code == 200, r.text
|
|
assert client.post("/api/channels/UCr/tags", json={"tag_id": tag.id}).status_code == 200
|
|
assert client.get("/api/channels/UCr").json()["tag_ids"] == [tag.id]
|
|
db.add(Subscription(user_id=stranger.id, channel_id="UCr"))
|
|
db.commit()
|
|
# Another user's tag is not theirs to attach: the same 404 as a missing one.
|
|
r = as_user(stranger).post("/api/channels/UCr/tags", json={"tag_id": tag.id})
|
|
assert r.json()["detail"] == {"code": "not_found", "reason": "tag"}
|
|
assert client.delete(f"/api/channels/UCr/tags/{tag.id}").status_code == 200
|
|
db.expire_all()
|
|
assert db.execute(select(ChannelTag)).first() is None
|
|
|
|
|
|
def test_the_sync_status_counts_the_callers_subscriptions(db, make_user, as_user):
|
|
user = make_user()
|
|
_seed(db, user)
|
|
body = as_user(user).get("/api/sync/status").json()
|
|
assert (body["subscriptions"], body["channels_total"]) == (1, 1)
|
|
mine = as_user(user).get("/api/sync/my-status").json()
|
|
assert (mine["channels_total"], mine["channels_recent_pending"]) == (1, 1)
|