403 lines
14 KiB
Python
403 lines
14 KiB
Python
from io import BytesIO
|
|
|
|
import pytest
|
|
from PIL import Image
|
|
from pydantic import ValidationError
|
|
|
|
from pigway_frame_flow.extensions.layout import LayoutInput, LayoutPlan, local_plan
|
|
from pigway_frame_flow.extensions.secrets import save_secret
|
|
|
|
|
|
@pytest.mark.parametrize("count", range(1, 13))
|
|
@pytest.mark.parametrize("orientation", ["portrait", "landscape"])
|
|
@pytest.mark.parametrize("style", ["grid", "hero", "poster"])
|
|
def test_local_layout_preserves_all_images_inside_canvas(count, orientation, style):
|
|
ids = [f"{i:032x}" for i in range(count)]
|
|
plan = local_plan(LayoutInput(asset_ids=ids, orientation=orientation, style=style))
|
|
assert [r.asset_id for r in plan.regions] == ids
|
|
assert LayoutPlan.model_validate(plan.model_dump()) == plan
|
|
|
|
|
|
def test_layout_rejects_overlap():
|
|
plan = local_plan(LayoutInput(asset_ids=["a" * 32, "b" * 32])).model_dump()
|
|
plan["regions"][1].update(x=plan["regions"][0]["x"], y=plan["regions"][0]["y"])
|
|
with pytest.raises(ValidationError):
|
|
LayoutPlan.model_validate(plan)
|
|
|
|
|
|
def test_secret_atomic_private_replacement(tmp_path):
|
|
path = tmp_path / "key.secret"
|
|
save_secret(path, "first")
|
|
save_secret(path, "second")
|
|
assert path.read_text() == "second"
|
|
assert path.stat().st_mode & 0o777 == 0o600
|
|
assert list(tmp_path.iterdir()) == [path]
|
|
|
|
|
|
def test_multi_image_import_keeps_aspect_and_never_uploads(client, service):
|
|
image = BytesIO()
|
|
Image.new("RGB", (320, 200), "red").save(image, "PNG")
|
|
response = client.post(
|
|
"/api/v1/layout/images",
|
|
files=[("files", ("photo.png", image.getvalue(), "image/png"))],
|
|
)
|
|
assert response.status_code == 201
|
|
asset_id = response.json()[0]["asset_id"]
|
|
with Image.open(service.assets.path(asset_id)) as imported:
|
|
assert imported.size == (320, 200)
|
|
assert service.registry.get("living-room").client.uploads == 0
|
|
|
|
|
|
def test_ai_requires_explicit_configuration(client):
|
|
image = BytesIO()
|
|
Image.new("RGB", (320, 200), "red").save(image, "PNG")
|
|
result = client.post(
|
|
"/api/v1/layout/images",
|
|
files=[("files", ("photo.png", image.getvalue(), "image/png"))],
|
|
).json()
|
|
response = client.post(
|
|
"/api/v1/layout/plan",
|
|
json={"asset_ids": [result[0]["asset_id"]], "use_ai": True},
|
|
)
|
|
assert response.status_code == 409
|
|
assert response.json()["error"]["code"] == "CONFIG_REQUIRED"
|
|
|
|
|
|
def test_default_retention_keeps_everything(service):
|
|
result = service.local_retention.sweep(force=True)
|
|
assert result["deleted"] == 0
|
|
assert result["disabled"] is True
|
|
|
|
|
|
def test_ai_secret_never_returned_or_stored_in_database(client, service):
|
|
key = "unit-test-private-key"
|
|
response = client.put("/api/v1/settings/ai-layout/key", json={"key": key})
|
|
assert response.status_code == 204
|
|
assert key not in client.get("/api/v1/settings/ai-layout").text
|
|
assert key not in str([dict(r) for r in service.db.all("SELECT * FROM app_config")])
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"kind,reply,location",
|
|
[
|
|
("google-drive", {"files": [{"id": "photo1", "name": "photo.jpg"}]}, "root"),
|
|
(
|
|
"alipan",
|
|
{"items": [{"file_id": "photo1", "name": "photo.jpg", "type": "file"}]},
|
|
"root",
|
|
),
|
|
(
|
|
"baidu",
|
|
{"list": [{"fs_id": 123, "server_filename": "photo.jpg", "isdir": 0}]},
|
|
"/Photos",
|
|
),
|
|
],
|
|
)
|
|
def test_native_drives_metadata_does_not_expose_token(
|
|
service, monkeypatch, kind, reply, location
|
|
):
|
|
from pigway_frame_flow.providers.cloud_drives import CloudDrives
|
|
from pigway_frame_flow.providers.photo_sources import PhotoSourceInput
|
|
|
|
drive = CloudDrives(service.photo_sources)
|
|
monkeypatch.setattr(
|
|
drive,
|
|
"credentials",
|
|
lambda *args: {"access_token": "private-token", "drive_id": "drive1"},
|
|
)
|
|
monkeypatch.setattr(drive, "api", lambda *args, **kwargs: reply)
|
|
result = drive.catalog(
|
|
"source", PhotoSourceInput(name="Cloud", kind=kind, location=location)
|
|
)
|
|
assert len(result) == 1
|
|
assert result[0]["name"] == "photo.jpg"
|
|
assert "private-token" not in str(result)
|
|
|
|
|
|
def test_native_drive_refuses_stalled_pagination(service, monkeypatch):
|
|
from pigway_frame_flow.errors import FlowError
|
|
from pigway_frame_flow.providers.cloud_drives import CloudDrives
|
|
from pigway_frame_flow.providers.photo_sources import PhotoSourceInput
|
|
|
|
drive = CloudDrives(service.photo_sources)
|
|
monkeypatch.setattr(
|
|
drive, "credentials", lambda *args: {"access_token": "private-token"}
|
|
)
|
|
monkeypatch.setattr(
|
|
drive, "api", lambda *args, **kwargs: {"files": [], "nextPageToken": "same"}
|
|
)
|
|
with pytest.raises(FlowError):
|
|
drive.catalog(
|
|
"source",
|
|
PhotoSourceInput(name="Cloud", kind="google-drive", location="root"),
|
|
)
|
|
|
|
|
|
def test_native_credentials_require_authorization(service):
|
|
from pigway_frame_flow.errors import FlowError
|
|
from pigway_frame_flow.providers.cloud_drives import CloudDrives
|
|
from pigway_frame_flow.providers.photo_sources import PhotoSourceInput
|
|
|
|
with pytest.raises(FlowError) as exc:
|
|
CloudDrives(service.photo_sources).credentials(
|
|
"source", PhotoSourceInput(name="Cloud", kind="baidu", location="/")
|
|
)
|
|
assert "CONFIG_REQUIRED" in str(exc.value.code)
|
|
|
|
|
|
def test_client_token_hash_and_revocation(client, service):
|
|
response = client.post(
|
|
"/api/v1/clients",
|
|
json={
|
|
"client_id": "calendar",
|
|
"name": "Calendar",
|
|
"permissions": ["devices:read"],
|
|
},
|
|
)
|
|
assert response.status_code == 201
|
|
token = response.json()["token"]
|
|
assert token not in client.get("/api/v1/clients").text
|
|
assert token not in str([dict(r) for r in service.db.all("SELECT * FROM clients")])
|
|
assert (
|
|
client.get(
|
|
"/api/v1/devices", headers={"Authorization": "Bearer " + token}
|
|
).status_code
|
|
== 200
|
|
)
|
|
assert (
|
|
client.post(
|
|
"/api/v1/display", headers={"Authorization": "Bearer " + token}, json={}
|
|
).status_code
|
|
== 403
|
|
)
|
|
assert client.delete("/api/v1/clients/calendar").status_code == 204
|
|
assert (
|
|
client.get(
|
|
"/api/v1/devices", headers={"Authorization": "Bearer " + token}
|
|
).status_code
|
|
== 401
|
|
)
|
|
|
|
|
|
def test_invalid_member_prevents_partial_multi_import(client, service):
|
|
image = BytesIO()
|
|
Image.new("RGB", (100, 100)).save(image, "PNG")
|
|
before = len(service.db.all("SELECT * FROM assets"))
|
|
result = client.post(
|
|
"/api/v1/layout/images",
|
|
files=[
|
|
("files", ("good.png", image.getvalue(), "image/png")),
|
|
("files", ("bad.jpg", b"bad", "image/jpeg")),
|
|
],
|
|
)
|
|
assert result.status_code == 400
|
|
assert len(service.db.all("SELECT * FROM assets")) == before
|
|
|
|
|
|
def test_widget_mutation_is_queued_and_idempotent(client, service):
|
|
catalog = service.registry.official["living-room"].widgets().catalog
|
|
cover_id = catalog[0].id
|
|
headers = {"Idempotency-Key": "widget-test"}
|
|
first = client.post(
|
|
"/api/v1/devices/living-room/widgets/selection",
|
|
json={"cover_id": cover_id},
|
|
headers=headers,
|
|
)
|
|
assert first.status_code == 202
|
|
assert (
|
|
client.post(
|
|
"/api/v1/devices/living-room/widgets/selection",
|
|
json={"cover_id": cover_id},
|
|
headers=headers,
|
|
).json()
|
|
== first.json()
|
|
)
|
|
service.process_one()
|
|
assert service.operation(first.json()["operation_id"])["status"] == "COMPLETED"
|
|
assert service.registry.get("living-room").client.uploads == 0
|
|
|
|
|
|
def test_widget_post_write_read_failure_blocks_further_writes(
|
|
client, service, monkeypatch
|
|
):
|
|
connection = service.registry.official["living-room"]
|
|
cover_id = connection.widgets().catalog[0].id
|
|
response = client.post(
|
|
"/api/v1/devices/living-room/widgets/selection", json={"cover_id": cover_id}
|
|
)
|
|
original = connection.widgets
|
|
calls = 0
|
|
|
|
def fail_confirmation():
|
|
nonlocal calls
|
|
calls += 1
|
|
if calls > 1:
|
|
raise RuntimeError("network unavailable")
|
|
return original()
|
|
|
|
monkeypatch.setattr(connection, "widgets", fail_confirmation)
|
|
service.process_one()
|
|
assert service.operation(response.json()["operation_id"])["status"] == "UNKNOWN"
|
|
assert service.blocked("living-room")
|
|
|
|
|
|
def test_webhook_signs_exact_body_and_records_delivery(client, service, monkeypatch):
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
|
|
service.settings.image_url_hosts.append("example.com")
|
|
response = client.post(
|
|
"/api/v1/webhooks",
|
|
json={
|
|
"callback_url": "https://example.com/events",
|
|
"event_types": ["display.changed"],
|
|
},
|
|
)
|
|
assert response.status_code == 201
|
|
key = response.json()["signing_secret"]
|
|
service.db.event("display.changed", {"device_id": "living-room"})
|
|
sent = []
|
|
monkeypatch.setattr(
|
|
service.assets, "fetch_url", lambda url, **kwargs: sent.append(kwargs) or b""
|
|
)
|
|
service.webhooks.tick()
|
|
assert len(sent) == 1
|
|
message = sent[0]
|
|
assert (
|
|
message["headers"]["X-Frame-Flow-Signature"]
|
|
== "sha256="
|
|
+ hmac.new(key.encode(), message["body"], hashlib.sha256).hexdigest()
|
|
)
|
|
assert json.loads(message["body"])["type"] == "display.changed"
|
|
assert key not in client.get("/api/v1/webhooks").text
|
|
deliveries = client.get(
|
|
"/api/v1/webhooks/" + response.json()["id"] + "/deliveries"
|
|
).json()
|
|
assert deliveries[0]["status"] == "DELIVERED"
|
|
|
|
|
|
def test_webhook_requires_explicit_host(client):
|
|
response = client.post(
|
|
"/api/v1/webhooks",
|
|
json={
|
|
"callback_url": "http://127.0.0.1/events",
|
|
"event_types": ["display.changed"],
|
|
},
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
|
|
def test_external_provider_requires_explicit_lan_approval(client):
|
|
payload = {
|
|
"kind": "http-json",
|
|
"name": "NAS",
|
|
"endpoint": "http://192.0.2.10/state",
|
|
}
|
|
assert client.put("/api/v1/providers/external/nas", json=payload).status_code == 403
|
|
payload["allow_insecure_lan"] = True
|
|
assert client.put("/api/v1/providers/external/nas", json=payload).status_code == 200
|
|
|
|
|
|
def test_google_calendar_requires_token(client):
|
|
assert (
|
|
client.put(
|
|
"/api/v1/providers/external/calendar",
|
|
json={"kind": "google-calendar", "name": "Calendar", "endpoint": "primary"},
|
|
).status_code
|
|
== 200
|
|
)
|
|
response = client.get("/api/v1/providers/external/calendar/data")
|
|
assert response.status_code == 409
|
|
assert response.json()["error"]["code"] == "CONFIG_REQUIRED"
|
|
|
|
|
|
def test_photo_quality_dark_and_duplicate(service):
|
|
from pigway_frame_flow.extensions.quality import passes, photo_quality
|
|
from pigway_frame_flow.providers.photo_sources import PhotoSourceInput
|
|
|
|
image = BytesIO()
|
|
Image.new("RGB", (100, 100), "black").save(image, "PNG")
|
|
result = photo_quality(image.getvalue())
|
|
assert result["brightness"] == 0
|
|
assert result == photo_quality(image.getvalue())
|
|
source = PhotoSourceInput(
|
|
name="NAS", kind="local", location=".", minimum_brightness=10
|
|
)
|
|
assert not passes(result, source)
|
|
|
|
|
|
def test_layout_missing_plan_is_rejected_before_queue(client):
|
|
response = client.post(
|
|
"/api/v1/render", json={"type": "template", "template": "collage", "data": {}}
|
|
)
|
|
assert response.status_code == 400
|
|
assert response.json()["error"]["code"] == "INVALID_REQUEST"
|
|
|
|
|
|
def test_ai_layout_sends_metadata_only_and_validates_response(
|
|
client, service, monkeypatch
|
|
):
|
|
import json
|
|
|
|
image = BytesIO()
|
|
Image.new("RGB", (100, 200), "red").save(image, "PNG")
|
|
asset = client.post(
|
|
"/api/v1/layout/images",
|
|
files=[("files", ("photo.png", image.getvalue(), "image/png"))],
|
|
).json()[0]
|
|
client.put(
|
|
"/api/v1/settings/ai-layout",
|
|
json={
|
|
"endpoint": "https://example.com/v1/chat/completions",
|
|
"model": "test-model",
|
|
},
|
|
)
|
|
client.put("/api/v1/settings/ai-layout/key", json={"key": "test-ai-secret"})
|
|
plan = local_plan(LayoutInput(asset_ids=[asset["asset_id"]]))
|
|
calls = []
|
|
|
|
def respond(url, **kwargs):
|
|
calls.append(kwargs)
|
|
return json.dumps(
|
|
{"choices": [{"message": {"content": plan.model_dump_json()}}]}
|
|
).encode()
|
|
|
|
monkeypatch.setattr(service.assets, "fetch_url", respond)
|
|
response = client.post(
|
|
"/api/v1/layout/plan", json={"asset_ids": [asset["asset_id"]], "use_ai": True}
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["engine"] == "ai"
|
|
assert b"base64" not in calls[0]["body"]
|
|
assert b"test-ai-secret" not in calls[0]["body"]
|
|
assert image.getvalue() not in calls[0]["body"]
|
|
assert calls[0]["headers"]["Authorization"] == "Bearer test-ai-secret"
|
|
|
|
|
|
def test_webhook_matching_event_after_unrelated_backlog(client, service, monkeypatch):
|
|
import json
|
|
|
|
service.settings.image_url_hosts.append("example.com")
|
|
response = client.post(
|
|
"/api/v1/webhooks",
|
|
json={
|
|
"callback_url": "https://example.com/events",
|
|
"event_types": ["operation.failed"],
|
|
},
|
|
)
|
|
assert response.status_code == 201
|
|
for _ in range(500):
|
|
service.db.event("display.changed", {})
|
|
service.db.event("operation.failed", {"operation_id": "expected"})
|
|
sent = []
|
|
monkeypatch.setattr(
|
|
service.assets, "fetch_url", lambda url, **kwargs: sent.append(kwargs) or b""
|
|
)
|
|
service.webhooks.tick()
|
|
service.webhooks.last = 0
|
|
service.webhooks.tick()
|
|
assert len(sent) == 1
|
|
assert json.loads(sent[0]["body"])["data"] == {"operation_id": "expected"}
|
|
assert len(service.db.all("SELECT * FROM webhook_deliveries")) == 1
|