from __future__ import annotations import asyncio import json import threading from datetime import UTC, datetime from pathlib import Path from typing import Any from uuid import uuid4 import pytest from missioncore_plugin_sdk.v0alpha2 import ( RuntimeActionInvocation, RuntimeActionResult, RuntimeHandshakeRequest, RuntimeHandshakeResult, RuntimePluginDescriptor, ) from pydantic import ValidationError import k1link.web.device_plugin_composition as plugin_composition from k1link.device_plugins.xgrids_k1.facade import ( ACTION_ACQUISITION_ABORT, ACTION_ACQUISITION_PREPARE, ACTION_ACQUISITION_START, ACTION_ACQUISITION_STOP, ACTION_APPLICATION_CONTROL_SESSION_CLOSE, ACTION_APPLICATION_CONTROL_SESSION_OPEN, ACTION_APPLICATION_CONTROL_WORKSPACE_ENTER, ACTION_CONFIGURED_ENDPOINT_PROBE, ACTION_CONNECTION_MODE_SELECT, ACTION_CONNECTION_RECONFIGURE_PREPARE, ACTION_CONNECTION_VERIFY, ACTION_DEVICE_CALIBRATION_SNAPSHOT_READ, ACTION_DISCOVERY_SCAN, ACTION_NETWORK_PROVISION, ACTION_PHYSICAL_COMMAND_RECONCILE, ACTION_PHYSICAL_COMMAND_REOPEN_RETIRED_RECONCILIATION, ACTION_PHYSICAL_COMMAND_RETIRE_UNAVAILABLE, ACTION_STATE_READ, ACTION_STREAM_START_LIVE, ACTION_STREAM_START_REPLAY, ACTION_STREAM_STOP, ACTION_VIEWER_SETTINGS_UPDATE, XGRIDS_K1_PLUGIN_ID, XGRIDS_K1_PLUGIN_VERSION, AbortAcquisitionRequest, BleScanRequest, CloseApplicationControlSessionRequest, CompatibilityAttestationRequest, ConfiguredEndpointProbeRequest, ConnectionVerificationError, ConnectionVerifyRequest, ConnectRequest, DesiredConnectionModeRequest, EnterApplicationWorkspaceRequest, NetworkProvisioningConflict, OpenApplicationControlSessionRequest, PrepareAcquisitionRequest, PrepareConnectionReconfigurationRequest, ReconcilePhysicalCommandRequest, ReopenRetiredPhysicalCommandReconciliationRequest, RetireUnavailablePhysicalCommandRequest, SnapshotRuntimeConflict, StartAcquisitionRequest, StopAcquisitionRequest, ViewerSettingsRequest, XgridsK1PluginFacade, ) from k1link.web.device_plugin_composition import ( DevicePluginCompositionError, InstalledDevicePluginEnvironment, load_installed_device_plugins, ) from k1link.web.plugin_catalog import DevicePluginCatalog from k1link.web.plugin_runtime import ( DevicePluginDispatcher, DevicePluginRuntimeContribution, InProcessDevicePluginRuntime, PluginActionNotFoundError, PluginExecutionError, PluginNotFoundError, PluginRuntimeCompatibilityError, PluginRuntimeUnavailableError, ) class FakeXgridsService: def __init__(self) -> None: self.calls: list[tuple[str, object]] = [] self.scan_loop: asyncio.AbstractEventLoop | None = None self.verify_loop: asyncio.AbstractEventLoop | None = None self.snapshot_runtime_id = "snapshot-runtime-test" self.bind_calls = 0 def require_snapshot_runtime_id(self, expected_snapshot_runtime_id: str) -> None: if expected_snapshot_runtime_id != self.snapshot_runtime_id: raise SnapshotRuntimeConflict() def bind_runtime_event_loop( self, loop: asyncio.AbstractEventLoop | None = None, ) -> None: del loop self.bind_calls += 1 def state(self) -> dict[str, Any]: self.calls.append(("state", None)) return {"phase": "idle"} def read_device_calibration_snapshot(self) -> dict[str, Any]: self.calls.append(("calibration", None)) return {"status": "available", "snapshot_id": "fixture-snapshot"} async def scan_ble(self, request: BleScanRequest) -> dict[str, Any]: self.scan_loop = asyncio.get_running_loop() self.calls.append(("scan", request)) return {"phase": "idle", "devices": []} async def connect(self, request: ConnectRequest) -> dict[str, Any]: self.calls.append(("connect", request)) return {"phase": "connected", "k1_ip": "192.168.1.20"} def select_connection_mode( self, request: DesiredConnectionModeRequest, ) -> dict[str, Any]: self.calls.append(("mode-select", request)) return { "phase": "idle", "desired_connection_mode": request.connection_mode, "desired_connection_mode_revision": request.expected_revision + 1, } async def prepare_connection_reconfiguration( self, request: PrepareConnectionReconfigurationRequest, ) -> dict[str, Any]: self.calls.append(("reconfigure", request)) return { "phase": "idle", "connection_reconfiguration": { "revision": request.expected_reconfiguration_revision + 1, "intent": request.intent, }, } async def verify_connection( self, request: ConnectionVerifyRequest | None = None, ) -> dict[str, Any]: self.verify_loop = asyncio.get_running_loop() self.calls.append(("verify", request)) return {"phase": "connected", "k1_ip": "192.168.1.20"} async def probe_configured_endpoint( self, request: ConfiguredEndpointProbeRequest | None = None, ) -> dict[str, Any]: self.calls.append(("endpoint-probe", request)) return { "phase": "idle", "configured_endpoint_probe": { "status": "reachable", "ble_operation_performed": False, }, } def start_live( self, project_name: str, host: str | None, duration_seconds: float | None, compatibility_attestation: CompatibilityAttestationRequest, ) -> dict[str, Any]: self.calls.append( ("live", (project_name, host, duration_seconds, compatibility_attestation)) ) return {"phase": "live"} def start_replay(self, path: str, speed: float, loop: bool) -> dict[str, Any]: self.calls.append(("replay", (path, speed, loop))) return {"phase": "replay"} def stop(self) -> dict[str, Any]: self.calls.append(("stop", None)) return {"phase": "idle"} def prepare_acquisition(self, request: PrepareAcquisitionRequest) -> dict[str, Any]: self.calls.append(("prepare", request)) return {"phase": "connected", "acquisition": {"state": "prepared"}} def start_acquisition(self, request: StartAcquisitionRequest) -> dict[str, Any]: self.calls.append(("start", request)) return {"phase": "starting_live"} def stop_acquisition(self, request: StopAcquisitionRequest) -> dict[str, Any]: self.calls.append(("acquisition-stop", request)) return {"phase": "stopping"} def abort_acquisition(self, request: AbortAcquisitionRequest) -> dict[str, Any]: self.calls.append(("abort", request)) return {"phase": "idle"} def open_application_control_session( self, request: OpenApplicationControlSessionRequest, ) -> dict[str, Any]: self.calls.append(("control-open", request)) return {"phase": "connected"} def enter_application_workspace( self, request: EnterApplicationWorkspaceRequest, ) -> dict[str, Any]: self.calls.append(("control-enter", request)) return {"phase": "connected"} def close_application_control_session( self, request: CloseApplicationControlSessionRequest, ) -> dict[str, Any]: self.calls.append(("control-close", request)) return {"phase": "connected"} def reconcile_physical_command( self, request: ReconcilePhysicalCommandRequest, ) -> dict[str, Any]: self.calls.append(("physical-reconcile", request)) return {"phase": "connected"} def update_viewer_settings(self, request: ViewerSettingsRequest) -> dict[str, Any]: self.calls.append(("viewer", request)) return {"phase": "idle", "viewer_settings": request.model_dump()} def retire_unavailable_physical_command( self, request: RetireUnavailablePhysicalCommandRequest, ) -> dict[str, Any]: self.calls.append(("physical-retire", request)) return { "phase": "idle", "physical_command": { "status": "resolved", "physical_outcome": "unknown", }, } def reopen_retired_physical_command_reconciliation( self, request: ReopenRetiredPhysicalCommandReconciliationRequest, ) -> dict[str, Any]: self.calls.append(("physical-reopen", request)) return { "phase": "idle", "physical_command": { "status": "unresolved", "requires_reconciliation": True, }, } def _snapshot_fenced(payload: dict[str, Any] | None = None) -> dict[str, Any]: return { **(payload or {}), "expected_snapshot_runtime_id": "snapshot-runtime-test", } def _in_process_runtime( adapter: Any, *, plugin_version: str = XGRIDS_K1_PLUGIN_VERSION, host_api_version: str = "missioncore.nodedc/v1alpha2", close: Any | None = None, activate: bool = True, ) -> InProcessDevicePluginRuntime: runtime = InProcessDevicePluginRuntime( adapter, RuntimePluginDescriptor( plugin_id=adapter.plugin_id, plugin_version=plugin_version, supported_host_api_versions=(host_api_version,), action_ids=tuple(sorted(adapter.action_ids)), ), **({"close": close} if close is not None else {}), ) if activate: runtime.handshake( RuntimeHandshakeRequest( handshake_id=uuid4().hex, plugin_id=adapter.plugin_id, plugin_version=plugin_version, host_api_version=host_api_version, requested_at=datetime.now(UTC), required_action_ids=tuple(sorted(adapter.action_ids)), ) ) return runtime def test_manifest_and_runtime_facade_declare_identical_actions() -> None: repository_root = Path(__file__).resolve().parents[1] manifest = next( candidate for candidate in DevicePluginCatalog(repository_root).manifests() if candidate.metadata.id == XGRIDS_K1_PLUGIN_ID ) assert {action.id for action in manifest.spec.actions} == XgridsK1PluginFacade.action_ids def test_calibration_snapshot_action_calls_the_read_only_service_method() -> None: service = FakeXgridsService() dispatcher = DevicePluginDispatcher([_in_process_runtime(XgridsK1PluginFacade(service))]) result = asyncio.run( dispatcher.invoke( XGRIDS_K1_PLUGIN_ID, ACTION_DEVICE_CALIBRATION_SNAPSHOT_READ, {}, ) ) assert result == {"status": "available", "snapshot_id": "fixture-snapshot"} assert service.calls == [("calibration", None)] def test_connection_verify_action_accepts_read_only_adoption_request() -> None: service = FakeXgridsService() dispatcher = DevicePluginDispatcher([_in_process_runtime(XgridsK1PluginFacade(service))]) result = asyncio.run( dispatcher.invoke( XGRIDS_K1_PLUGIN_ID, ACTION_CONNECTION_VERIFY, _snapshot_fenced({ "device_id": "test-ble-transport", "compatibility_attestation": { "firmware_version": "3.0.2", "topology": "direct-lan", "verification": "live-device-info", }, "expected_discovery_generation": 0, }), ) ) assert result == {"phase": "connected", "k1_ip": "192.168.1.20"} assert len(service.calls) == 1 action, request = service.calls[0] assert action == "verify" assert isinstance(request, ConnectionVerifyRequest) assert request.device_id == "test-ble-transport" assert request.compatibility_attestation is not None assert request.compatibility_attestation.topology == "direct-lan" def test_connection_verify_action_keeps_empty_refresh_request_compatible() -> None: service = FakeXgridsService() dispatcher = DevicePluginDispatcher([_in_process_runtime(XgridsK1PluginFacade(service))]) asyncio.run( dispatcher.invoke( XGRIDS_K1_PLUGIN_ID, ACTION_CONNECTION_VERIFY, _snapshot_fenced(), ) ) action, request = service.calls[0] assert action == "verify" assert isinstance(request, ConnectionVerifyRequest) assert request.device_id is None assert request.compatibility_attestation is None def test_connection_mode_select_delegates_exact_cas_payload() -> None: service = FakeXgridsService() dispatcher = DevicePluginDispatcher([_in_process_runtime(XgridsK1PluginFacade(service))]) result = asyncio.run( dispatcher.invoke( XGRIDS_K1_PLUGIN_ID, ACTION_CONNECTION_MODE_SELECT, _snapshot_fenced({ "connection_mode": "quick-connect", "expected_revision": 7, }), ) ) assert result["desired_connection_mode"] == "quick-connect" assert result["desired_connection_mode_revision"] == 8 assert len(service.calls) == 1 action, request = service.calls[0] assert action == "mode-select" assert isinstance(request, DesiredConnectionModeRequest) assert request.connection_mode == "quick-connect" assert request.expected_revision == 7 def test_connection_scenario_reset_dispatches_before_runtime_loop_binding() -> None: service = FakeXgridsService() dispatcher = DevicePluginDispatcher([_in_process_runtime(XgridsK1PluginFacade(service))]) result = asyncio.run( dispatcher.invoke( XGRIDS_K1_PLUGIN_ID, ACTION_CONNECTION_MODE_SELECT, _snapshot_fenced( { "connection_mode": "bridge", "expected_revision": 3, "reset_scenario": True, "reset_id": "op-reset-dispatch-local-only-01", } ), ) ) assert result["desired_connection_mode_revision"] == 4 assert service.bind_calls == 0 assert len(service.calls) == 1 action, request = service.calls[0] assert action == "mode-select" assert isinstance(request, DesiredConnectionModeRequest) assert request.reset_scenario is True assert request.reset_id == "op-reset-dispatch-local-only-01" @pytest.mark.parametrize( ("intent", "intent_id"), [("select-device", None), ("change-network", None), ("cancel", "intent-9")], ) def test_connection_reconfigure_action_delegates_exact_stable_cas_payload( intent: str, intent_id: str | None, ) -> None: service = FakeXgridsService() dispatcher = DevicePluginDispatcher([_in_process_runtime(XgridsK1PluginFacade(service))]) payload = { "intent": intent, "expected_reconfiguration_revision": 9, "expected_reconfiguration_intent_id": intent_id, "expected_desired_mode_revision": 4, "expected_active_binding_key": "a" * 64 if intent_id is None else None, } result = asyncio.run( dispatcher.invoke( XGRIDS_K1_PLUGIN_ID, ACTION_CONNECTION_RECONFIGURE_PREPARE, _snapshot_fenced(payload), ) ) assert result["connection_reconfiguration"] == { "revision": 10, "intent": intent, } assert len(service.calls) == 1 action, request = service.calls[0] assert action == "reconfigure" assert isinstance(request, PrepareConnectionReconfigurationRequest) assert request.model_dump(mode="json") == payload def test_configured_endpoint_probe_action_is_separate_from_ble_verify() -> None: service = FakeXgridsService() dispatcher = DevicePluginDispatcher([_in_process_runtime(XgridsK1PluginFacade(service))]) result = asyncio.run( dispatcher.invoke( XGRIDS_K1_PLUGIN_ID, ACTION_CONFIGURED_ENDPOINT_PROBE, _snapshot_fenced( {"operation_id": "op-00000000-0000-4000-8000-000000000652"} ), ) ) assert result["configured_endpoint_probe"] == { "status": "reachable", "ble_operation_performed": False, } assert len(service.calls) == 1 action, request = service.calls[0] assert action == "endpoint-probe" assert isinstance(request, ConfiguredEndpointProbeRequest) assert request.operation_id == "op-00000000-0000-4000-8000-000000000652" def test_physical_retirement_action_delegates_exact_confirmed_cas_payload() -> None: service = FakeXgridsService() dispatcher = DevicePluginDispatcher([_in_process_runtime(XgridsK1PluginFacade(service))]) payload = { "retirement_id": "retirement-browser-stable-id", "expected_operation_id": "physical-stop-persisted", "expected_revision": 17, "expected_transport_ref": "F89438FA-55ED-85AD-EED7-734AC84746D8", "operator_confirmed": True, "reason": "device-permanently-unavailable-or-replaced", } result = asyncio.run( dispatcher.invoke( XGRIDS_K1_PLUGIN_ID, ACTION_PHYSICAL_COMMAND_RETIRE_UNAVAILABLE, _snapshot_fenced(payload), ) ) assert result["physical_command"] == { "status": "resolved", "physical_outcome": "unknown", } assert service.bind_calls == 0 assert len(service.calls) == 1 action, request = service.calls[0] assert action == "physical-retire" assert isinstance(request, RetireUnavailablePhysicalCommandRequest) assert request.model_dump(mode="json") == payload def test_physical_reopen_action_is_runtime_fenced_exact_and_never_binds_loop() -> None: service = FakeXgridsService() adapter = XgridsK1PluginFacade(service) dispatcher = DevicePluginDispatcher([_in_process_runtime(adapter)]) payload = { "reopening_id": "reopening-browser-stable-id", "expected_revision": 18, "expected_retirement_id": "retirement-browser-stable-id", "expected_transport_ref": "f89438fa-55ed-85ad-eed7-734ac84746d8", "expected_discovery_generation": 7, "expected_desired_mode": "bridge", "expected_desired_mode_revision": 4, "operator_confirmed": True, "reason": "device-returned-for-explicit-reconciliation", } result = asyncio.run( dispatcher.invoke( XGRIDS_K1_PLUGIN_ID, ACTION_PHYSICAL_COMMAND_REOPEN_RETIRED_RECONCILIATION, _snapshot_fenced(payload), ) ) assert result["physical_command"] == { "status": "unresolved", "requires_reconciliation": True, } assert service.bind_calls == 0 assert len(service.calls) == 1 action, request = service.calls[0] assert action == "physical-reopen" assert isinstance(request, ReopenRetiredPhysicalCommandReconciliationRequest) assert request.model_dump(mode="json") == payload assert ACTION_PHYSICAL_COMMAND_REOPEN_RETIRED_RECONCILIATION in adapter.action_ids @pytest.mark.parametrize( "reason_code", [ "physical-command-reconciliation-reopen-stale-checkpoint", "network-provisioning-idempotency-operation-mismatch", "reconciliation-target-mode-mismatch", "reconciliation-target-physical-recovery-mismatch", "fresh-ble-candidate-required", "physical-command-recovery-target-not-observed", "network-provision-operation-active", "control-local-retirement-pending", "device-calibration-read-active", "acquisition-active", "acquisition-cleanup-pending", "acquisition-start-operation-active", "acquisition-stop-operation-active", "local-runtime-active", "control-session-not-admissible-for-network-change", "k1-lifecycle-process-lease-control-owned", "k1-lifecycle-process-lease-active", ], ) def test_physical_reopen_expected_conflicts_are_http_409(reason_code: str) -> None: class ReopenConflictService(FakeXgridsService): def reopen_retired_physical_command_reconciliation( self, request: ReopenRetiredPhysicalCommandReconciliationRequest, ) -> dict[str, Any]: del request raise NetworkProvisioningConflict( "reopen checkpoint is no longer executable", reason_code=reason_code, ) dispatcher = DevicePluginDispatcher( [_in_process_runtime(XgridsK1PluginFacade(ReopenConflictService()))] ) payload = { "reopening_id": "reopening-http-conflict", "expected_revision": 18, "expected_retirement_id": "retirement-http-conflict", "expected_transport_ref": "f89438fa-55ed-85ad-eed7-734ac84746d8", "expected_discovery_generation": 7, "expected_desired_mode": "bridge", "expected_desired_mode_revision": 4, "operator_confirmed": True, "reason": "device-returned-for-explicit-reconciliation", } with pytest.raises(PluginExecutionError) as raised: asyncio.run( dispatcher.invoke( XGRIDS_K1_PLUGIN_ID, ACTION_PHYSICAL_COMMAND_REOPEN_RETIRED_RECONCILIATION, _snapshot_fenced(payload), ) ) assert raised.value.http_status_code == 409 assert raised.value.reason_code == reason_code @pytest.mark.parametrize( "action_id", [ ACTION_DISCOVERY_SCAN, ACTION_NETWORK_PROVISION, ACTION_CONNECTION_MODE_SELECT, ACTION_CONNECTION_RECONFIGURE_PREPARE, ACTION_CONNECTION_VERIFY, ACTION_CONFIGURED_ENDPOINT_PROBE, ACTION_ACQUISITION_PREPARE, ACTION_ACQUISITION_START, ACTION_ACQUISITION_STOP, ACTION_ACQUISITION_ABORT, ACTION_APPLICATION_CONTROL_SESSION_OPEN, ACTION_APPLICATION_CONTROL_WORKSPACE_ENTER, ACTION_APPLICATION_CONTROL_SESSION_CLOSE, ACTION_PHYSICAL_COMMAND_RECONCILE, ACTION_STREAM_STOP, ACTION_PHYSICAL_COMMAND_RETIRE_UNAVAILABLE, ACTION_PHYSICAL_COMMAND_REOPEN_RETIRED_RECONCILIATION, ], ) @pytest.mark.parametrize( "runtime_fence", [None, "snapshot-runtime-stale-browser"], ) def test_snapshot_fenced_actions_reject_stale_browser_before_service_or_io( action_id: str, runtime_fence: str | None, ) -> None: service = FakeXgridsService() dispatcher = DevicePluginDispatcher([_in_process_runtime(XgridsK1PluginFacade(service))]) payload = ( {} if runtime_fence is None else {"expected_snapshot_runtime_id": runtime_fence} ) with pytest.raises(PluginExecutionError) as raised: asyncio.run(dispatcher.invoke(XGRIDS_K1_PLUGIN_ID, action_id, payload)) assert raised.value.http_status_code == 409 assert raised.value.reason_code == "snapshot-runtime-conflict" assert service.bind_calls == 0 assert service.calls == [] def test_lifecycle_ui_payload_contract_passes_backend_validation_exactly() -> None: service = FakeXgridsService() dispatcher = DevicePluginDispatcher([_in_process_runtime(XgridsK1PluginFacade(service))]) compatibility = { "firmware_version": "3.0.2", "topology": "direct-lan", "verification": "live-device-info", } payloads = ( ( ACTION_ACQUISITION_PREPARE, { "operation_id": "op-00000000-0000-4000-8000-000000000201", "idempotency_key": ( "acquisition.prepare:op-00000000-0000-4000-8000-000000000201" ), "project_name": "CONTRACT01", "mount_type": "handheld", "gnss_mode": "none", "compatibility_attestation": compatibility, "expected_control_session_generation": 7, "expected_control_state_revision": 11, }, ), ( ACTION_ACQUISITION_START, { "operation_id": "op-00000000-0000-4000-8000-000000000202", "idempotency_key": ( "acquisition.start:op-00000000-0000-4000-8000-000000000202" ), "acquisition_id": "acquisition-contract", "expected_control_session_generation": 7, "expected_control_state_revision": 12, }, ), ( ACTION_ACQUISITION_STOP, { "operation_id": "op-00000000-0000-4000-8000-000000000203", "idempotency_key": ( "acquisition.stop:op-00000000-0000-4000-8000-000000000203" ), "acquisition_id": "acquisition-contract", "mode": "graceful", "expected_control_session_generation": 7, "expected_control_state_revision": 13, }, ), ( ACTION_ACQUISITION_ABORT, { "operation_id": "op-00000000-0000-4000-8000-000000000204", "idempotency_key": ( "acquisition.abort:op-00000000-0000-4000-8000-000000000204" ), "acquisition_id": "acquisition-contract", "expected_control_session_generation": 7, "expected_control_state_revision": 14, }, ), ( ACTION_APPLICATION_CONTROL_SESSION_OPEN, { "operator_present": True, "owner_controlled_device": True, "lixelgo_closed": True, "battery_storage_confirmed": True, "expected_physical_state_confirmed": True, "timezone_name": "Europe/Moscow", }, ), ( ACTION_APPLICATION_CONTROL_WORKSPACE_ENTER, { "operator_confirmed": True, "expected_session_generation": 7, "expected_state_revision": 15, }, ), ( ACTION_APPLICATION_CONTROL_SESSION_CLOSE, { "expected_session_generation": 7, "expected_state_revision": 16, }, ), ( ACTION_PHYSICAL_COMMAND_RECONCILE, { "reconciliation_id": "reconciliation-contract", "expected_session_generation": 7, "expected_state_revision": 17, }, ), ) for action_id, payload in payloads: asyncio.run( dispatcher.invoke( XGRIDS_K1_PLUGIN_ID, action_id, _snapshot_fenced(payload), ) ) assert [call[0] for call in service.calls] == [ "prepare", "start", "acquisition-stop", "abort", "control-open", "control-enter", "control-close", "physical-reconcile", ] for _, request in service.calls[:4]: assert request.operation_id is not None assert request.idempotency_key.endswith(request.operation_id) def test_connection_verify_expected_state_is_not_reported_as_bad_gateway() -> None: class AddressUnavailableService(FakeXgridsService): async def verify_connection( self, request: ConnectionVerifyRequest | None = None, ) -> dict[str, Any]: del request raise ConnectionVerificationError( "K1 не сообщил адрес общей локальной сети", reason_code="connection-verify-address-unavailable", ) dispatcher = DevicePluginDispatcher( [_in_process_runtime(XgridsK1PluginFacade(AddressUnavailableService()))] ) with pytest.raises(PluginExecutionError) as raised: asyncio.run( dispatcher.invoke( XGRIDS_K1_PLUGIN_ID, ACTION_CONNECTION_VERIFY, _snapshot_fenced(), ) ) assert raised.value.http_status_code == 409 assert raised.value.reason_code == "connection-verify-address-unavailable" @pytest.mark.parametrize( ("reason_code", "expected_status"), [ ("ble-runtime-busy", 409), ("ble-runtime-cleanup-pending", 409), ("provisioning-already-running", 409), ("connection-verify-mqtt-unreachable", 409), ("connection-verify-lease-changed", 409), ("connection-verify-resolved-apply-target-mismatch", 409), ("configured-endpoint-topology-corrupt", 409), ("connection-mode-selection-lifecycle-busy", 409), ("connection-mode-selection-physical-state-unsafe", 409), ("connection-mode-selection-control-state-unsafe", 409), ("connection-mode-switch-acquisition-changed", 409), ("connection-scenario-reset-pending", 409), ("connection-scenario-reset-lifecycle-timeout", 409), ("acquisition-start-lifecycle-busy", 409), ("application-control-process-lease-unavailable", 409), ("ble-runtime-owner-loop-conflict", 503), ("ble-runtime-restart-required", 503), ("connection-verify-exact-uuid-scan-timeout", 504), ("ble-discovery-timeout", 504), ("ble-status-read-timeout", 504), ("ble-provisioning-timeout", 504), ("ble-ap-enable-timeout", 504), ("network-not-found", 504), ("host-wifi-operation-timeout", 504), ("physical-command-reconciliation-proof-timeout", 504), ("physical-command-reconciliation-control-adoption-timeout", 504), ("keychain-authorization-required", 409), ("profile-unavailable", 409), ("profile-credential-source-mismatch", 409), ("wifi-interface-unavailable", 503), ], ) def test_ble_runtime_failure_preserves_actionable_http_class( reason_code: str, expected_status: int, ) -> None: class ClassifiedRuntimeError(RuntimeError): def __init__(self) -> None: super().__init__("classified BLE failure") self.reason_code = reason_code class FailingScanService(FakeXgridsService): async def scan_ble(self, request: BleScanRequest) -> dict[str, Any]: del request raise ClassifiedRuntimeError() dispatcher = DevicePluginDispatcher( [_in_process_runtime(XgridsK1PluginFacade(FailingScanService()))] ) with pytest.raises(PluginExecutionError) as raised: asyncio.run( dispatcher.invoke( XGRIDS_K1_PLUGIN_ID, ACTION_DISCOVERY_SCAN, _snapshot_fenced({"duration_seconds": 6}), ) ) assert raised.value.http_status_code == expected_status assert raised.value.reason_code == reason_code def test_connection_verify_reuses_the_discovery_event_loop() -> None: service = FakeXgridsService() dispatcher = DevicePluginDispatcher([_in_process_runtime(XgridsK1PluginFacade(service))]) async def scenario() -> asyncio.AbstractEventLoop: loop = asyncio.get_running_loop() await dispatcher.invoke( XGRIDS_K1_PLUGIN_ID, ACTION_DISCOVERY_SCAN, _snapshot_fenced({"duration_seconds": 6}), ) await dispatcher.invoke( XGRIDS_K1_PLUGIN_ID, ACTION_CONNECTION_VERIFY, _snapshot_fenced({ "device_id": "test-ble-transport", "compatibility_attestation": { "firmware_version": "3.0.2", "topology": "direct-lan", "verification": "live-device-info", }, "expected_discovery_generation": 0, }), ) return loop dispatcher_loop = asyncio.run(scenario()) assert service.scan_loop is dispatcher_loop assert service.verify_loop is dispatcher_loop def test_repository_runtime_composition_exactly_matches_catalog() -> None: repository_root = Path(__file__).resolve().parents[1] environment = load_installed_device_plugins(repository_root) try: assert set(environment.dispatcher.action_declarations) == { manifest.metadata.id for manifest in environment.catalog.manifests() } finally: environment.close() def test_composition_closes_a_runtime_that_does_not_match_its_manifest( monkeypatch: pytest.MonkeyPatch, ) -> None: repository_root = Path(__file__).resolve().parents[1] manifest = next( candidate for candidate in DevicePluginCatalog(repository_root).manifests() if candidate.metadata.id == XGRIDS_K1_PLUGIN_ID ) mismatched_manifest = manifest.model_copy( update={"metadata": manifest.metadata.model_copy(update={"id": "example.other-plugin"})} ) closed: list[bool] = [] contribution = DevicePluginRuntimeContribution( runtime=_in_process_runtime( XgridsK1PluginFacade(FakeXgridsService()), close=lambda: closed.append(True), activate=False, ), ) monkeypatch.setattr( plugin_composition, "_load_factory", lambda _: lambda __: contribution, ) with pytest.raises(DevicePluginCompositionError, match="id mismatch"): plugin_composition._load_contribution(repository_root, mismatched_manifest) assert closed == [True] def test_composition_rejects_an_uncorrelated_runtime_handshake( monkeypatch: pytest.MonkeyPatch, ) -> None: class UncorrelatedHandshakeRuntime(InProcessDevicePluginRuntime): def handshake( self, request: RuntimeHandshakeRequest, ) -> RuntimeHandshakeResult: result = super().handshake(request) return result.model_copy(update={"handshake_id": "another-handshake"}) repository_root = Path(__file__).resolve().parents[1] manifest = next( candidate for candidate in DevicePluginCatalog(repository_root).manifests() if candidate.metadata.id == XGRIDS_K1_PLUGIN_ID ) closed: list[bool] = [] adapter = XgridsK1PluginFacade(FakeXgridsService()) runtime = UncorrelatedHandshakeRuntime( adapter, RuntimePluginDescriptor( plugin_id=adapter.plugin_id, plugin_version=XGRIDS_K1_PLUGIN_VERSION, supported_host_api_versions=(manifest.apiVersion,), action_ids=tuple(sorted(adapter.action_ids)), ), close=lambda: closed.append(True), ) monkeypatch.setattr( plugin_composition, "_load_factory", lambda _: lambda __: DevicePluginRuntimeContribution(runtime=runtime), ) with pytest.raises(DevicePluginCompositionError, match="handshake is inconsistent"): plugin_composition._load_contribution(repository_root, manifest) assert closed == [True] def test_composition_loads_and_dispatches_two_synthetic_plugins( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: class SyntheticAdapter: action_ids = frozenset({"state.read"}) def __init__(self, plugin_id: str) -> None: self.plugin_id = plugin_id async def invoke( self, invocation: RuntimeActionInvocation, ) -> dict[str, Any]: assert invocation.plugin_id == self.plugin_id assert invocation.action_id == "state.read" assert invocation.parameters == {} return {"plugin_id": self.plugin_id} closed: list[str] = [] contributions = { f"synthetic:{suffix}": DevicePluginRuntimeContribution( runtime=_in_process_runtime( SyntheticAdapter(f"example.{suffix}"), plugin_version="0.1.0", host_api_version="missioncore.nodedc/v1alpha1", close=lambda suffix=suffix: closed.append(suffix), activate=False, ), ) for suffix in ("first", "second") } for suffix in ("first", "second"): document = { "apiVersion": "missioncore.nodedc/v1alpha1", "kind": "DevicePlugin", "metadata": { "id": f"example.{suffix}", "version": "0.1.0", "displayName": f"Example {suffix}", }, "spec": { "hostApiRange": "v1alpha1", "runtime": { "backendEntrypoint": f"synthetic:{suffix}", "isolation": "transitional-in-process", }, "permissions": [], "actions": [{"id": "state.read", "mutating": False, "secretFields": []}], "models": [ { "id": f"example.{suffix}-model", "vendor": "Example", "displayName": f"Example {suffix}", "category": "Synthetic", "description": "Synthetic runtime composition fixture.", "verified": False, "capabilities": [], "ui": { "slot": "device.connection", "componentKey": f"example.{suffix}.connection", }, } ], }, } manifest_path = tmp_path / "plugins" / suffix / "plugin.manifest.json" manifest_path.parent.mkdir(parents=True) manifest_path.write_text(json.dumps(document), encoding="utf-8") monkeypatch.setattr( plugin_composition, "_load_factory", lambda entrypoint: lambda _: contributions[entrypoint], ) environment = load_installed_device_plugins(tmp_path) try: for suffix in ("first", "second"): state = asyncio.run( environment.dispatcher.invoke(f"example.{suffix}", "state.read", {}) ) assert state == {"plugin_id": f"example.{suffix}"} assert {item["status"] for item in environment.runtime_health} == {"ready"} finally: environment.close() assert closed == ["second", "first"] def test_environment_shutdown_attempts_every_plugin_after_close_failure( tmp_path: Path, ) -> None: closed: list[str] = [] def failing_close() -> None: closed.append("failing") raise RuntimeError("synthetic close failure") environment = InstalledDevicePluginEnvironment( catalog=DevicePluginCatalog(tmp_path), dispatcher=DevicePluginDispatcher([]), legacy_routers=(), _contributions=( DevicePluginRuntimeContribution( runtime=_in_process_runtime( XgridsK1PluginFacade(FakeXgridsService()), close=lambda: closed.append("healthy"), ), ), DevicePluginRuntimeContribution( runtime=_in_process_runtime( XgridsK1PluginFacade(FakeXgridsService()), close=failing_close, ), ), ), ) with pytest.raises(DevicePluginCompositionError, match="failed during shutdown") as raised: environment.close() assert closed == ["failing", "healthy"] assert "synthetic close failure" in " ".join(raised.value.__notes__) def test_dispatcher_routes_allowlisted_action_to_xgrids_facade() -> None: service = FakeXgridsService() dispatcher = DevicePluginDispatcher([_in_process_runtime(XgridsK1PluginFacade(service))]) state = asyncio.run( dispatcher.invoke( XGRIDS_K1_PLUGIN_ID, ACTION_DISCOVERY_SCAN, _snapshot_fenced({"duration_seconds": 6}), ) ) assert state == {"phase": "idle", "devices": []} assert len(service.calls) == 1 action, request = service.calls[0] assert action == "scan" assert isinstance(request, BleScanRequest) assert request.duration_seconds == 6.0 def test_runtime_classifies_non_json_plugin_output_as_execution_failure( caplog: pytest.LogCaptureFixture, ) -> None: retained_marker = "must-not-appear-in-the-server-log" class NonJsonAdapter: plugin_id = "example.non-json" action_ids = frozenset({"state.read"}) async def invoke( self, _invocation: RuntimeActionInvocation, ) -> dict[str, Any]: return {"dialogue": {"response_evidence": (retained_marker,)}} dispatcher = DevicePluginDispatcher( [ _in_process_runtime( NonJsonAdapter(), plugin_version="0.1.0", host_api_version="missioncore.nodedc/v1alpha2", ) ] ) with pytest.raises(PluginExecutionError, match="non-JSON"): asyncio.run(dispatcher.invoke("example.non-json", "state.read", {})) assert retained_marker not in caplog.text def test_dispatcher_rejects_unknown_plugin_and_action() -> None: dispatcher = DevicePluginDispatcher( [_in_process_runtime(XgridsK1PluginFacade(FakeXgridsService()))] ) with pytest.raises(PluginNotFoundError): asyncio.run(dispatcher.invoke("missing.plugin", ACTION_STREAM_STOP, {})) with pytest.raises(PluginActionNotFoundError): asyncio.run(dispatcher.invoke(XGRIDS_K1_PLUGIN_ID, "unknown.action", {})) def test_facade_validates_payload_before_calling_service() -> None: service = FakeXgridsService() dispatcher = DevicePluginDispatcher([_in_process_runtime(XgridsK1PluginFacade(service))]) with pytest.raises(ValidationError): asyncio.run( dispatcher.invoke( XGRIDS_K1_PLUGIN_ID, ACTION_NETWORK_PROVISION, _snapshot_fenced({ "device_id": "id", "ssid": "network", "password": "x" * 24, "extra": True, }), ) ) assert service.calls == [] @pytest.mark.parametrize( ("action_id", "payload", "expected_call"), [ ( ACTION_STREAM_START_LIVE, { "project_name": "Plugin runtime test", "host": "192.168.1.20", "compatibility_attestation": { "firmware_version": "3.0.2", "topology": "direct-lan", "verification": "live-device-info", }, }, "live", ), ( ACTION_STREAM_START_REPLAY, {"path": "sessions/capture.k1mqtt", "speed": 1, "loop": False}, "replay", ), (ACTION_STREAM_STOP, _snapshot_fenced(), "stop"), ( ACTION_VIEWER_SETTINGS_UPDATE, { "point_size": 3, "color_mode": "height", "palette": "viridis", "custom_color": "#102030", "accumulation_seconds": 12, "show_points": True, "show_trajectory": True, "show_grid": False, }, "viewer", ), ], ) def test_facade_preserves_existing_runtime_operations( action_id: str, payload: dict[str, object], expected_call: str, ) -> None: service = FakeXgridsService() dispatcher = DevicePluginDispatcher([_in_process_runtime(XgridsK1PluginFacade(service))]) asyncio.run(dispatcher.invoke(XGRIDS_K1_PLUGIN_ID, action_id, payload)) assert service.calls[0][0] == expected_call def test_sync_runtime_actions_run_outside_the_api_event_loop() -> None: class ThreadAwareService(FakeXgridsService): thread_id: int | None = None def stop(self) -> dict[str, Any]: self.thread_id = threading.get_ident() return super().stop() service = ThreadAwareService() dispatcher = DevicePluginDispatcher([_in_process_runtime(XgridsK1PluginFacade(service))]) event_loop_thread = threading.get_ident() asyncio.run( dispatcher.invoke( XGRIDS_K1_PLUGIN_ID, ACTION_STREAM_STOP, _snapshot_fenced(), ) ) assert service.thread_id is not None assert service.thread_id != event_loop_thread def test_state_reads_are_single_flight_and_survive_one_cancelled_waiter() -> None: class BlockingStateService(FakeXgridsService): def __init__(self) -> None: super().__init__() self.state_calls = 0 self.started = threading.Event() self.release = threading.Event() def state(self) -> dict[str, Any]: self.state_calls += 1 self.started.set() if not self.release.wait(timeout=2): raise AssertionError("state read was not released by the test") return {"phase": "idle", "revision": self.state_calls} service = BlockingStateService() adapter = XgridsK1PluginFacade(service) def invocation(invocation_id: str) -> RuntimeActionInvocation: return RuntimeActionInvocation( invocation_id=invocation_id, plugin_id=adapter.plugin_id, action_id=ACTION_STATE_READ, requested_at=datetime.now(UTC), parameters={}, ) async def exercise() -> None: first = asyncio.create_task(adapter.invoke(invocation("state-read-first"))) while not service.started.is_set(): await asyncio.sleep(0) second = asyncio.create_task(adapter.invoke(invocation("state-read-second"))) await asyncio.sleep(0.02) assert service.state_calls == 1 first.cancel() with pytest.raises(asyncio.CancelledError): await first assert service.state_calls == 1 service.release.set() assert await second == {"phase": "idle", "revision": 1} assert await adapter.invoke(invocation("state-read-fresh")) == { "phase": "idle", "revision": 2, } asyncio.run(exercise()) assert service.state_calls == 2 def test_runtime_requires_successful_handshake_before_dispatch() -> None: adapter = XgridsK1PluginFacade(FakeXgridsService()) runtime = _in_process_runtime(adapter, activate=False) dispatcher = DevicePluginDispatcher([runtime]) assert runtime.health().status == "starting" with pytest.raises(PluginRuntimeUnavailableError, match="not ready"): asyncio.run(dispatcher.invoke(XGRIDS_K1_PLUGIN_ID, ACTION_STREAM_STOP, {})) with pytest.raises(PluginRuntimeCompatibilityError, match="host API"): runtime.handshake( RuntimeHandshakeRequest( handshake_id="handshake-incompatible", plugin_id=XGRIDS_K1_PLUGIN_ID, plugin_version=XGRIDS_K1_PLUGIN_VERSION, host_api_version="missioncore.nodedc/v1alpha1", requested_at=datetime.now(UTC), required_action_ids=tuple(sorted(adapter.action_ids)), ) ) assert runtime.health().status == "starting" def test_runtime_health_transitions_to_stopped_and_close_is_idempotent() -> None: closed: list[bool] = [] runtime = _in_process_runtime( XgridsK1PluginFacade(FakeXgridsService()), close=lambda: closed.append(True), ) assert runtime.health().status == "ready" runtime.close() runtime.close() assert runtime.health().status == "stopped" assert closed == [True] def test_dispatcher_rejects_uncorrelated_transport_result() -> None: class UncorrelatedRuntime(InProcessDevicePluginRuntime): async def invoke( self, invocation: RuntimeActionInvocation, ) -> RuntimeActionResult: result = await super().invoke(invocation) return result.model_copy(update={"invocation_id": "another-invocation"}) adapter = XgridsK1PluginFacade(FakeXgridsService()) descriptor = RuntimePluginDescriptor( plugin_id=adapter.plugin_id, plugin_version=XGRIDS_K1_PLUGIN_VERSION, supported_host_api_versions=("missioncore.nodedc/v1alpha2",), action_ids=tuple(sorted(adapter.action_ids)), ) runtime = UncorrelatedRuntime(adapter, descriptor) runtime.handshake( RuntimeHandshakeRequest( handshake_id="handshake-correlation", plugin_id=adapter.plugin_id, plugin_version=XGRIDS_K1_PLUGIN_VERSION, host_api_version="missioncore.nodedc/v1alpha2", requested_at=datetime.now(UTC), required_action_ids=descriptor.action_ids, ) ) with pytest.raises(PluginExecutionError, match="uncorrelated"): asyncio.run( DevicePluginDispatcher([runtime]).invoke( XGRIDS_K1_PLUGIN_ID, ACTION_STREAM_STOP, _snapshot_fenced(), ) )