SILVER/NODEDC_MISSION_CORE
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

fix(k1): stabilize repeated acquisition and live viewer recovery

Browse Source
This commit is contained in:
DCCONSTRUCTIONS
2026-07-29 01:55:12 +03:00
parent 37b8930527
commit 2a97cf28c0
28 changed files with 2776 additions and 286 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_xgrids_acquisition_lifecycle.py
+485 -50
View File
@@ -247,6 +247,8 @@ def test_verify_connection_refreshes_dynamic_dhcp_address_and_rotates_session(
assert state["device_session"]["device_session_id"] != "old-device-session"
assert state["connection_verification"] == {
"status": "live-address-observed",
"lease_state": "configured",
"lease_generation": 1,
"endpoint_validation": "ble-wifi-status-read",
"network_reachability": "not-probed",
"address_changed": True,
@@ -283,7 +285,7 @@ def test_implicit_acquisition_target_uses_current_ble_dhcp_address(
assert state["acquisition"]["target_host"] == "10.255.254.77"
def test_control_session_opens_against_current_ble_dhcp_address(
def test_control_session_reuses_reachable_process_owned_connection_lease(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
@@ -291,6 +293,8 @@ def test_control_session_opens_against_current_ble_dhcp_address(
service._selected_device_id = "test-ble-transport" # noqa: SLF001
service._connection_mode = "bridge" # noqa: SLF001
service._k1_ip = "10.255.254.54" # noqa: SLF001
service._device_id = "known-k1" # noqa: SLF001
service._device_session_id = "known-session" # noqa: SLF001
service._compatibility_attestation = {"profile": "exact"} # noqa: SLF001
opened_hosts: list[str] = []
@@ -302,12 +306,12 @@ def test_control_session_opens_against_current_ble_dhcp_address(
opened_hosts.append(host)
return self.snapshot()
async def fake_status_read(*_: object, **__: object) -> dict[str, Any]:
return _wifi_status_read("10.255.254.77")
async def forbidden_status_read(*_: object, **__: object) -> dict[str, Any]:
raise AssertionError("reachable connection lease must not reopen BLE")
service._application_control_session = FakeOpenControlSession() # type: ignore[assignment] # noqa: SLF001
monkeypatch.setattr(facade_module, "read_wifi_status_once", fake_status_read)
monkeypatch.setattr(facade_module, "_target_is_local_ipv4", lambda _target: False)
monkeypatch.setattr(facade_module, "read_wifi_status_once", forbidden_status_read)
monkeypatch.setattr(facade_module, "_control_endpoint_reachable", lambda _target: True)
state = service.open_application_control_session(
OpenApplicationControlSessionRequest(
@@ -320,8 +324,241 @@ def test_control_session_opens_against_current_ble_dhcp_address(
)
)
assert opened_hosts == ["10.255.254.54"]
assert state["k1_ip"] == "10.255.254.54"
assert state["connection_verification"]["lease_state"] == "reachable"
assert state["connection_verification"]["network_reachability"] == "reachable"
open_operation = next(
operation
for operation in state["operations"]
if operation["action"] == "application-control.session.open"
)
assert open_operation["status"] == "succeeded"
assert open_operation["result"] == {
"lease_generation": 0,
"connection_lease_reused": True,
"recovery_performed": False,
"address_changed": False,
"device_write_performed": False,
}
def test_reachable_connection_lease_supports_repeated_independent_control_sessions(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
service, _ = service_with_fake_runtime(tmp_path)
service._selected_device_id = "test-ble-transport" # noqa: SLF001
service._connection_mode = "bridge" # noqa: SLF001
service._k1_ip = "10.255.254.54" # noqa: SLF001
service._device_id = "known-k1" # noqa: SLF001
service._device_session_id = "known-session" # noqa: SLF001
service._connection_lease_generation = 7 # noqa: SLF001
service._compatibility_attestation = {"profile": "exact"} # noqa: SLF001
opened_hosts: list[str] = []
class FakeCompletedControlSession:
def snapshot(self) -> dict[str, object]:
return {
"state": "completed",
"can_open": True,
"can_confirm_standby": False,
}
def open(self, *, host: str, **_: object) -> dict[str, object]:
opened_hosts.append(host)
return self.snapshot()
async def forbidden_status_read(*_: object, **__: object) -> dict[str, Any]:
raise AssertionError("repeated scans must not repeat BLE or Wi-Fi setup")
service._application_control_session = FakeCompletedControlSession() # type: ignore[assignment] # noqa: SLF001
monkeypatch.setattr(facade_module, "read_wifi_status_once", forbidden_status_read)
monkeypatch.setattr(facade_module, "_control_endpoint_reachable", lambda _target: True)
request = OpenApplicationControlSessionRequest(
operator_present=True,
owner_controlled_device=True,
lixelgo_closed=True,
battery_storage_confirmed=True,
expected_physical_state_confirmed=True,
timezone_name="Europe/Moscow",
)
service.open_application_control_session(request)
state = service.open_application_control_session(request)
assert opened_hosts == ["10.255.254.54", "10.255.254.54"]
assert state["device_session"]["device_session_id"] == "known-session"
assert state["connection_verification"]["lease_generation"] == 7
open_operations = [
operation
for operation in state["operations"]
if operation["action"] == "application-control.session.open"
]
assert len(open_operations) == 2
assert {operation["status"] for operation in open_operations} == {"succeeded"}
assert all(
operation["result"]["connection_lease_reused"] is True for operation in open_operations
)
def test_control_session_recovers_changed_bridge_address_without_wifi_write(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
service, _ = service_with_fake_runtime(tmp_path)
service._selected_device_id = "test-ble-transport" # noqa: SLF001
service._connection_mode = "bridge" # noqa: SLF001
service._k1_ip = "10.255.254.54" # noqa: SLF001
service._device_id = "known-k1" # noqa: SLF001
service._device_session_id = "old-session" # noqa: SLF001
service._compatibility_attestation = {"profile": "exact"} # noqa: SLF001
opened_hosts: list[str] = []
status_calls: list[dict[str, object]] = []
class FakeOpenControlSession:
def snapshot(self) -> dict[str, object]:
return {"state": "idle", "can_confirm_standby": False}
def open(self, *, host: str, **_: object) -> dict[str, object]:
opened_hosts.append(host)
return self.snapshot()
async def fake_status_read(*_: object, **kwargs: object) -> dict[str, Any]:
status_calls.append(kwargs)
return _wifi_status_read("10.255.254.77")
service._application_control_session = FakeOpenControlSession() # type: ignore[assignment] # noqa: SLF001
monkeypatch.setattr(facade_module, "read_wifi_status_once", fake_status_read)
monkeypatch.setattr(facade_module, "_target_is_local_ipv4", lambda _target: False)
monkeypatch.setattr(facade_module, "_host_route_class", lambda _target: "direct-or-routed")
monkeypatch.setattr(
facade_module,
"_control_endpoint_reachable",
lambda target: target == "10.255.254.77",
)
state = service.open_application_control_session(
OpenApplicationControlSessionRequest(
operator_present=True,
owner_controlled_device=True,
lixelgo_closed=True,
battery_storage_confirmed=True,
expected_physical_state_confirmed=True,
timezone_name="Europe/Moscow",
)
)
assert status_calls == [{"timeout_seconds": 20.0, "rediscover": True}]
assert opened_hosts == ["10.255.254.77"]
assert state["k1_ip"] == "10.255.254.77"
assert state["device_session"]["device_session_id"] != "old-session"
assert state["connection_verification"]["status"] == "recovered"
assert state["connection_verification"]["write_performed"] is False
open_operation = next(
operation
for operation in state["operations"]
if operation["action"] == "application-control.session.open"
)
assert open_operation["result"]["recovery_performed"] is True
assert open_operation["result"]["address_changed"] is True
assert open_operation["result"]["device_write_performed"] is False
def test_control_session_prestart_failure_is_journaled_and_marks_lease_offline(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
service, _ = service_with_fake_runtime(tmp_path)
service._selected_device_id = "test-ble-transport" # noqa: SLF001
service._connection_mode = "bridge" # noqa: SLF001
service._k1_ip = "10.255.254.54" # noqa: SLF001
service._device_id = "known-k1" # noqa: SLF001
service._device_session_id = "known-session" # noqa: SLF001
service._compatibility_attestation = {"profile": "exact"} # noqa: SLF001
async def failed_status_read(*_: object, **__: object) -> dict[str, Any]:
raise TimeoutError("synthetic BLE recovery timeout")
monkeypatch.setattr(facade_module, "read_wifi_status_once", failed_status_read)
monkeypatch.setattr(facade_module, "_control_endpoint_reachable", lambda _target: False)
monkeypatch.setattr(facade_module, "_host_route_class", lambda _target: "direct-or-routed")
with pytest.raises(
facade_module.ConnectionLeaseUnavailable,
match="повторное чтение состояния",
):
service.open_application_control_session(
OpenApplicationControlSessionRequest(
operator_present=True,
owner_controlled_device=True,
lixelgo_closed=True,
battery_storage_confirmed=True,
expected_physical_state_confirmed=True,
timezone_name="Europe/Moscow",
)
)
state = service.state()
assert state["application_control_session"]["state"] == "idle"
assert state["device_session"]["connectivity"] == "offline"
assert state["connection_verification"]["lease_state"] == "disconnected"
open_operation = next(
operation
for operation in state["operations"]
if operation["action"] == "application-control.session.open"
)
assert open_operation["status"] == "failed"
assert open_operation["error"] == {
"category": "connection",
"code": "connection_lease_ble_recovery_failed",
"retryable": False,
"safe_to_retry": True,
"side_effect_status": "none",
}
def test_bridge_route_mismatch_stops_before_ble_recovery_and_vendor_commands(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
service, _ = service_with_fake_runtime(tmp_path)
service._selected_device_id = "test-ble-transport" # noqa: SLF001
service._connection_mode = "bridge" # noqa: SLF001
service._k1_ip = "192.168.68.50" # noqa: SLF001
service._device_id = "known-k1" # noqa: SLF001
service._device_session_id = "known-session" # noqa: SLF001
service._compatibility_attestation = {"profile": "exact"} # noqa: SLF001
async def forbidden_status_read(*_: object, **__: object) -> dict[str, Any]:
raise AssertionError("host route mismatch must stop before BLE recovery")
monkeypatch.setattr(facade_module, "_control_endpoint_reachable", lambda _target: False)
monkeypatch.setattr(facade_module, "_host_route_class", lambda _target: "tunnel")
monkeypatch.setattr(facade_module, "read_wifi_status_once", forbidden_status_read)
with pytest.raises(
facade_module.ConnectionLeaseUnavailable,
match="другой локальной сети",
):
service.open_application_control_session(
OpenApplicationControlSessionRequest(
operator_present=True,
owner_controlled_device=True,
lixelgo_closed=True,
battery_storage_confirmed=True,
expected_physical_state_confirmed=True,
timezone_name="Europe/Moscow",
)
)
state = service.state()
assert state["connection_verification"]["reason_code"] == (
"connection_lease_host_route_mismatch"
)
assert state["connection_verification"]["endpoint_validation"] == "host-route"
assert state["connection_verification"]["host_route_class"] == "tunnel"
assert state["application_control_session"]["state"] == "idle"
def test_prepare_creates_provisional_device_session_and_profiled_acquisition(
@@ -384,18 +621,24 @@ def test_acquisition_is_unbounded_by_default_and_accepts_ten_hour_hint() -> None
def test_connection_modes_require_their_exact_topology_attestation() -> None:
assert ConnectRequest(
device_id="synthetic-device",
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"
assert (
ConnectRequest(
device_id="synthetic-device",
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",
@@ -628,6 +871,91 @@ def test_plugin_commanded_acquisition_keeps_start_and_stop_as_explicit_actions(
assert completed["acquisition"]["result"]["device_state"] == "ready"
assert completed["last_operation"]["status"] == "succeeded"
assert runtime.stop_calls == 1
assert completed["live_perception_shadow"]["active"] is False
def test_next_scan_retires_stale_terminal_live_perception_ingress_before_start(
tmp_path: Path,
) -> None:
service, runtime = service_with_fake_runtime(tmp_path)
service.live_perception_ingress.begin_session("stale-completed-session")
prepared = service.prepare_acquisition(
PrepareAcquisitionRequest(
project_name=PROJECT_NAME,
host="192.168.1.20",
compatibility_attestation=ATTESTATION,
)
)
state = service.start_acquisition(
StartAcquisitionRequest(acquisition_id=prepared["acquisition"]["acquisition_id"])
)
assert runtime.start_calls
assert state["acquisition"]["state"] == "starting"
assert state["live_perception_shadow"]["active"] is True
assert state["live_perception_shadow"]["session_id"] != "stale-completed-session"
def test_confirmed_scanning_activates_and_records_right_camera(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
service, runtime = service_with_fake_runtime(tmp_path)
control = FakeInteractiveControlSession()
service._application_control_session = control # type: ignore[assignment] # noqa: SLF001
service._k1_ip = "192.168.1.20" # noqa: SLF001
service._compatibility_attestation = {"profile": "exact"} # noqa: SLF001
prepared = service.prepare_acquisition(
PrepareAcquisitionRequest(
project_name="TEST001",
host="192.168.1.20",
compatibility_attestation=ATTESTATION,
)
)
acquisition_id = prepared["acquisition"]["acquisition_id"]
service.start_acquisition(
StartAcquisitionRequest(
acquisition_id=acquisition_id,
physical_acceptance=PHYSICAL_ACCEPTANCE,
)
)
out_dir = service._acquisition_out_dir # noqa: SLF001
assert out_dir is not None
out_dir.mkdir(parents=True)
events: list[tuple[str, object]] = []
camera_state: dict[str, object] = {
"phase": "idle",
"active_source_id": None,
"recording": {"active": False},
}
def select_camera(source_id: str, target: str) -> dict[str, object]:
events.append(("select", (source_id, target)))
camera_state["phase"] = "selected"
camera_state["active_source_id"] = source_id
return dict(camera_state)
def start_recording(session_dir: Path) -> dict[str, object]:
events.append(("record", session_dir))
camera_state["recording"] = {"active": True}
camera_state["phase"] = "connecting"
return dict(camera_state)
monkeypatch.setattr(service.camera_preview, "snapshot", lambda: dict(camera_state))
monkeypatch.setattr(service.camera_preview, "select", select_camera)
monkeypatch.setattr(service.camera_preview, "start_recording", start_recording)
service._activate_default_acquisition_camera() # noqa: SLF001
assert events == [
("select", ("sensor.camera.right", "192.168.1.20")),
("record", out_dir),
]
assert camera_state["active_source_id"] == "sensor.camera.right"
assert camera_state["recording"] == {"active": True}
assert runtime.source_mode == "live"
def test_device_standby_retires_sources_after_terminal_local_stop_failure(
@@ -1611,6 +1939,9 @@ def test_runtime_error_terminalizes_pending_graceful_stop(tmp_path: Path) -> Non
assert failed["acquisition"]["state"] == "failed"
assert stop_operation["status"] == "failed"
assert stop_operation["error"]["side_effect_status"] == "unknown"
assert failed["source_mode"] == "idle"
assert failed["phase"] != "error"
assert runtime.stop_calls == 1
def test_receiver_completion_terminalizes_unconfirmed_graceful_stop(
@@ -1970,6 +2301,11 @@ def test_network_provisioning_is_single_flight_and_secret_is_unwrapped_only_at_b
}
monkeypatch.setattr(facade_module, "provision_wifi_once", fake_provision)
monkeypatch.setattr(
facade_module,
"_host_route_class",
lambda _target: "direct-or-routed",
)
first = asyncio.create_task(
service.connect(
ConnectRequest(
@@ -2005,6 +2341,117 @@ 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_bridge_network_change_retires_terminal_acquisition_and_receiver_error(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
service, runtime = service_with_fake_runtime(tmp_path)
service._devices = [{"device_id": "k1-a", "name": "XGR-TEST-A"}] # noqa: SLF001
prepared = service.prepare_acquisition(
PrepareAcquisitionRequest(
project_name=PROJECT_NAME,
host="192.168.1.20",
compatibility_attestation=ATTESTATION,
)
)
service.abort_acquisition(
AbortAcquisitionRequest(
acquisition_id=prepared["acquisition"]["acquisition_id"],
)
)
runtime.phase = "error"
runtime.source_mode = "live"
stop_calls_before_connect = runtime.stop_calls
async def fake_provision(*_: object, **__: object) -> dict[str, Any]:
return {
"started_at_utc": "2026-07-28T18:11:37Z",
"completed_at_utc": "2026-07-28T18:11:44Z",
"profile_id": "xgrids-k1-fw3-wifi-v1",
"outcome": "lan_address_observed",
"observations": [{"status": {"ipv4": "192.168.1.20"}}],
}
monkeypatch.setattr(facade_module, "provision_wifi_once", fake_provision)
monkeypatch.setattr(
facade_module,
"_host_route_class",
lambda _target: "direct-or-routed",
)
connected = asyncio.run(
service.connect(
ConnectRequest(
device_id="k1-a",
ssid="lab-network",
password=SecretStr(PRIMARY_TEST_CREDENTIAL),
connection_mode="bridge",
compatibility_attestation=ATTESTATION,
)
)
)
assert connected["connection_mode"] == "bridge"
assert connected["k1_ip"] == "192.168.1.20"
assert connected["acquisition"] is None
assert connected["source_mode"] == "idle"
assert connected["phase"] == "connected"
assert runtime.stop_calls == stop_calls_before_connect + 1
def test_bridge_provisioning_reports_host_route_mismatch_without_hiding_device_success(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
service, _ = service_with_fake_runtime(tmp_path)
service._devices = [{"device_id": "k1-a", "name": "XGR-TEST-A"}] # noqa: SLF001
async def fake_provision(*_: object, **__: object) -> dict[str, Any]:
return {
"started_at_utc": "2026-07-28T19:43:03Z",
"completed_at_utc": "2026-07-28T19:43:16Z",
"profile_id": "xgrids-k1-fw3-wifi-v1",
"outcome": "lan_address_observed",
"observations": [{"status": {"ipv4": "192.168.68.50"}}],
}
monkeypatch.setattr(facade_module, "provision_wifi_once", fake_provision)
monkeypatch.setattr(facade_module, "_host_route_class", lambda _target: "tunnel")
connected = asyncio.run(
service.connect(
ConnectRequest(
device_id="k1-a",
ssid="lab-router",
password=SecretStr(PRIMARY_TEST_CREDENTIAL),
connection_mode="bridge",
compatibility_attestation=ATTESTATION,
)
)
)
assert connected["k1_ip"] == "192.168.68.50"
assert connected["phase"] == "device_selected"
assert "компьютер подключён к другой сети" in connected["message"]
assert connected["device_session"]["connectivity"] == "offline"
assert connected["connection_verification"] == {
"status": "host-route-mismatch",
"lease_state": "disconnected",
"lease_generation": 1,
"endpoint_validation": "host-route",
"network_reachability": "unreachable",
"reason_code": "connection_lease_host_route_mismatch",
"host_route_class": "tunnel",
"write_performed": True,
"observed_at": connected["connection_verification"]["observed_at"],
}
operation = next(
item for item in connected["operations"] if item["action"] == "network.provision"
)
assert operation["status"] == "succeeded"
assert operation["result"]["host_route_ready"] is False
assert operation["result"]["host_route_class"] == "tunnel"
def test_quick_connect_activates_the_device_ap_then_associates_the_host(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
@@ -2028,9 +2475,7 @@ def test_quick_connect_activates_the_device_ap_then_associates_the_host(
)
@asynccontextmanager
async def fake_activation_session(
device_id: str, **_: object
) -> AsyncIterator[dict[str, Any]]:
async def fake_activation_session(device_id: str, **_: object) -> AsyncIterator[dict[str, Any]]:
nonlocal ble_session_open
activation_calls.append(device_id)
ble_session_open = True
@@ -2092,9 +2537,7 @@ def test_quick_connect_activates_the_device_ap_then_associates_the_host(
assert not ble_session_open
assert len(association_calls) == 1
assert association_calls[0][0].name == "associate_wifi.swift"
assert association_calls[0][1] == facade_module.quick_connect_host_profile_id(
"XGR-TEST-A"
)
assert association_calls[0][1] == facade_module.quick_connect_host_profile_id("XGR-TEST-A")
assert association_calls[0][2] == "XGR-TEST-A"
assert state["connection_mode"] == "quick-connect"
assert state["k1_ip"] == "192.168.56.1"
@@ -2102,9 +2545,7 @@ def test_quick_connect_activates_the_device_ap_then_associates_the_host(
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"
)
redacted_manifest = (quick_sessions[0] / "manifest.redacted.json").read_text(encoding="utf-8")
assert PRIMARY_TEST_CREDENTIAL not in redacted_manifest
assert "host_wifi_profile_id" in redacted_manifest
assert '"host_wifi_profile_ready_before_device_write": true' in redacted_manifest
@@ -2112,9 +2553,12 @@ def test_quick_connect_activates_the_device_ap_then_associates_the_host(
assert "credential_provider_id" in redacted_manifest
assert "device_ap_activation_profile_id" in redacted_manifest
assert (quick_sessions[0] / "ap-activation.redacted.json").exists()
assert service._camera_target_for_session( # noqa: SLF001
state["device_session"]["device_session_id"]
) == "192.168.56.1"
assert (
service._camera_target_for_session( # noqa: SLF001
state["device_session"]["device_session_id"]
)
== "192.168.56.1"
)
prepared = service.prepare_acquisition(
PrepareAcquisitionRequest(
@@ -2158,6 +2602,11 @@ def test_direct_connect_reuses_the_reviewed_ble_provisioning_frame(
forbidden_host_association,
)
monkeypatch.setattr(facade_module, "_target_is_local_ipv4", lambda _target: False)
monkeypatch.setattr(
facade_module,
"_host_route_class",
lambda _target: "direct-or-routed",
)
state = asyncio.run(
service.connect(
@@ -2171,14 +2620,10 @@ def test_direct_connect_reuses_the_reviewed_ble_provisioning_frame(
)
)
assert provisioning_calls == [
("k1-a", "controller-hotspot", PRIMARY_TEST_CREDENTIAL)
]
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"
)
assert state["compatibility"]["attestation"]["topology"] == ("controller-hotspot")
def test_quick_connect_missing_credential_provider_stops_before_ap_write(
@@ -2232,9 +2677,7 @@ def test_quick_connect_missing_credential_provider_stops_before_ap_write(
assert state["k1_ip"] == "192.168.1.20"
assert state["connection_mode"] == "bridge"
assert not list(service.evidence_root.glob("*viewer_k1_ap_association*"))
operation = next(
item for item in state["operations"] if item["action"] == "network.provision"
)
operation = next(item for item in state["operations"] if item["action"] == "network.provision")
assert operation["status"] == "failed"
assert operation["error"]["side_effect_status"] == "none"
assert operation["error"]["safe_to_retry"] is True
@@ -2259,9 +2702,7 @@ def test_quick_connect_does_not_start_host_wifi_without_ap_ready_flag(
)
@asynccontextmanager
async def not_ready_session(
*_: object, **__: object
) -> AsyncIterator[dict[str, Any]]:
async def not_ready_session(*_: object, **__: object) -> AsyncIterator[dict[str, Any]]:
yield {
"profile_id": "xgrids-k1-fw3-quick-connect-ap-v1",
"started_at_utc": "2026-07-19T15:00:00Z",
@@ -2325,9 +2766,7 @@ def test_failed_connection_change_revokes_the_previous_route(
)
@asynccontextmanager
async def fake_activation_session(
*_: object, **__: object
) -> AsyncIterator[dict[str, Any]]:
async def fake_activation_session(*_: object, **__: object) -> AsyncIterator[dict[str, Any]]:
yield {
"profile_id": "xgrids-k1-fw3-quick-connect-ap-v1",
"started_at_utc": "2026-07-19T15:00:00Z",
@@ -2370,9 +2809,7 @@ def test_failed_connection_change_revokes_the_previous_route(
assert state["compatibility"]["attestation"] is None
quick_sessions = sorted(service.evidence_root.glob("*viewer_k1_ap_association*"))
failure_evidence = json.loads(
(quick_sessions[0] / "host-wifi-association.redacted.json").read_text(
encoding="utf-8"
)
(quick_sessions[0] / "host-wifi-association.redacted.json").read_text(encoding="utf-8")
)
assert failure_evidence["reason_code"] == "network-not-found"
assert failure_evidence["scan_attempt_count"] == 4
@@ -2413,9 +2850,7 @@ def test_network_provisioning_rejects_an_ipv4_owned_by_the_local_host(
state = service.state()
assert state["selected_device_id"] is None
assert state["k1_ip"] is None
operation = next(
item for item in state["operations"] if item["action"] == "network.provision"
)
operation = next(item for item in state["operations"] if item["action"] == "network.provision")
assert operation["status"] == "failed"
assert operation["error"]["safe_to_retry"] is False
assert operation["error"]["side_effect_status"] == "unknown"