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"