from __future__ import annotations import json import socket import pytest from bleak.exc import ( BleakBluetoothNotAvailableError, BleakBluetoothNotAvailableReason, ) from k1link.device_plugins.xgrids_k1.host_diagnostics import ( host_diagnostic_for_exception, host_diagnostic_for_reason, host_diagnostics_for_reasons, ) from k1link.device_plugins.xgrids_k1.network_mutation_ledger import ( NetworkMutationLedgerSnapshot, ) @pytest.mark.parametrize( ("reason_code", "expected_code", "expected_domain", "expected_action"), [ ( "ble-permission-denied", "host.bluetooth.permission-denied", "corebluetooth", "grant-bluetooth-permission", ), ( "ble-adapter-powered-off", "host.bluetooth.adapter-powered-off", "corebluetooth", "power-on-bluetooth", ), ( "ble-adapter-unavailable", "host.bluetooth.adapter-unavailable", "corebluetooth", "restore-bluetooth-adapter", ), ( "ble-provisioning-timeout", "host.bluetooth.operation-timeout", "corebluetooth", "explicit-retry", ), ( "corewlan-authorization-denied", "host.wifi.permission-denied", "corewlan", "grant-wifi-permission", ), ( "wifi-interface-inactive", "host.wifi.adapter-powered-off", "corewlan", "power-on-wifi", ), ( "wifi-interface-unavailable", "host.wifi.interface-unavailable", "corewlan", "restore-wifi-interface", ), ( "network-not-found", "host.wifi.ssid-unavailable", "corewlan", "join-expected-network", ), ( "keychain-authorization-required", "host.keychain.interaction-required", "keychain", "unlock-or-authorize-keychain", ), ( "keychain-authorization-denied", "host.keychain.permission-denied", "keychain", "review-keychain-access", ), ( "keychain-access-failed", "host.keychain.unavailable", "keychain", "unlock-or-authorize-keychain", ), ( "host-route-unavailable", "host.route.unavailable", "route", "inspect-host-route", ), ( "tcp-connection-refused", "host.tcp.connection-refused", "tcp", "verify-broker-endpoint", ), ( "tcp-connection-timeout", "host.tcp.connection-timeout", "tcp", "verify-broker-endpoint", ), ( "mqtt_connection_timeout", "host.mqtt.connection-timeout", "mqtt", "verify-broker-endpoint", ), ( "mqtt_network_loop_failed", "host.mqtt.transport-unavailable", "mqtt", "verify-broker-endpoint", ), ( "network-mutation-ledger-corrupt", "host.filesystem.ledger-unavailable", "filesystem", "inspect-local-storage", ), ], ) def test_reviewed_host_reason_codes_map_to_typed_redacted_diagnostics( reason_code: str, expected_code: str, expected_domain: str, expected_action: str, ) -> None: diagnostic = host_diagnostic_for_reason(reason_code) assert diagnostic is not None document = diagnostic.as_dict() assert document["schema_version"] == "missioncore.host-failure-diagnostic/v1" assert document["code"] == expected_code assert document["domain"] == expected_domain assert document["operator_action"] == expected_action assert document["automatic_retry"] is False assert document["redacted"] is True @pytest.mark.parametrize( ("private_message", "expected_code"), [ ( "CoreBluetooth not authorized; private device UUID 1111", "host.bluetooth.permission-denied", ), ( "CBManagerStatePoweredOff for private adapter record", "host.bluetooth.adapter-powered-off", ), ( "No Bluetooth adapter; private host path /Users/operator", "host.bluetooth.adapter-unavailable", ), ], ) def test_corebluetooth_string_classification_never_reflects_private_details( private_message: str, expected_code: str, ) -> None: diagnostic = host_diagnostic_for_exception( RuntimeError(private_message), boundary="corebluetooth", ) assert diagnostic is not None document = diagnostic.as_dict() assert document["code"] == expected_code assert private_message not in json.dumps(document) assert "1111" not in json.dumps(document) assert "/Users/operator" not in json.dumps(document) @pytest.mark.parametrize( ("reason", "expected_code"), [ ( BleakBluetoothNotAvailableReason.DENIED_BY_USER, "host.bluetooth.permission-denied", ), ( BleakBluetoothNotAvailableReason.POWERED_OFF, "host.bluetooth.adapter-powered-off", ), ( BleakBluetoothNotAvailableReason.NO_BLUETOOTH, "host.bluetooth.adapter-unavailable", ), ], ) def test_structured_bleak_adapter_failure_maps_without_message_reflection( reason: BleakBluetoothNotAvailableReason, expected_code: str, ) -> None: private_message = "private CoreBluetooth registry record" diagnostic = host_diagnostic_for_exception( BleakBluetoothNotAvailableError(private_message, reason), boundary="corebluetooth", ) assert diagnostic is not None assert diagnostic.code == expected_code assert private_message not in json.dumps(diagnostic.as_dict()) def test_builtin_host_errors_require_an_explicit_boundary() -> None: assert host_diagnostic_for_exception(PermissionError("private")) is None assert host_diagnostic_for_exception(TimeoutError("private")) is None assert host_diagnostic_for_exception(ConnectionRefusedError("private")) is None filesystem = host_diagnostic_for_exception( PermissionError("private ledger path"), boundary="filesystem", ) tcp = host_diagnostic_for_exception( TimeoutError("private broker address"), boundary="tcp", ) mqtt = host_diagnostic_for_exception( ConnectionRefusedError("private broker address"), boundary="mqtt", ) assert filesystem is not None assert filesystem.code == "host.filesystem.permission-denied" assert tcp is not None assert tcp.code == "host.tcp.connection-timeout" assert mqtt is not None assert mqtt.code == "host.mqtt.connection-refused" def test_unknown_reason_is_not_reflected_and_duplicate_diagnostics_are_collapsed() -> None: private_reason = "private-ssid-or-ledger-path" assert host_diagnostic_for_reason(private_reason) is None diagnostics = host_diagnostics_for_reasons( "host-route-unavailable", "host-path-unavailable", private_reason, ) assert [diagnostic.code for diagnostic in diagnostics] == ["host.route.unavailable"] assert private_reason not in json.dumps([item.as_dict() for item in diagnostics]) def test_operation_error_projects_nested_diagnostic_without_raw_exception_text() -> None: from k1link.device_plugins.xgrids_k1.facade import _operation_error private_message = "CoreBluetooth not authorized for UUID private-device-17" error = _operation_error( RuntimeError(private_message), category="transport", side_effect_status="none", host_boundary="corebluetooth", ) assert error["code"] == "RuntimeError" assert error["host_diagnostic"] == { "schema_version": "missioncore.host-failure-diagnostic/v1", "code": "host.bluetooth.permission-denied", "domain": "corebluetooth", "impact": "discovery", "operator_action": "grant-bluetooth-permission", "automatic_retry": False, "redacted": True, } assert private_message not in json.dumps(error) assert "private-device-17" not in json.dumps(error) def test_control_session_failure_projects_keychain_diagnostic_without_message_reflection() -> None: from k1link.device_plugins.xgrids_k1.facade import ( _application_control_session_public_snapshot, ) private_message = "private Keychain item label and account" document = _application_control_session_public_snapshot( { "state": "failed", "failure": { "reason_code": "keychain-authorization-required", "message": private_message, }, } ) failure = document["failure"] assert isinstance(failure, dict) assert failure["host_diagnostic"] == { "schema_version": "missioncore.host-failure-diagnostic/v1", "code": "host.keychain.interaction-required", "domain": "keychain", "impact": "control", "operator_action": "unlock-or-authorize-keychain", "automatic_retry": False, "redacted": True, } assert private_message not in json.dumps(failure["host_diagnostic"]) def test_corrupt_ledger_projection_exposes_only_typed_storage_diagnostic() -> None: from k1link.device_plugins.xgrids_k1.facade import ( _network_mutation_ledger_public_snapshot, ) document = _network_mutation_ledger_public_snapshot( NetworkMutationLedgerSnapshot( status="corrupt", record=None, reason_code="network-mutation-ledger-corrupt", ) ) assert document["mutation_allowed"] is False assert document["diagnostic"] == { "schema_version": "missioncore.host-failure-diagnostic/v1", "code": "host.filesystem.ledger-unavailable", "domain": "filesystem", "impact": "durable-safety", "operator_action": "inspect-local-storage", "automatic_retry": False, "redacted": True, } @pytest.mark.parametrize( ("socket_error", "expected_reason", "expected_code"), [ ( ConnectionRefusedError("private broker address refused"), "tcp-connection-refused", "host.tcp.connection-refused", ), ( TimeoutError("private broker address timed out"), "tcp-connection-timeout", "host.tcp.connection-timeout", ), ], ) def test_broker_tcp_probe_preserves_only_refused_or_timeout_class( monkeypatch: pytest.MonkeyPatch, socket_error: OSError, expected_reason: str, expected_code: str, ) -> None: from k1link.device_plugins.xgrids_k1.facade import _probe_control_endpoint_socket def fail_connect(*_args: object, **_kwargs: object) -> socket.socket: raise socket_error monkeypatch.setattr(socket, "create_connection", fail_connect) result = _probe_control_endpoint_socket("192.168.68.52") diagnostic = host_diagnostic_for_reason(result.reason_code) assert result.reachable is False assert result.reason_code == expected_reason assert diagnostic is not None assert diagnostic.code == expected_code assert "private broker" not in json.dumps(diagnostic.as_dict())