Files
pigway-frame-flow/tests/test_content_provenance.py

88 lines
3.1 KiB
Python

from pigway_frame_flow.display.service import DisplayService
from pigway_frame_flow.models import DisplayRequest, TemplateContent
def display(service, content):
request = DisplayRequest(content=content)
operation = service.submit(
"display",
request.model_dump(mode="json"),
"living-room",
"admin",
"source-test",
)
service.process_one()
service.sync()
assert service.operation(operation)["status"] == "COMPLETED"
return operation
def test_vendor_image_has_unknown_template(service):
current = service.current_content("living-room")
assert current["image_id"] == 1
assert current["source"] is None and not current["source_known"]
def test_template_snapshot_survives_changed_defaults_and_restart(service):
display(
service,
{
"type": "template",
"template": "quote",
"data": {"quote": "Original text"},
"profile": "grayscale",
},
)
first = service.current_content("living-room")
assert first["source"]["data"]["quote"] == "Original text"
assert first["source"]["profile"] == "grayscale"
service.db.execute(
"INSERT OR REPLACE INTO template_config VALUES(?,?)",
("quote", '{"quote":"Changed defaults"}'),
)
restarted = DisplayService(
service.db, service.settings, service.registry, service.assets, service.renderer
)
assert restarted.current_content("living-room")["source"] == first["source"]
def test_pre_rendered_asset_retains_template_on_display(service):
asset = service.content_asset(
TemplateContent(
type="template", template="calendar", data={"date": "2026-10-02"}
)
)
display(service, {"type": "image", "asset_id": asset["asset_id"]})
assert service.current_content("living-room")["source"]["template"] == "calendar"
def test_external_switch_does_not_claim_old_template(service):
display(service, {"type": "template", "template": "quote", "data": {}})
service.registry.get("living-room").client.display(1)
service.registry.get("living-room").invalidate()
assert service.current_content("living-room")["source"] is None
def test_restore_returns_original_source(service):
display(
service, {"type": "template", "template": "quote", "data": {"quote": "Art"}}
)
display(service, {"type": "template", "template": "calendar", "data": {}})
operation = service.submit("restore", {}, "living-room", "admin", "restore-source")
service.process_one()
service.sync()
assert service.operation(operation)["status"] == "COMPLETED"
assert service.current_content("living-room")["source"]["template"] == "quote"
def test_asset_source_api(client, service):
asset = service.content_asset(
TemplateContent(type="template", template="quote", data={"quote": "Source"})
)
response = client.get("/api/v1/assets/" + asset["asset_id"] + "/source")
assert (
response.status_code == 200
and response.json()["source"]["data"]["quote"] == "Source"
)
assert client.get("/api/v1/devices/living-room/current-content").status_code == 200