pigway-console-adapter-frame-flow/tests/test_adapter.py

import importlib.util
import json
import subprocess
import sys
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from unittest.mock import patch

import pytest

ROOT = Path(__file__).resolve().parents[1]
spec = importlib.util.spec_from_file_location(
    "frame_adapter", ROOT / "plugin/worker.py"
)
worker = importlib.util.module_from_spec(spec)
spec.loader.exec_module(worker)


@pytest.fixture
def endpoint():
    calls = []
    data = {
        "/health": {"status": "ok"},
        "/ready": {"status": "ready", "checks": {"sqlite": True}},
        "/api/v1/devices": [
            {
                "id": "frame",
                "name": "Frame",
                "status": "offline",
                "power": {"battery_percent": 45},
            }
        ],
        "/api/v1/jobs": [{"id": "morning", "enabled": True, "last_result": "FAILED"}],
    }

    class Handler(BaseHTTPRequestHandler):
        def do_GET(self):
            calls.append((self.path, self.headers.get("Authorization")))
            value = data[self.path]
            self.send_response(
                503 if self.path == "/ready" and value["status"] == "not_ready" else 200
            )
            self.end_headers()
            self.wfile.write(json.dumps(value).encode())

        def log_message(self, *args):
            pass

    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
    thread = threading.Thread(target=server.serve_forever)
    thread.start()
    device = {
        "id": "test",
        "connection": {"url": f"http://127.0.0.1:{server.server_port}"},
        "credentials": {"token": "private-test-token"},
    }
    yield device, data, calls
    server.shutdown()
    server.server_close()
    thread.join()


def test_collect_and_not_ready(endpoint):
    device, data, calls = endpoint
    snapshot = worker.collect(device)
    assert {a["id"] for a in snapshot["alerts"]} == {
        "DEVICE_OFFLINE:frame",
        "JOB_FAILED:morning",
    }
    assert snapshot["alerts_complete"] is True
    assert len(snapshot["resources"]) == 3
    assert all(
        auth == "Bearer private-test-token"
        for path, auth in calls
        if path.startswith("/api/")
    )
    data["/ready"]["status"] = "not_ready"
    assert "SERVICE_NOT_READY" in {a["id"] for a in worker.collect(device)["alerts"]}


def test_no_result_is_not_success(endpoint):
    device, data, _ = endpoint
    data["/api/v1/jobs"][0]["last_result"] = None
    assert not any(
        m["key"] == "last_run_success" for m in worker.collect(device)["metrics"]
    )
    data["/api/v1/devices"] = {}
    with pytest.raises(worker.AdapterError, match="INVALID_RESPONSE"):
        worker.collect(device)


def test_rpc_and_write_rejected(endpoint):
    device, _, _ = endpoint
    messages = [
        {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "adapter.initialize",
            "params": {"protocol_version": 1},
        },
        {
            "jsonrpc": "2.0",
            "id": 2,
            "method": "device.collect",
            "params": {"device": device},
        },
        {
            "jsonrpc": "2.0",
            "id": 3,
            "method": "device.write",
            "params": {"device": device},
        },
    ]
    result = subprocess.run(
        [sys.executable, str(ROOT / "plugin/worker.py")],
        input="".join(json.dumps(m) + "\n" for m in messages),
        text=True,
        capture_output=True,
        timeout=15,
        check=True,
    )
    replies = {r["id"]: r for r in map(json.loads, result.stdout.splitlines())}
    assert replies[1]["result"]["id"] == "pigway-frame-flow"
    assert replies[2]["result"]["schema_version"] == 1
    assert replies[3]["error"]["data"] == "UNSUPPORTED"
    assert "private-test-token" not in result.stdout + result.stderr


def test_auth_and_redirect_are_not_followed(endpoint):
    from urllib.error import HTTPError

    device, _, _ = endpoint
    for status, code in ((401, "AUTH_FAILED"), (302, "DEVICE_API_ERROR")):
        with patch.object(worker, "build_opener") as opener:
            opener.return_value.open.side_effect = HTTPError(
                "url", status, "private", {}, None
            )
            with pytest.raises(worker.AdapterError, match=code):
                worker.request(device, "/api/v1/jobs")


def test_current_frame_flow_api_contract(tmp_path):
    pytest.importorskip("pigway_frame_flow")
    from fastapi.testclient import TestClient
    from pigway_frame_flow.api.app import create_app
    from pigway_frame_flow.settings import Settings

    settings = Settings(
        data_dir=tmp_path / "data",
        config_dir=tmp_path / "config",
        api_token="contract-token",
        mock=True,
    )
    app = create_app(settings, start_worker=False)
    with TestClient(app) as client:
        client.headers["Authorization"] = "Bearer contract-token"

        def request(device, path):
            response = client.get(path)
            assert response.status_code == 200
            return response.json()

        with patch.object(worker, "request", side_effect=request):
            snapshot = worker.collect({})
        assert snapshot["alerts_complete"] is True
        assert any(
            m["key"] == "service_ready" and m["value"] == 1 for m in snapshot["metrics"]
        )
        assert any(r["type"] == "device" for r in snapshot["resources"])


def test_metric_and_alert_labels_follow_console_language(endpoint):
    device, _, _ = endpoint
    snapshot = worker.collect(device)
    assert all(set(metric["label"]) == {"zh", "en"} for metric in snapshot["metrics"])
    assert all(set(alert["label"]) == {"zh", "en"} for alert in snapshot["alerts"])
    frame = next(r for r in snapshot["resources"] if r["type"] == "device")
    assert frame["id"] == "device:frame"
    assert frame["name"] == "Frame"