feat(k1): add local connection matrix
This commit is contained in:
DCCONSTRUCTIONS
@@ -33,6 +33,16 @@ ATTESTATION = CompatibilityAttestationRequest(
|
||||
topology="direct-lan",
|
||||
verification="live-device-info",
|
||||
)
|
||||
QUICK_CONNECT_ATTESTATION = CompatibilityAttestationRequest(
|
||||
firmware_version="3.0.2",
|
||||
topology="device-ap",
|
||||
verification="live-device-info",
|
||||
)
|
||||
DIRECT_CONNECT_ATTESTATION = CompatibilityAttestationRequest(
|
||||
firmware_version="3.0.2",
|
||||
topology="controller-hotspot",
|
||||
verification="live-device-info",
|
||||
)
|
||||
PRIMARY_TEST_CREDENTIAL = "x" * 24
|
||||
SECONDARY_TEST_CREDENTIAL = "y" * 24
|
||||
PROJECT_NAME = "K1 lifecycle test"
|
||||
@@ -225,15 +235,48 @@ def test_project_name_is_normalized_and_control_characters_are_rejected() -> Non
|
||||
host="192.168.1.20",
|
||||
compatibility_attestation=ATTESTATION,
|
||||
)
|
||||
def test_only_physically_accepted_configuration_values_are_admitted() -> None:
|
||||
|
||||
|
||||
def test_connection_modes_require_their_exact_topology_attestation() -> None:
|
||||
assert ConnectRequest(
|
||||
device_id="synthetic-device",
|
||||
ssid="synthetic-network",
|
||||
password=SecretStr(PRIMARY_TEST_CREDENTIAL),
|
||||
connection_mode="quick-connect",
|
||||
compatibility_attestation=QUICK_CONNECT_ATTESTATION,
|
||||
).connection_mode == "quick-connect"
|
||||
assert ConnectRequest(
|
||||
device_id="synthetic-device",
|
||||
ssid="synthetic-network",
|
||||
password=SecretStr(PRIMARY_TEST_CREDENTIAL),
|
||||
connection_mode="direct-connect",
|
||||
compatibility_attestation=DIRECT_CONNECT_ATTESTATION,
|
||||
).connection_mode == "direct-connect"
|
||||
with pytest.raises(ValidationError):
|
||||
ConnectRequest(
|
||||
device_id="synthetic-device",
|
||||
ssid="synthetic-network",
|
||||
password=SecretStr(PRIMARY_TEST_CREDENTIAL),
|
||||
connection_mode="quick-connect", # type: ignore[arg-type]
|
||||
connection_mode="quick-connect",
|
||||
compatibility_attestation=ATTESTATION,
|
||||
)
|
||||
with pytest.raises(ValidationError, match="32 UTF-8 bytes"):
|
||||
ConnectRequest(
|
||||
device_id="synthetic-device",
|
||||
ssid="🛰️" * 9,
|
||||
password=SecretStr(PRIMARY_TEST_CREDENTIAL),
|
||||
compatibility_attestation=ATTESTATION,
|
||||
)
|
||||
with pytest.raises(ValidationError, match="64 UTF-8 bytes"):
|
||||
ConnectRequest(
|
||||
device_id="synthetic-device",
|
||||
ssid="synthetic-network",
|
||||
password=SecretStr("🔒" * 17),
|
||||
compatibility_attestation=ATTESTATION,
|
||||
)
|
||||
|
||||
|
||||
def test_only_physically_accepted_mount_and_gnss_values_are_admitted() -> None:
|
||||
with pytest.raises(ValidationError):
|
||||
PrepareAcquisitionRequest(
|
||||
project_name=PROJECT_NAME,
|
||||
@@ -1372,15 +1415,15 @@ def test_exact_profile_is_inactive_until_selected_for_live_device_info_verificat
|
||||
)
|
||||
|
||||
|
||||
def test_prepare_rejects_device_ap_fallback_as_direct_lan_target(tmp_path: Path) -> None:
|
||||
def test_prepare_rejects_device_ap_without_completed_quick_connect(tmp_path: Path) -> None:
|
||||
service, _ = service_with_fake_runtime(tmp_path)
|
||||
|
||||
with pytest.raises(ValueError, match="точки доступа"):
|
||||
with pytest.raises(ValueError, match="connection flow"):
|
||||
service.prepare_acquisition(
|
||||
PrepareAcquisitionRequest(
|
||||
project_name=PROJECT_NAME,
|
||||
host="192.168.56.1",
|
||||
compatibility_attestation=ATTESTATION,
|
||||
compatibility_attestation=QUICK_CONNECT_ATTESTATION,
|
||||
)
|
||||
)
|
||||
|
||||
@@ -1812,6 +1855,165 @@ def test_network_provisioning_is_single_flight_and_secret_is_unwrapped_only_at_b
|
||||
assert {item["status"] for item in provision_operations} == {"succeeded", "failed"}
|
||||
|
||||
|
||||
def test_quick_connect_associates_the_host_without_a_ble_write(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
service, _ = service_with_fake_runtime(tmp_path)
|
||||
service._devices = [{"device_id": "k1-a"}] # noqa: SLF001
|
||||
association_calls: list[tuple[Path, str, str]] = []
|
||||
|
||||
def fake_associate(
|
||||
helper_path: Path,
|
||||
ssid: str,
|
||||
password: str,
|
||||
**_: object,
|
||||
) -> dict[str, Any]:
|
||||
association_calls.append((helper_path, ssid, password))
|
||||
return {
|
||||
"schema_version": 1,
|
||||
"adapter": "CoreWLAN",
|
||||
"outcome": "associated",
|
||||
"already_associated": False,
|
||||
}
|
||||
|
||||
async def forbidden_ble_write(*_: object, **__: object) -> dict[str, Any]:
|
||||
raise AssertionError("Quick Connect must not provision the K1 over BLE")
|
||||
|
||||
monkeypatch.setattr(facade_module, "associate_with_wifi_once", fake_associate)
|
||||
monkeypatch.setattr(facade_module, "provision_wifi_once", forbidden_ble_write)
|
||||
monkeypatch.setattr(facade_module, "_target_is_local_ipv4", lambda _target: False)
|
||||
|
||||
state = asyncio.run(
|
||||
service.connect(
|
||||
ConnectRequest(
|
||||
device_id="k1-a",
|
||||
ssid="XGR-TEST",
|
||||
password=SecretStr(PRIMARY_TEST_CREDENTIAL),
|
||||
connection_mode="quick-connect",
|
||||
compatibility_attestation=QUICK_CONNECT_ATTESTATION,
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
assert len(association_calls) == 1
|
||||
assert association_calls[0][0].name == "associate_wifi.swift"
|
||||
assert association_calls[0][1:] == ("XGR-TEST", PRIMARY_TEST_CREDENTIAL)
|
||||
assert state["connection_mode"] == "quick-connect"
|
||||
assert state["k1_ip"] == "192.168.56.1"
|
||||
assert state["compatibility"]["attestation"]["topology"] == "device-ap"
|
||||
quick_sessions = sorted(service.evidence_root.glob("*viewer_k1_ap_association*"))
|
||||
assert len(quick_sessions) == 1
|
||||
assert not (quick_sessions[0] / "provisioning.sensitive.json").exists()
|
||||
redacted_manifest = (quick_sessions[0] / "manifest.redacted.json").read_text(
|
||||
encoding="utf-8"
|
||||
)
|
||||
assert PRIMARY_TEST_CREDENTIAL not in redacted_manifest
|
||||
assert "XGR-TEST" not in redacted_manifest
|
||||
assert service._camera_target_for_session( # noqa: SLF001
|
||||
state["device_session"]["device_session_id"]
|
||||
) == "192.168.56.1"
|
||||
|
||||
prepared = service.prepare_acquisition(
|
||||
PrepareAcquisitionRequest(
|
||||
project_name=PROJECT_NAME,
|
||||
compatibility_attestation=QUICK_CONNECT_ATTESTATION,
|
||||
)
|
||||
)
|
||||
assert prepared["acquisition"]["target_host"] == "192.168.56.1"
|
||||
|
||||
|
||||
def test_direct_connect_reuses_the_reviewed_ble_provisioning_frame(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
service, _ = service_with_fake_runtime(tmp_path)
|
||||
service._devices = [{"device_id": "k1-a"}] # noqa: SLF001
|
||||
provisioning_calls: list[tuple[str, str, str]] = []
|
||||
|
||||
async def fake_provision(
|
||||
device_id: str,
|
||||
ssid: str,
|
||||
password: str,
|
||||
**_: object,
|
||||
) -> dict[str, Any]:
|
||||
provisioning_calls.append((device_id, ssid, password))
|
||||
return {
|
||||
"started_at_utc": "2026-07-19T01:00:00Z",
|
||||
"completed_at_utc": "2026-07-19T01:00:01Z",
|
||||
"profile_id": "xgrids-k1-fw3-wifi-v1",
|
||||
"outcome": "lan_address_observed",
|
||||
"observations": [{"status": {"ipv4": "172.20.10.2"}}],
|
||||
}
|
||||
|
||||
def forbidden_host_association(*_: object, **__: object) -> dict[str, Any]:
|
||||
raise AssertionError("Direct Connect must not switch the host Wi-Fi network")
|
||||
|
||||
monkeypatch.setattr(facade_module, "provision_wifi_once", fake_provision)
|
||||
monkeypatch.setattr(
|
||||
facade_module,
|
||||
"associate_with_wifi_once",
|
||||
forbidden_host_association,
|
||||
)
|
||||
monkeypatch.setattr(facade_module, "_target_is_local_ipv4", lambda _target: False)
|
||||
|
||||
state = asyncio.run(
|
||||
service.connect(
|
||||
ConnectRequest(
|
||||
device_id="k1-a",
|
||||
ssid="controller-hotspot",
|
||||
password=SecretStr(PRIMARY_TEST_CREDENTIAL),
|
||||
connection_mode="direct-connect",
|
||||
compatibility_attestation=DIRECT_CONNECT_ATTESTATION,
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
assert provisioning_calls == [
|
||||
("k1-a", "controller-hotspot", PRIMARY_TEST_CREDENTIAL)
|
||||
]
|
||||
assert state["connection_mode"] == "direct-connect"
|
||||
assert state["k1_ip"] == "172.20.10.2"
|
||||
assert state["compatibility"]["attestation"]["topology"] == (
|
||||
"controller-hotspot"
|
||||
)
|
||||
|
||||
|
||||
def test_failed_connection_change_revokes_the_previous_route(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
service, _ = service_with_fake_runtime(tmp_path)
|
||||
service._devices = [{"device_id": "k1-a"}] # noqa: SLF001
|
||||
service._selected_device_id = "previous-k1" # noqa: SLF001
|
||||
service._k1_ip = "192.168.1.20" # noqa: SLF001
|
||||
service._connection_mode = "bridge" # noqa: SLF001
|
||||
|
||||
def failed_association(*_: object, **__: object) -> dict[str, Any]:
|
||||
raise RuntimeError("offline association fixture failed")
|
||||
|
||||
monkeypatch.setattr(facade_module, "associate_with_wifi_once", failed_association)
|
||||
|
||||
with pytest.raises(RuntimeError, match="offline association fixture failed"):
|
||||
asyncio.run(
|
||||
service.connect(
|
||||
ConnectRequest(
|
||||
device_id="k1-a",
|
||||
ssid="XGR-OFFLINE",
|
||||
password=SecretStr(PRIMARY_TEST_CREDENTIAL),
|
||||
connection_mode="quick-connect",
|
||||
compatibility_attestation=QUICK_CONNECT_ATTESTATION,
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
state = service.state()
|
||||
assert state["selected_device_id"] is None
|
||||
assert state["k1_ip"] is None
|
||||
assert state["connection_mode"] is None
|
||||
assert state["compatibility"]["attestation"] is None
|
||||
|
||||
|
||||
def test_network_provisioning_rejects_an_ipv4_owned_by_the_local_host(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
|
||||
Reference in New Issue
Block a user