pigway-frame-flow/tests/test_extensions.py

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