481 lines
16 KiB
Python
481 lines
16 KiB
Python
import hashlib
|
|
import json
|
|
from io import BytesIO
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from PIL import Image
|
|
|
|
from pigway_frame_flow.api.app import create_app
|
|
from pigway_frame_flow.settings import Settings
|
|
from pigway_frame_flow.storage.database import encode, now_iso
|
|
|
|
|
|
def test_health_ready(client):
|
|
assert client.get("/health").status_code == 200
|
|
assert client.get("/ready").status_code == 200
|
|
assert all(client.get("/ready").json()["checks"].values())
|
|
|
|
|
|
def test_without_external_project(client):
|
|
assert client.get("/").status_code == 200
|
|
assert client.get("/api/v1/devices").status_code == 200
|
|
assert len(client.get("/api/v1/templates").json()) == 9
|
|
assert client.get("/api/v1/jobs").status_code == 200
|
|
|
|
|
|
def test_auth(client):
|
|
assert (
|
|
client.get("/api/v1/devices", headers={"Authorization": ""}).status_code == 401
|
|
)
|
|
r = client.get("/api/v1/settings", headers={"Authorization": "Bearer wrong"})
|
|
assert r.status_code == 401
|
|
assert set(r.json()["error"]) == {"code", "message", "request_id", "details"}
|
|
|
|
|
|
def test_scope(client, service):
|
|
service.db.execute(
|
|
"INSERT INTO clients VALUES(?,?,?,?,?,?,?)",
|
|
(
|
|
"reader",
|
|
hashlib.sha256(b"reader").hexdigest(),
|
|
"Reader",
|
|
encode(["devices:read"]),
|
|
now_iso(),
|
|
None,
|
|
1,
|
|
),
|
|
)
|
|
assert (
|
|
client.get(
|
|
"/api/v1/devices", headers={"Authorization": "Bearer reader"}
|
|
).status_code
|
|
== 200
|
|
)
|
|
assert (
|
|
client.post(
|
|
"/api/v1/display",
|
|
headers={"Authorization": "Bearer reader"},
|
|
json={"content": {"type": "template", "template": "quote"}},
|
|
).status_code
|
|
== 403
|
|
)
|
|
|
|
|
|
def test_device_schema(client):
|
|
device = client.get("/api/v1/devices/living-room").json()
|
|
assert device["id"] == "living-room"
|
|
assert device["display"]["width"] == 1200
|
|
assert device["capabilities"]["color"] is True
|
|
assert device["slideshow"]["mode"] == "queue"
|
|
assert device["slideshow"]["members_available"] is False
|
|
assert "nickname" not in device
|
|
assert "api_key" not in json.dumps(device)
|
|
|
|
|
|
def test_all_cloud_capabilities(client, service):
|
|
response = client.post(
|
|
"/api/v1/display", json={"content": {"type": "template", "template": "quote"}}
|
|
)
|
|
assert response.status_code == 202
|
|
id = response.json()["operation_id"]
|
|
service.process_one()
|
|
service.sync()
|
|
assert client.get("/api/v1/operations/" + id).json()["status"] == "COMPLETED"
|
|
images = client.get("/api/v1/devices/living-room/images").json()
|
|
assert len(images) == 2
|
|
assert client.get(images[0]["url"] + "?thumbnail=true").status_code == 200
|
|
name = images[1]["name"]
|
|
named = client.post(
|
|
"/api/v1/display", json={"content": {"type": "image", "image_name": name}}
|
|
)
|
|
assert named.status_code == 202
|
|
service.process_one()
|
|
service.sync()
|
|
assert service.registry.get("living-room").get_current_image() == 1
|
|
|
|
|
|
def test_delete_current_confirmation(client):
|
|
assert client.delete("/api/v1/devices/living-room/images/1").status_code == 409
|
|
|
|
|
|
def test_preview_local_upload(client, service):
|
|
out = BytesIO()
|
|
Image.new("RGB", (100, 200), "blue").save(out, "PNG")
|
|
preview = client.post(
|
|
"/api/v1/preview",
|
|
files={"file": ("sample.png", out.getvalue(), "image/png")},
|
|
data={"fit": "contain", "profile": "grayscale"},
|
|
)
|
|
assert preview.status_code == 200
|
|
asset = preview.json()
|
|
assert client.get(asset["url"]).status_code == 200
|
|
r = client.post(
|
|
"/api/v1/devices/living-room/upload",
|
|
json={"type": "image", "asset_id": asset["asset_id"]},
|
|
)
|
|
assert r.status_code == 202
|
|
service.process_one()
|
|
assert service.registry.get("living-room").get_current_image() == 1
|
|
assert service.registry.get("living-room").client.uploads == 0
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"content",
|
|
[{"type": "html", "data": "hello"}, {"type": "markdown", "data": "hello"}],
|
|
)
|
|
def test_unsupported(client, content):
|
|
r = client.post("/api/v1/display", json={"content": content})
|
|
assert r.status_code == 422
|
|
assert r.json()["error"]["code"] == "UNSUPPORTED_CAPABILITY"
|
|
|
|
|
|
def test_idempotency_api(client, service):
|
|
p = {"content": {"type": "template", "template": "quote"}}
|
|
a = client.post("/api/v1/display", json=p, headers={"Idempotency-Key": "repeat"})
|
|
b = client.post("/api/v1/display", json=p, headers={"Idempotency-Key": "repeat"})
|
|
assert a.json() == b.json()
|
|
service.process_one()
|
|
assert service.registry.get("living-room").client.uploads == 1
|
|
|
|
|
|
def test_request_id(client):
|
|
r = client.get(
|
|
"/api/v1/devices/no-device", headers={"X-Request-ID": "external-123"}
|
|
)
|
|
assert r.status_code == 404
|
|
assert r.headers["X-Request-ID"] == "external-123"
|
|
assert r.json()["error"]["request_id"] == "external-123"
|
|
|
|
|
|
def test_secret_never_returned(client, service):
|
|
assert "test-frame-token" not in client.get("/api/v1/settings").text
|
|
data = client.post(
|
|
"/api/v1/display",
|
|
json={
|
|
"content": {
|
|
"type": "template",
|
|
"template": "quote",
|
|
"data": {"title": "test-frame-token"},
|
|
}
|
|
},
|
|
)
|
|
row = service.db.one(
|
|
"SELECT payload FROM operations WHERE id=?", (data.json()["operation_id"],)
|
|
)
|
|
assert "test-frame-token" not in row["payload"]
|
|
assert "test-frame-token" not in str(service.settings)
|
|
|
|
|
|
def test_cors_default(client):
|
|
response = client.options(
|
|
"/api/v1/display",
|
|
headers={
|
|
"Origin": "https://external.example",
|
|
"Access-Control-Request-Method": "POST",
|
|
},
|
|
)
|
|
assert "access-control-allow-origin" not in response.headers
|
|
|
|
|
|
def test_missing_token_not_ready(tmp_path):
|
|
app = create_app(
|
|
Settings(
|
|
data_dir=tmp_path / "data", config_dir=tmp_path / "config", api_token=""
|
|
),
|
|
False,
|
|
)
|
|
with TestClient(app) as c:
|
|
assert c.get("/ready").status_code == 503
|
|
assert (
|
|
c.post(
|
|
"/api/v1/display",
|
|
json={"content": {"type": "template", "template": "quote"}},
|
|
).status_code
|
|
== 401
|
|
)
|
|
|
|
|
|
def test_live_read_only(tmp_path):
|
|
app = create_app(
|
|
Settings(
|
|
data_dir=tmp_path / "data",
|
|
config_dir=tmp_path / "config",
|
|
api_token="test",
|
|
api_key="fake-device-key",
|
|
mock=False,
|
|
write_enabled=False,
|
|
),
|
|
False,
|
|
)
|
|
with TestClient(app) as c:
|
|
c.headers["Authorization"] = "Bearer test"
|
|
assert c.get("/ready").status_code == 200
|
|
assert (
|
|
c.post(
|
|
"/api/v1/display",
|
|
json={"content": {"type": "template", "template": "quote"}},
|
|
).status_code
|
|
== 403
|
|
)
|
|
assert c.delete("/api/v1/devices/living-room/images/1").status_code == 403
|
|
assert c.post("/api/v1/devices/living-room/restore").status_code == 403
|
|
|
|
|
|
def test_multidevice(tmp_path):
|
|
config = tmp_path / "config"
|
|
config.mkdir()
|
|
(config / "devices.yml").write_text(
|
|
"devices:\n living-room:\n driver: kokonna\n office:\n driver: kokonna\n"
|
|
)
|
|
app = create_app(
|
|
Settings(data_dir=tmp_path / "data", config_dir=config, api_token="test"), False
|
|
)
|
|
with TestClient(app) as c:
|
|
c.headers["Authorization"] = "Bearer test"
|
|
assert len(c.get("/api/v1/devices").json()) == 2
|
|
assert (
|
|
c.post(
|
|
"/api/v1/devices/office/display",
|
|
json={"content": {"type": "image", "image_id": 1}},
|
|
).status_code
|
|
== 202
|
|
)
|
|
app.state.service.process_one()
|
|
assert app.state.service.registry.get("living-room").client.displays == []
|
|
assert app.state.service.registry.get("office").client.displays == [1]
|
|
|
|
|
|
def test_job_crud(client):
|
|
p = {
|
|
"name": "Morning",
|
|
"schedule": {"type": "daily", "time": "07:00"},
|
|
"source": {"type": "template", "template": "daily-dashboard"},
|
|
}
|
|
result = client.post("/api/v1/jobs", json=p)
|
|
assert result.status_code == 201
|
|
id = result.json()["id"]
|
|
p["enabled"] = False
|
|
assert client.put("/api/v1/jobs/" + id, json=p).json()["enabled"] is False
|
|
assert client.delete("/api/v1/jobs/" + id).status_code == 204
|
|
|
|
|
|
def test_contract(client, app):
|
|
schema = client.get("/openapi.json").json()
|
|
expected = json.loads((Path(__file__).parent / "contract.json").read_text())
|
|
actual = {
|
|
path: {method: sorted(spec["responses"]) for method, spec in methods.items()}
|
|
for path, methods in schema["paths"].items()
|
|
}
|
|
assert actual == expected
|
|
assert schema["components"]["securitySchemes"]["HTTPBearer"]["scheme"] == "bearer"
|
|
assert "Device" in schema["components"]["schemas"]
|
|
assert schema["components"]["schemas"]["OperationStatus"]["enum"] == [
|
|
"QUEUED",
|
|
"RENDERING",
|
|
"UPLOADING",
|
|
"SYNCING",
|
|
"COMPLETED",
|
|
"FAILED",
|
|
"UNKNOWN",
|
|
]
|
|
assert client.get("/docs").status_code == 200
|
|
|
|
|
|
def test_core_has_no_client_special_case():
|
|
for path in Path("pigway_frame_flow").rglob("*.py"):
|
|
assert "ticktar" not in path.read_text().lower()
|
|
assert (
|
|
"KokonnaClient" not in Path("pigway_frame_flow/display/service.py").read_text()
|
|
)
|
|
|
|
|
|
def test_official_brand_assets(client):
|
|
html = client.get("/").text
|
|
assert "/ui/favicon.svg" in html
|
|
assert "/ui/apple-touch-icon.png" in html
|
|
assert 'class="mobile-brand-logo pigway-logo"' in html
|
|
for filename, content_type in [
|
|
("logo-symbol.svg", "image/svg+xml"),
|
|
("favicon.svg", "image/svg+xml"),
|
|
("favicon.ico", "image/vnd.microsoft.icon"),
|
|
("apple-touch-icon.png", "image/png"),
|
|
]:
|
|
response = client.get("/ui/" + filename)
|
|
assert response.status_code == 200
|
|
assert response.headers["content-type"].split(";")[0] in (
|
|
content_type,
|
|
"image/x-icon",
|
|
)
|
|
assert (
|
|
response.content == (Path("pigway_frame_flow/web") / filename).read_bytes()
|
|
)
|
|
|
|
|
|
def test_ui_languages(client):
|
|
assert client.get("/ui/app.js").status_code == 200
|
|
i18n = client.get("/ui/i18n.js").text
|
|
assert "zh:" in i18n and "en:" in i18n
|
|
assert "data-theme=dark" in client.get("/ui/style.css").text
|
|
|
|
|
|
def test_local_asset_deletion(client, service):
|
|
asset = service.assets.put(service.renderer.render("quote", {}), "quote.jpg")
|
|
assert client.delete("/api/v1/assets/" + asset["asset_id"]).status_code == 204
|
|
assert client.get(asset["url"]).status_code == 404
|
|
|
|
|
|
def test_asset_reference_prevents_delete(client, service):
|
|
asset = service.assets.put(service.renderer.render("quote", {}), "photo.jpg")
|
|
service.submit(
|
|
"display",
|
|
{"content": {"type": "image", "asset_id": asset["asset_id"]}},
|
|
"living-room",
|
|
"admin",
|
|
"reference",
|
|
)
|
|
assert client.delete("/api/v1/assets/" + asset["asset_id"]).status_code == 409
|
|
|
|
|
|
def test_unknown_same_idempotency_key_returns_original(client, service):
|
|
from pigway_frame_flow.errors import ErrorCode, FlowError
|
|
|
|
payload = {"content": {"type": "template", "template": "quote"}}
|
|
|
|
def fail(*args):
|
|
raise FlowError(ErrorCode.WRITE_RESULT_UNKNOWN, "Write outcome unknown", 502)
|
|
|
|
service.registry.get("living-room").upload = fail
|
|
original = client.post(
|
|
"/api/v1/display", json=payload, headers={"Idempotency-Key": "timeout-test"}
|
|
)
|
|
service.process_one()
|
|
assert service.operation(original.json()["operation_id"])["status"] == "UNKNOWN"
|
|
retry = client.post(
|
|
"/api/v1/display", json=payload, headers={"Idempotency-Key": "timeout-test"}
|
|
)
|
|
assert retry.status_code == 202
|
|
assert retry.json() == original.json()
|
|
fresh = client.post(
|
|
"/api/v1/display", json=payload, headers={"Idempotency-Key": "new-key"}
|
|
)
|
|
assert fresh.status_code == 202
|
|
assert fresh.json()["operation_id"] != original.json()["operation_id"]
|
|
|
|
|
|
def test_schema_contract(client):
|
|
schemas = client.get("/openapi.json").json()["components"]["schemas"]
|
|
expected = json.loads((Path(__file__).parent / "schema-contract.json").read_text())
|
|
assert {key: schemas[key] for key in expected} == expected
|
|
|
|
|
|
def test_preview_is_draft_until_saved(client, service):
|
|
out = BytesIO()
|
|
Image.new("RGB", (50, 50), "red").save(out, "PNG")
|
|
preview = client.post(
|
|
"/api/v1/preview", files={"file": ("draft.png", out.getvalue(), "image/png")}
|
|
).json()
|
|
assert preview["name"] == "draft.jpg"
|
|
assert not any(
|
|
x["asset_id"] == preview["asset_id"]
|
|
for x in client.get("/api/v1/assets").json()
|
|
)
|
|
saved = client.post(
|
|
"/api/v1/devices/living-room/upload",
|
|
json={"type": "image", "asset_id": preview["asset_id"]},
|
|
)
|
|
assert saved.status_code == 202
|
|
service.process_one()
|
|
assert any(
|
|
x["asset_id"] == preview["asset_id"]
|
|
for x in client.get("/api/v1/assets").json()
|
|
)
|
|
assert service.registry.get("living-room").client.uploads == 0
|
|
|
|
|
|
def test_restore_availability(client, service):
|
|
assert client.get("/api/v1/devices/living-room").json()["can_restore"] is False
|
|
client.post(
|
|
"/api/v1/display", json={"content": {"type": "template", "template": "quote"}}
|
|
)
|
|
service.process_one()
|
|
assert client.get("/api/v1/devices/living-room").json()["can_restore"] is True
|
|
|
|
|
|
def test_template_configuration_reload(client, service):
|
|
assert (
|
|
client.put(
|
|
"/api/v1/templates/quote/config", json={"data": {"quote": "Saved words"}}
|
|
).status_code
|
|
== 200
|
|
)
|
|
assert (
|
|
client.get("/api/v1/templates/quote").json()["defaults"]["quote"]
|
|
== "Saved words"
|
|
)
|
|
|
|
|
|
def test_docs_offline_resources(client):
|
|
html = client.get("/docs").text
|
|
assert "/ui/swagger-ui-bundle.js" in html
|
|
assert "cdn.jsdelivr" not in html
|
|
assert client.get("/ui/swagger-ui-bundle.js").status_code == 200
|
|
assert client.get("/ui/swagger-ui.css").status_code == 200
|
|
|
|
|
|
def test_scoped_client_operation_isolation(client, service):
|
|
token = "restricted-client-fixture"
|
|
service.db.execute(
|
|
"INSERT INTO clients VALUES(?,?,?,?,?,?,?)",
|
|
(
|
|
"reader",
|
|
hashlib.sha256(token.encode()).hexdigest(),
|
|
"Reader",
|
|
encode(["devices:read", "history:read"]),
|
|
now_iso(),
|
|
None,
|
|
1,
|
|
),
|
|
)
|
|
own = service.submit("render", {}, "living-room", "reader", "own-request")
|
|
other = service.submit("render", {}, "living-room", "scheduler", "other-request")
|
|
client.headers["Authorization"] = "Bearer " + token
|
|
assert client.get("/api/v1/operations/" + own).status_code == 200
|
|
assert client.get("/api/v1/operations/" + other).status_code == 403
|
|
assert [op["operation_id"] for op in client.get("/api/v1/operations").json()] == [
|
|
own
|
|
]
|
|
|
|
|
|
def test_settings_exposes_storage_display_name(client, service):
|
|
service.settings.storage_name = "Family NAS"
|
|
result = client.get("/api/v1/settings").json()
|
|
assert result["storage_name"] == "Family NAS"
|
|
destinations = {d["id"]: d for d in result["gallery_destinations"]}
|
|
assert set(destinations) == {"cloud", "local"}
|
|
assert destinations["cloud"]["name"] == "KoKonna"
|
|
assert destinations["local"]["name"] == "Family NAS"
|
|
assert all(d["connected"] and d["supports_upload"] for d in destinations.values())
|
|
assert "test-frame-token" not in str(result)
|
|
|
|
|
|
def test_storage_display_name_environment(monkeypatch):
|
|
monkeypatch.setenv("FRAME_FLOW_STORAGE_NAME", " Family NAS ")
|
|
assert Settings().storage_name == "Family NAS"
|
|
monkeypatch.setenv("FRAME_FLOW_STORAGE_NAME", " ")
|
|
assert Settings().storage_name == "NAS"
|
|
|
|
|
|
def test_settings_device_catalog_does_not_fetch_cloud_status(
|
|
client, service, monkeypatch
|
|
):
|
|
registry = service.registry
|
|
for driver in registry.drivers.values():
|
|
monkeypatch.setattr(
|
|
driver, "device", lambda *a, **kw: pytest.fail("Cloud status fetched")
|
|
)
|
|
result = client.get("/api/v1/settings")
|
|
assert result.status_code == 200
|
|
assert {d["id"] for d in result.json()["device_catalog"]} == set(registry.drivers)
|