332 lines
10 KiB
Python
332 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from k1link.host_network import wifi
|
|
|
|
TEST_PASSWORD = "fixture-only-network-secret"
|
|
TEST_PROFILE_ID = "fixture.quick-connect.v1"
|
|
TEST_CREDENTIAL_SOURCE_ID = "fixture.firmware-provider.v1"
|
|
|
|
|
|
def _helper(tmp_path: Path) -> Path:
|
|
helper = tmp_path / "associate_wifi.swift"
|
|
helper.write_text("// offline fixture\n", encoding="utf-8")
|
|
return helper
|
|
|
|
|
|
def test_association_exposes_only_profile_id_and_expected_ssid_to_platform_helper(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
monkeypatch.setattr(wifi.sys, "platform", "darwin")
|
|
calls: list[dict[str, object]] = []
|
|
live_inputs: list[bytearray] = []
|
|
|
|
def fake_runner(argv: list[str], **kwargs: object) -> subprocess.CompletedProcess[bytes]:
|
|
assert isinstance(kwargs["input"], bytearray)
|
|
live_inputs.append(kwargs["input"])
|
|
calls.append({"argv": argv, **kwargs, "input": bytes(kwargs["input"])})
|
|
return subprocess.CompletedProcess(
|
|
argv,
|
|
0,
|
|
stdout=(
|
|
b'{"ok":true,"adapter":"CoreWLAN","already_associated":false,'
|
|
b'"profile_enrolled":true,"scan_attempt_count":3,'
|
|
b'"scan_elapsed_ms":1840,"credential_source":"system-wifi-keychain"}'
|
|
),
|
|
stderr=b"",
|
|
)
|
|
|
|
result = wifi.associate_with_wifi_profile_once(
|
|
_helper(tmp_path),
|
|
TEST_PROFILE_ID,
|
|
"XGR-OFFLINE",
|
|
runner=fake_runner,
|
|
)
|
|
|
|
assert result == {
|
|
"schema_version": 1,
|
|
"adapter": "CoreWLAN",
|
|
"outcome": "associated",
|
|
"already_associated": False,
|
|
"profile_enrolled": True,
|
|
"scan_attempt_count": 3,
|
|
"scan_elapsed_ms": 1840,
|
|
"credential_source": "system-wifi-keychain",
|
|
}
|
|
assert len(calls) == 1
|
|
call = calls[0]
|
|
assert call["argv"][:2] == ["/usr/bin/xcrun", "swift"]
|
|
request = json.loads(bytes(call["input"]).decode("utf-8"))
|
|
assert request == {
|
|
"action": "associate",
|
|
"profile_id": TEST_PROFILE_ID,
|
|
"ssid": "XGR-OFFLINE",
|
|
"scan_timeout_seconds": 15.0,
|
|
}
|
|
assert call["check"] is False
|
|
assert call["timeout"] == 180.0
|
|
assert live_inputs and not any(live_inputs[0])
|
|
|
|
|
|
def test_profile_store_passes_secret_only_through_stdin(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
monkeypatch.setattr(wifi.sys, "platform", "darwin")
|
|
captured: dict[str, object] = {}
|
|
live_inputs: list[bytearray] = []
|
|
|
|
def fake_runner(argv: list[str], **kwargs: object) -> subprocess.CompletedProcess[bytes]:
|
|
assert isinstance(kwargs["input"], bytearray)
|
|
live_inputs.append(kwargs["input"])
|
|
captured.update({"argv": argv, **kwargs, "input": bytes(kwargs["input"])})
|
|
return subprocess.CompletedProcess(
|
|
argv,
|
|
0,
|
|
stdout=b'{"ok":true,"adapter":"macOS Keychain","stored":true}',
|
|
stderr=b"",
|
|
)
|
|
|
|
result = wifi.store_wifi_profile_once(
|
|
_helper(tmp_path),
|
|
TEST_PROFILE_ID,
|
|
"XGR-OFFLINE",
|
|
TEST_PASSWORD,
|
|
runner=fake_runner,
|
|
)
|
|
|
|
assert result["outcome"] == "stored"
|
|
assert TEST_PASSWORD not in " ".join(captured["argv"])
|
|
request = json.loads(bytes(captured["input"]).decode("utf-8"))
|
|
assert request == {
|
|
"action": "store-profile",
|
|
"profile_id": TEST_PROFILE_ID,
|
|
"ssid": "XGR-OFFLINE",
|
|
"password": TEST_PASSWORD,
|
|
}
|
|
assert live_inputs and not any(live_inputs[0])
|
|
|
|
|
|
def test_firmware_material_store_passes_secret_only_through_stdin(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
monkeypatch.setattr(wifi.sys, "platform", "darwin")
|
|
captured: dict[str, object] = {}
|
|
live_inputs: list[bytearray] = []
|
|
|
|
def fake_runner(argv: list[str], **kwargs: object) -> subprocess.CompletedProcess[bytes]:
|
|
assert isinstance(kwargs["input"], bytearray)
|
|
live_inputs.append(kwargs["input"])
|
|
captured.update({"argv": argv, **kwargs, "input": bytes(kwargs["input"])})
|
|
return subprocess.CompletedProcess(
|
|
argv,
|
|
0,
|
|
stdout=b'{"ok":true,"adapter":"macOS Keychain","stored":true}',
|
|
stderr=b"",
|
|
)
|
|
|
|
result = wifi.store_wifi_credential_material(
|
|
_helper(tmp_path),
|
|
TEST_CREDENTIAL_SOURCE_ID,
|
|
TEST_PASSWORD,
|
|
runner=fake_runner,
|
|
)
|
|
|
|
assert result["outcome"] == "stored"
|
|
assert TEST_PASSWORD not in " ".join(captured["argv"])
|
|
request = json.loads(bytes(captured["input"]).decode("utf-8"))
|
|
assert request == {
|
|
"action": "store-credential-material",
|
|
"profile_id": TEST_CREDENTIAL_SOURCE_ID,
|
|
"password": TEST_PASSWORD,
|
|
}
|
|
assert live_inputs and not any(live_inputs[0])
|
|
|
|
|
|
def test_profile_is_materialized_from_opaque_firmware_source(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
monkeypatch.setattr(wifi.sys, "platform", "darwin")
|
|
captured: dict[str, object] = {}
|
|
|
|
def fake_runner(argv: list[str], **kwargs: object) -> subprocess.CompletedProcess[bytes]:
|
|
captured.update({"argv": argv, **kwargs, "input": bytes(kwargs["input"])})
|
|
return subprocess.CompletedProcess(
|
|
argv,
|
|
0,
|
|
stdout=(
|
|
b'{"ok":true,"adapter":"macOS Keychain","profile_available":true,'
|
|
b'"profile_enrolled":true,"credential_source":"exact-firmware-profile"}'
|
|
),
|
|
stderr=b"",
|
|
)
|
|
|
|
result = wifi.ensure_wifi_profile_from_credential_source(
|
|
_helper(tmp_path),
|
|
TEST_PROFILE_ID,
|
|
"XGR-OFFLINE",
|
|
TEST_CREDENTIAL_SOURCE_ID,
|
|
runner=fake_runner,
|
|
)
|
|
|
|
assert result == {
|
|
"schema_version": 1,
|
|
"adapter": "macOS Keychain",
|
|
"available": True,
|
|
"profile_enrolled": True,
|
|
"credential_source": "exact-firmware-profile",
|
|
}
|
|
request = json.loads(bytes(captured["input"]).decode("utf-8"))
|
|
assert request == {
|
|
"action": "ensure-profile",
|
|
"profile_id": TEST_PROFILE_ID,
|
|
"ssid": "XGR-OFFLINE",
|
|
"credential_source_id": TEST_CREDENTIAL_SOURCE_ID,
|
|
}
|
|
assert TEST_PASSWORD not in bytes(captured["input"]).decode("utf-8")
|
|
|
|
|
|
@pytest.mark.parametrize("available", [True, False])
|
|
def test_profile_preflight_checks_only_the_expected_keychain_item(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
available: bool,
|
|
) -> None:
|
|
monkeypatch.setattr(wifi.sys, "platform", "darwin")
|
|
captured: dict[str, object] = {}
|
|
|
|
def fake_runner(argv: list[str], **kwargs: object) -> subprocess.CompletedProcess[bytes]:
|
|
captured.update({"argv": argv, **kwargs, "input": bytes(kwargs["input"])})
|
|
encoded_available = b"true" if available else b"false"
|
|
return subprocess.CompletedProcess(
|
|
argv,
|
|
0,
|
|
stdout=(
|
|
b'{"ok":true,"adapter":"macOS Keychain","profile_available":'
|
|
+ encoded_available
|
|
+ b"}"
|
|
),
|
|
stderr=b"",
|
|
)
|
|
|
|
result = wifi.check_wifi_profile(
|
|
_helper(tmp_path),
|
|
TEST_PROFILE_ID,
|
|
"XGR-OFFLINE",
|
|
runner=fake_runner,
|
|
)
|
|
|
|
assert result == {
|
|
"schema_version": 1,
|
|
"adapter": "macOS Keychain",
|
|
"available": available,
|
|
}
|
|
request = json.loads(bytes(captured["input"]).decode("utf-8"))
|
|
assert request == {
|
|
"action": "check-profile",
|
|
"profile_id": TEST_PROFILE_ID,
|
|
"ssid": "XGR-OFFLINE",
|
|
}
|
|
|
|
|
|
def test_association_reports_only_sanitized_helper_reason(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
monkeypatch.setattr(wifi.sys, "platform", "darwin")
|
|
|
|
def fake_runner(argv: list[str], **_: object) -> subprocess.CompletedProcess[bytes]:
|
|
return subprocess.CompletedProcess(
|
|
argv,
|
|
1,
|
|
stdout=(
|
|
b'{"ok":false,"reason_code":"network-not-found",'
|
|
b'"scan_attempt_count":4,"scan_elapsed_ms":15021}'
|
|
),
|
|
stderr=f"private diagnostic {TEST_PASSWORD}".encode(),
|
|
)
|
|
|
|
with pytest.raises(wifi.HostWifiProfileError, match="network-not-found") as raised:
|
|
wifi.associate_with_wifi_profile_once(
|
|
_helper(tmp_path),
|
|
TEST_PROFILE_ID,
|
|
"XGR-OFFLINE",
|
|
runner=fake_runner,
|
|
)
|
|
|
|
assert TEST_PASSWORD not in str(raised.value)
|
|
assert raised.value.scan_attempt_count == 4
|
|
assert raised.value.scan_elapsed_ms == 15021
|
|
|
|
|
|
def test_association_reports_operator_timeout_separately_from_missing_helper(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
monkeypatch.setattr(wifi.sys, "platform", "darwin")
|
|
|
|
def timed_out_runner(
|
|
argv: list[str], **_: object
|
|
) -> subprocess.CompletedProcess[bytes]:
|
|
raise subprocess.TimeoutExpired(argv, timeout=180.0)
|
|
|
|
with pytest.raises(
|
|
wifi.HostWifiProfileError,
|
|
match="host-wifi-operation-timeout",
|
|
) as raised:
|
|
wifi.associate_with_wifi_profile_once(
|
|
_helper(tmp_path),
|
|
TEST_PROFILE_ID,
|
|
"XGR-OFFLINE",
|
|
runner=timed_out_runner,
|
|
)
|
|
|
|
assert raised.value.reason_code == "host-wifi-operation-timeout"
|
|
|
|
|
|
def test_association_reports_an_unavailable_helper_separately_from_timeout(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
monkeypatch.setattr(wifi.sys, "platform", "darwin")
|
|
|
|
def unavailable_runner(
|
|
_: list[str], **__: object
|
|
) -> subprocess.CompletedProcess[bytes]:
|
|
raise OSError("offline fixture")
|
|
|
|
with pytest.raises(
|
|
wifi.HostWifiProfileError,
|
|
match="host-wifi-helper-unavailable",
|
|
) as raised:
|
|
wifi.associate_with_wifi_profile_once(
|
|
_helper(tmp_path),
|
|
TEST_PROFILE_ID,
|
|
"XGR-OFFLINE",
|
|
runner=unavailable_runner,
|
|
)
|
|
|
|
assert raised.value.reason_code == "host-wifi-helper-unavailable"
|
|
|
|
|
|
def test_association_rejects_an_uninstalled_platform(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr(wifi.sys, "platform", "linux")
|
|
|
|
with pytest.raises(wifi.HostWifiProfileError, match="unsupported-platform"):
|
|
wifi.associate_with_wifi_profile_once(
|
|
_helper(tmp_path),
|
|
TEST_PROFILE_ID,
|
|
"XGR-OFFLINE",
|
|
)
|