From 5aa0e9f762ca4c9ede3bc2d77184b3555f63020b Mon Sep 17 00:00:00 2001 From: Pigway Date: Sat, 3 Oct 2026 07:03:35 +0800 Subject: [PATCH] Isolate official account RPC capabilities from API key device driver --- docs/research/KOKONNA_ACCOUNT_CAPABILITIES.md | 10 ++ pigway_frame_flow/api/app.py | 2 +- pigway_frame_flow/core/registry.py | 10 +- pigway_frame_flow/drivers/base.py | 1 - pigway_frame_flow/drivers/kokonna/driver.py | 6 +- pigway_frame_flow/drivers/kokonna/widgets.py | 119 +----------------- pigway_frame_flow/integrations/__init__.py | 1 + .../integrations/kokonna_official/README.md | 23 ++++ .../integrations/kokonna_official/__init__.py | 1 + .../kokonna_official/capabilities.py | 109 ++++++++++++++++ .../integrations/kokonna_official/rpc.py | 82 ++++++++++++ .../kokonna_official}/widget_catalog.json | 0 .../integrations/kokonna_official/widgets.py | 100 +++++++++++++++ pyproject.toml | 2 +- tests/test_official_module.py | 94 ++++++++++++++ 15 files changed, 434 insertions(+), 126 deletions(-) create mode 100644 pigway_frame_flow/integrations/__init__.py create mode 100644 pigway_frame_flow/integrations/kokonna_official/README.md create mode 100644 pigway_frame_flow/integrations/kokonna_official/__init__.py create mode 100644 pigway_frame_flow/integrations/kokonna_official/capabilities.py create mode 100644 pigway_frame_flow/integrations/kokonna_official/rpc.py rename pigway_frame_flow/{drivers/kokonna => integrations/kokonna_official}/widget_catalog.json (100%) create mode 100644 pigway_frame_flow/integrations/kokonna_official/widgets.py create mode 100644 tests/test_official_module.py diff --git a/docs/research/KOKONNA_ACCOUNT_CAPABILITIES.md b/docs/research/KOKONNA_ACCOUNT_CAPABILITIES.md index 72da400..4b75879 100644 --- a/docs/research/KOKONNA_ACCOUNT_CAPABILITIES.md +++ b/docs/research/KOKONNA_ACCOUNT_CAPABILITIES.md @@ -58,3 +58,13 @@ reset, firmware, payment or subscription mutation was executed during research. - Unconfigured access returns `AUTH_REQUIRED` and an explicitly dated snapshot. - Widget selection and the other RPC writes above are not implemented. Account login capability discovery is not permission to exercise those writes. + +## Module extraction + +Official capabilities now live in `pigway_frame_flow/integrations/kokonna_official/`. +The account RPC client, login/session handling, operation inventory and widget +normalization/catalog are isolated there. API Key drivers no longer own a widget +reader. DeviceRegistry composes the optional integration separately per device; +the existing public widget route and old Python import remain compatible. +Only login and the three verified reads are enabled. The module has no generic +public RPC execution endpoint, and unenabled methods cannot reach the network. diff --git a/pigway_frame_flow/api/app.py b/pigway_frame_flow/api/app.py index 5fbf78f..89f19c3 100644 --- a/pigway_frame_flow/api/app.py +++ b/pigway_frame_flow/api/app.py @@ -762,7 +762,7 @@ def create_app(settings=None, start_worker=True): dependencies=[Depends(require("devices:read"))], ) def official_widgets(device_id: str): - return registry.get(device_id).widgets() + return registry.official_widgets(device_id) @app.get( "/api/v1/devices/{device_id}/current-content", diff --git a/pigway_frame_flow/core/registry.py b/pigway_frame_flow/core/registry.py index 5f7a9c8..fcb4dde 100644 --- a/pigway_frame_flow/core/registry.py +++ b/pigway_frame_flow/core/registry.py @@ -4,9 +4,9 @@ from pigway_frame_flow.drivers.base import FrameDeviceDriver from pigway_frame_flow.drivers.kokonna.client import KokonnaClient from pigway_frame_flow.drivers.kokonna.driver import KokonnaDriver from pigway_frame_flow.drivers.kokonna.mock import MockClient -from pigway_frame_flow.drivers.kokonna.widgets import WidgetReader from pigway_frame_flow.errors import ErrorCode as E from pigway_frame_flow.errors import FlowError +from pigway_frame_flow.integrations.kokonna_official.widgets import WidgetReader from pigway_frame_flow.settings import secret @@ -26,6 +26,7 @@ class DeviceRegistry: } ) self.drivers: dict[str, FrameDeviceDriver] = {} + self.official: dict[str, WidgetReader] = {} self.keys = [] for id, entry in devices.items(): if not entry.get("enabled", True): @@ -55,9 +56,14 @@ class DeviceRegistry: widgets = WidgetReader( widget_token, entry.get("widget_robot_id"), settings.mock ) - self.drivers[id] = KokonnaDriver(client, id, settings.sanitize, widgets) + self.official[id] = widgets + self.drivers[id] = KokonnaDriver(client, id, settings.sanitize) def get(self, id: str) -> FrameDeviceDriver: if id not in self.drivers: raise FlowError(E.DEVICE_NOT_FOUND, "Device not found", 404) return self.drivers[id] + + def official_widgets(self, id: str): + self.get(id) + return self.official[id].read() diff --git a/pigway_frame_flow/drivers/base.py b/pigway_frame_flow/drivers/base.py index 30a7659..4c5ccc8 100644 --- a/pigway_frame_flow/drivers/base.py +++ b/pigway_frame_flow/drivers/base.py @@ -4,7 +4,6 @@ from pigway_frame_flow.models import Device, GalleryImage class FrameDeviceDriver(Protocol): - def widgets(self): ... def invalidate(self) -> None: ... def device(self, refresh: bool = False) -> Device: ... def get_current_image(self) -> int | None: ... diff --git a/pigway_frame_flow/drivers/kokonna/driver.py b/pigway_frame_flow/drivers/kokonna/driver.py index bff0c64..1c67482 100644 --- a/pigway_frame_flow/drivers/kokonna/driver.py +++ b/pigway_frame_flow/drivers/kokonna/driver.py @@ -8,9 +8,8 @@ from pigway_frame_flow.models import ( class KokonnaDriver: - def __init__(self, client, device_id, sanitize=lambda v: v, widgets=None): + def __init__(self, client, device_id, sanitize=lambda v: v): self.client, self.device_id, self.sanitize = client, device_id, sanitize - self.widget_reader = widgets self.cached: Device | None = None self.cached_at = 0.0 @@ -63,9 +62,6 @@ class KokonnaDriver: self.cached_at = time.monotonic() return result - def widgets(self): - return self.widget_reader.read() - def get_current_image(self): return self.device().current_content.id diff --git a/pigway_frame_flow/drivers/kokonna/widgets.py b/pigway_frame_flow/drivers/kokonna/widgets.py index 96a1bd1..0a5a4d6 100644 --- a/pigway_frame_flow/drivers/kokonna/widgets.py +++ b/pigway_frame_flow/drivers/kokonna/widgets.py @@ -1,118 +1,5 @@ -"""Read-only official widget RPC; separate from the device OpenAPI credential.""" +"""Compatibility import; official capabilities live outside device drivers.""" -import json -import time -from pathlib import Path +from pigway_frame_flow.integrations.kokonna_official.widgets import WidgetReader -import requests - -from pigway_frame_flow.errors import ErrorCode as E -from pigway_frame_flow.errors import FlowError -from pigway_frame_flow.models import Widget, WidgetState - - -class WidgetReader: - def __init__(self, token="", robot_id=None, mock=False): - self._token = token - self._robot_id = robot_id - self.mock = mock - self._cached = None - self._cached_at = 0.0 - - def read(self) -> WidgetState: - if self._cached is not None and time.monotonic() - self._cached_at < 60: - return self._cached - snapshot = json.loads( - Path(__file__).with_name("widget_catalog.json").read_text() - ) - catalog = [ - Widget(**{k: row[k] for k in ("id", "name", "tags", "arg_keys")}) - for row in snapshot["catalog"] - ] - if self.mock: - return WidgetState( - source="mock", - configured=True, - catalog=catalog, - current=catalog[0], - current_verified=True, - ) - if not self._token or self._robot_id is None: - return WidgetState( - source="snapshot", - configured=False, - catalog=catalog, - observed_at=snapshot["observed_at"], - status="AUTH_REQUIRED", - ) - try: - response = requests.post( - "https://robot.galaxyguide.cn/api/rpc", - headers={"Authorization": f"Bearer {self._token}"}, - json={ - "jsonrpc": "2.0", - "method": "appFrame.getFrameCovers", - "params": {"robotId": self._robot_id}, - "id": int(time.time() * 1000), - }, - timeout=15, - allow_redirects=False, - ) - if response.status_code in (401, 403): - raise FlowError(E.UNAUTHORIZED, "Official widget session expired", 401) - response.raise_for_status() - payload = response.json() - if payload.get("error"): - raise FlowError( - E.CLOUD_UNAVAILABLE, "Official widget query rejected", 502 - ) - data = payload["result"] - rows = data["rows"] - catalog = [ - Widget( - id=row["id"], - name=row["name"], - tags=row.get("tags") or [], - arg_keys=list((row.get("args") or {}).keys()) - if isinstance(row.get("args"), dict) - else [], - ) - for row in rows - ] - selected = data.get("frameCoverRobot") or {} - cover = selected.get("frameCover") or {} - current_id = cover.get("id", selected.get("coverId")) - current = next((row for row in catalog if row.id == current_id), None) - robot = data.get("robot") or {} - no_widget = ( - "frameCoverRobot" in data - and data["frameCoverRobot"] is None - and "frameCoverRobotId" in robot - and robot["frameCoverRobotId"] is None - ) - result = WidgetState( - source="live", - configured=True, - catalog=catalog, - current=current, - current_verified=current is not None or no_widget, - status="OK" - if current - else "NO_WIDGET" - if no_widget - else "CURRENT_UNVERIFIED", - ) - except FlowError: - raise - except ( - requests.RequestException, - ValueError, - KeyError, - TypeError, - AttributeError, - ): - raise FlowError( - E.CLOUD_UNAVAILABLE, "Official widget query unavailable", 502 - ) from None - self._cached, self._cached_at = result, time.monotonic() - return result +__all__ = ["WidgetReader"] diff --git a/pigway_frame_flow/integrations/__init__.py b/pigway_frame_flow/integrations/__init__.py new file mode 100644 index 0000000..46dbfa3 --- /dev/null +++ b/pigway_frame_flow/integrations/__init__.py @@ -0,0 +1 @@ +"""Optional integrations, independent of device drivers.""" diff --git a/pigway_frame_flow/integrations/kokonna_official/README.md b/pigway_frame_flow/integrations/kokonna_official/README.md new file mode 100644 index 0000000..23f31ba --- /dev/null +++ b/pigway_frame_flow/integrations/kokonna_official/README.md @@ -0,0 +1,23 @@ +# 官方能力模块 / Official capabilities + +Optional official-account integration, independent of the KoKonna device API Key +client. The application composition root attaches it per configured device; the +FrameDeviceDriver protocol does not depend on account login or widgets. + +- `rpc.py`: in-memory official login/session, disconnect, device discovery/detail + and widget query. Strict method allowlist, no retries, no raw vendor errors. +- `widgets.py`: normalized widget states and 60-second successful query cache. +- `capabilities.py`: full discovered client operation inventory and evidence. + Discovered does not mean implemented or enabled. Unenabled operations are blocked. +- `widget_catalog.json`: dated, non-secret fallback catalog. + +The public `/api/v1/devices/{device_id}/widgets` endpoint is unchanged. The old +`drivers.kokonna.widgets.WidgetReader` import remains a compatibility re-export. +No raw RPC proxy or public password-login endpoint is introduced. Future admin +account connection UI must use this module and keep credentials server-side. +No account or session secret is automatically persisted by this module. + +设备 API Key 仍负责现有设备、图片和图库接口。官方账号功能集中在本目录， +不进入 DisplayService、Renderer、Scheduler 或 FrameDeviceDriver。 +账号登录已提供内部客户端能力；页面连接入口、写操作和会话持久化尚未实现。 +所有已发现但未启用的 RPC 方法均拒绝执行，不复用 API Key 的写入开关。 diff --git a/pigway_frame_flow/integrations/kokonna_official/__init__.py b/pigway_frame_flow/integrations/kokonna_official/__init__.py new file mode 100644 index 0000000..64a1711 --- /dev/null +++ b/pigway_frame_flow/integrations/kokonna_official/__init__.py @@ -0,0 +1 @@ +"""Optional KoKonna official-account capabilities; no device API Key dependency.""" diff --git a/pigway_frame_flow/integrations/kokonna_official/capabilities.py b/pigway_frame_flow/integrations/kokonna_official/capabilities.py new file mode 100644 index 0000000..db877cd --- /dev/null +++ b/pigway_frame_flow/integrations/kokonna_official/capabilities.py @@ -0,0 +1,109 @@ +"""Discovered RPC operations, not an authorization or live support guarantee.""" + +from dataclasses import dataclass + + +@dataclass(frozen=True) +class OfficialCapability: + method: str + implemented: bool + effect: str + evidence: str + + +READ_OPERATIONS = frozenset( + {"app.getRobots", "appFrame.getFrame", "appFrame.getFrameCovers"} +) +DISCOVERED_METHODS = ( + "app.addFriendRobot", + "app.addShareRobot", + "app.bindRobot", + "app.changePassword", + "app.cleanMemory", + "app.createCustomRole", + "app.createPointOrder", + "app.deleteFriendRobot", + "app.forgetPassword", + "app.getChatHistory", + "app.getCustomRoleInfo", + "app.getDistributors", + "app.getFirmwares", + "app.getPayOrder", + "app.getPayOrderList", + "app.getPoints", + "app.getRedeem", + "app.getRobot", + "app.getRobotByUUID", + "app.getRobots", + "app.getUserInfo", + "app.login", + "app.redeemRobotPointByCode", + "app.register", + "app.removeShare", + "app.sendChat", + "app.sendCtrl", + "app.setRobotNetwork", + "app.setTimezone", + "app.setUpgradeFirmware", + "app.testCustomRoleVoice", + "app.unlink", + "app.updateCustomRole", + "app.updateRobot", + "app.wxlogin", + "appFrame.addFrameArtist", + "appFrame.addFrameGalleryImage", + "appFrame.addTag", + "appFrame.changeFrameImageName", + "appFrame.changeName", + "appFrame.chat", + "appFrame.deleteAllImage", + "appFrame.deleteImage", + "appFrame.displayImage", + "appFrame.enablePair", + "appFrame.enableShareCode", + "appFrame.getFrame", + "appFrame.getFrameArtists", + "appFrame.getFrameCovers", + "appFrame.getFrameGalleryImages", + "appFrame.getFrameGallerys", + "appFrame.getFrameSplashs", + "appFrame.getImages", + "appFrame.getShareImages", + "appFrame.importImagesByTransferToken", + "appFrame.likeImage", + "appFrame.recognizeVoice", + "appFrame.refreshImage", + "appFrame.refreshShareUUID", + "appFrame.removeTag", + "appFrame.reportFrameImage", + "appFrame.reset", + "appFrame.setFrameCover", + "appFrame.setFrameImageChangeInterval", + "appFrame.setFrameImageChangeMinute", + "appFrame.setFrameImageChangeTags", + "appFrame.setFrameImageChangeType", + "appFrame.setImageTags", + "appFrame.setRotate", + "appFrame.setSplash", + "appFrame.shareImage", + "appFrame.text2speech", + "appFrame.uploadImage", +) + + +def capabilities() -> tuple[OfficialCapability, ...]: + return tuple( + OfficialCapability( + method=method, + implemented=method in READ_OPERATIONS or method == "app.login", + effect="read" + if method in READ_OPERATIONS + else "authentication" + if method == "app.login" + else "unverified", + evidence="live_verified" + if method in READ_OPERATIONS or method == "app.login" + else "apk_discovered", + ) + for method in DISCOVERED_METHODS + ) diff --git a/pigway_frame_flow/integrations/kokonna_official/rpc.py b/pigway_frame_flow/integrations/kokonna_official/rpc.py new file mode 100644 index 0000000..fb008f9 --- /dev/null +++ b/pigway_frame_flow/integrations/kokonna_official/rpc.py @@ -0,0 +1,82 @@ +"""Strict account RPC allowlist. Never falls back to a device key or retries.""" + +import time + +import requests + +from pigway_frame_flow.errors import ErrorCode as E +from pigway_frame_flow.errors import FlowError +from pigway_frame_flow.integrations.kokonna_official.capabilities import READ_OPERATIONS + + +class OfficialRPC: + def __init__(self, token: str = ""): + self._token = token + + @property + def authorized(self) -> bool: + return bool(self._token) + + def _call(self, method: str, params: dict) -> dict: + if method not in READ_OPERATIONS and method != "app.login": + raise FlowError( + E.UNSUPPORTED_CAPABILITY, "Official operation is not enabled", 422 + ) + if method != "app.login" and not self._token: + raise FlowError( + E.CONFIG_REQUIRED, "Official account authorization required", 503 + ) + try: + response = requests.post( + "https://robot.galaxyguide.cn/api/rpc", + headers={"Authorization": "Bearer " + self._token} + if method != "app.login" + else {}, + json={ + "jsonrpc": "2.0", + "method": method, + "params": params, + "id": int(time.time() * 1000), + }, + timeout=15, + allow_redirects=False, + ) + if response.status_code in (401, 403): + raise FlowError( + E.UNAUTHORIZED, "Official account authentication required", 401 + ) + payload = response.json() + error = payload.get("error") + if isinstance(error, dict) and error.get("message") == "请重新登陆": + raise FlowError(E.UNAUTHORIZED, "Official account session expired", 401) + response.raise_for_status() + if error or not isinstance(payload.get("result"), dict): + raise ValueError() + return payload["result"] + except FlowError: + raise + except (requests.RequestException, ValueError, AttributeError): + raise FlowError( + E.CLOUD_UNAVAILABLE, "Official service query failed", 502 + ) from None + + def login(self, username: str, password: str) -> None: + if not username or not password: + raise FlowError(E.INVALID_REQUEST, "Official account credentials required") + result = self._call("app.login", {"username": username, "password": password}) + token = result.get("token") + if not isinstance(token, str) or not token: + raise FlowError(E.UNAUTHORIZED, "Official account login failed", 401) + self._token = token + + def disconnect(self) -> None: + self._token = "" + + def devices(self) -> dict: + return self._call("app.getRobots", {}) + + def device(self, robot_id: int) -> dict: + return self._call("appFrame.getFrame", {"id": robot_id}) + + def widgets(self, robot_id: int) -> dict: + return self._call("appFrame.getFrameCovers", {"robotId": robot_id}) diff --git a/pigway_frame_flow/drivers/kokonna/widget_catalog.json b/pigway_frame_flow/integrations/kokonna_official/widget_catalog.json similarity index 100% rename from pigway_frame_flow/drivers/kokonna/widget_catalog.json rename to pigway_frame_flow/integrations/kokonna_official/widget_catalog.json diff --git a/pigway_frame_flow/integrations/kokonna_official/widgets.py b/pigway_frame_flow/integrations/kokonna_official/widgets.py new file mode 100644 index 0000000..56f5d56 --- /dev/null +++ b/pigway_frame_flow/integrations/kokonna_official/widgets.py @@ -0,0 +1,100 @@ +"""Read-only official widget RPC; separate from the device OpenAPI credential.""" + +import json +import time +from pathlib import Path + +from pigway_frame_flow.errors import ErrorCode as E +from pigway_frame_flow.errors import FlowError +from pigway_frame_flow.integrations.kokonna_official.rpc import OfficialRPC +from pigway_frame_flow.models import Widget, WidgetState + + +class WidgetReader: + def __init__(self, token="", robot_id=None, mock=False): + self.rpc = OfficialRPC(token) + self._robot_id = robot_id + self.mock = mock + self._cached = None + self._cached_at = 0.0 + + def read(self) -> WidgetState: + if ( + self.rpc.authorized + and self._cached is not None + and time.monotonic() - self._cached_at < 60 + ): + return self._cached + snapshot = json.loads( + Path(__file__).with_name("widget_catalog.json").read_text() + ) + catalog = [ + Widget(**{k: row[k] for k in ("id", "name", "tags", "arg_keys")}) + for row in snapshot["catalog"] + ] + if self.mock: + return WidgetState( + source="mock", + configured=True, + catalog=catalog, + current=catalog[0], + current_verified=True, + ) + if not self.rpc.authorized or self._robot_id is None: + return WidgetState( + source="snapshot", + configured=False, + catalog=catalog, + observed_at=snapshot["observed_at"], + status="AUTH_REQUIRED", + ) + try: + data = self.rpc.widgets(self._robot_id) + rows = data["rows"] + catalog = [ + Widget( + id=row["id"], + name=row["name"], + tags=row.get("tags") or [], + arg_keys=list((row.get("args") or {}).keys()) + if isinstance(row.get("args"), dict) + else [], + ) + for row in rows + ] + selected = data.get("frameCoverRobot") or {} + cover = selected.get("frameCover") or {} + current_id = cover.get("id", selected.get("coverId")) + current = next((row for row in catalog if row.id == current_id), None) + robot = data.get("robot") or {} + no_widget = ( + "frameCoverRobot" in data + and data["frameCoverRobot"] is None + and "frameCoverRobotId" in robot + and robot["frameCoverRobotId"] is None + ) + result = WidgetState( + source="live", + configured=True, + catalog=catalog, + current=current, + current_verified=current is not None or no_widget, + status="OK" + if current + else "NO_WIDGET" + if no_widget + else "CURRENT_UNVERIFIED", + ) + except FlowError: + raise + except ( + ValueError, + KeyError, + TypeError, + AttributeError, + ): + raise FlowError( + E.CLOUD_UNAVAILABLE, "Official widget query unavailable", 502 + ) from None + self._cached, self._cached_at = result, time.monotonic() + return result diff --git a/pyproject.toml b/pyproject.toml index 43a325d..6114468 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -16,7 +16,7 @@ build-backend = "setuptools.build_meta" [tool.setuptools.packages.find] include = ["pigway_frame_flow*", "vendor*"] [tool.setuptools.package-data] -pigway_frame_flow = ["templates/*.json", "web/*", "drivers/kokonna/*.json"] +pigway_frame_flow = ["templates/*.json", "web/*", "integrations/kokonna_official/*.json"] [tool.ruff.lint] select = ["E4", "E7", "E9", "F", "I", "B"] ignore = ["B008"] # FastAPI dependency defaults are declarative framework markers. diff --git a/tests/test_official_module.py b/tests/test_official_module.py new file mode 100644 index 0000000..28d51df --- /dev/null +++ b/tests/test_official_module.py @@ -0,0 +1,94 @@ +from pathlib import Path +from unittest.mock import Mock, patch + +import pytest + +from pigway_frame_flow.errors import ErrorCode, FlowError +from pigway_frame_flow.integrations.kokonna_official.capabilities import capabilities +from pigway_frame_flow.integrations.kokonna_official.rpc import OfficialRPC + + +def test_login_session_not_password_used_for_reads(): + login = Mock(status_code=200) + login.json.return_value = {"result": {"token": "session-secret"}} + devices = Mock(status_code=200) + devices.json.return_value = {"result": {"data": []}} + rpc = OfficialRPC() + with patch("requests.post", side_effect=[login, devices]) as post: + rpc.login("account", "password-secret") + assert rpc.devices() == {"data": []} + assert post.call_args.kwargs["headers"] == { + "Authorization": "Bearer session-secret" + } + assert "password-secret" not in str(post.call_args.kwargs) + rpc.disconnect() + assert not rpc.authorized + with patch("requests.post") as post, pytest.raises(FlowError): + rpc.devices() + post.assert_not_called() + + +@pytest.mark.parametrize( + "method", + [ + "appFrame.setFrameCover", + "appFrame.reset", + "appFrame.uploadImage", + "arbitrary.method", + ], +) +def test_all_unenabled_operations_blocked_without_network(method): + with patch("requests.post") as post, pytest.raises(FlowError) as error: + OfficialRPC("session")._call(method, {}) + assert error.value.code == ErrorCode.UNSUPPORTED_CAPABILITY + post.assert_not_called() + + +def test_rpc_http_500_relogin_is_auth_error_not_cloud_failure(): + response = Mock(status_code=500) + response.json.return_value = {"error": {"code": -1, "message": "请重新登陆"}} + with ( + patch("requests.post", return_value=response), + pytest.raises(FlowError) as error, + ): + OfficialRPC("secret").devices() + assert error.value.code == ErrorCode.UNAUTHORIZED + assert "secret" not in str(error.value) + + +def test_inventory_does_not_claim_unverified_writes_implemented(): + entries = {entry.method: entry for entry in capabilities()} + assert entries["appFrame.setRotate"].implemented is False + assert entries["appFrame.getFrameCovers"].evidence == "live_verified" + assert all( + not entry.implemented + for entry in entries.values() + if entry.evidence == "apk_discovered" + ) + + +def test_api_key_driver_has_no_official_dependency(): + for name in ["client.py", "driver.py"]: + source = (Path("pigway_frame_flow/drivers/kokonna") / name).read_text() + assert "kokonna_official" not in source + assert "WidgetReader" not in source + + +def test_widget_disconnect_does_not_return_authenticated_cache(): + from pigway_frame_flow.integrations.kokonna_official.widgets import WidgetReader + + response = Mock(status_code=200) + response.json.return_value = { + "result": { + "rows": [], + "robot": {"frameCoverRobotId": None}, + "frameCoverRobot": None, + } + } + reader = WidgetReader("session", 123) + with patch("requests.post", return_value=response): + assert reader.read().status == "NO_WIDGET" + reader.rpc.disconnect() + with patch("requests.post") as post: + assert reader.read().status == "AUTH_REQUIRED" + post.assert_not_called()