NODEDC_MISSION_CORE/tests/test_xgrids_host_diagnostics.py

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())