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

Защита Bridge и камеры K1 при переподключении

Browse Source
This commit is contained in:
DCCONSTRUCTIONS
2026-08-23 01:25:25 +03:00
parent 0752e5c6bf
commit 71c8b043c9
8 changed files with 737 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_xgrids_acquisition_lifecycle.py
+309 -63
View File
@@ -14120,6 +14120,7 @@ def test_confirmed_scanning_waits_for_first_authoritative_pcl_before_camera(
assert pre_prepare_fence(lambda: True) is True
events.append(("select", (source_id, target)))
camera_state["active_source_id"] = source_id
camera_state["generation"] = 1
events.append(("record", session_dir))
camera_state["recording"] = {
"active": True,
@@ -14313,9 +14314,23 @@ def test_confirmed_scanning_waits_for_first_authoritative_pcl_before_camera(
assert any(
stream["source_id"] == "sensor.camera.right"
and stream["activation"]["selected"] is True
and stream["activation"]["controllable"] is True
and stream["activation"]["controllable"] is False
for stream in admitted_camera_streams
)
admitted_generation = admitted_state["camera_preview"]["generation"]
assert isinstance(admitted_generation, int)
with pytest.raises(facade_module.LocalAcquisitionLifecycleError) as stop_error:
service.stop_camera_preview(
facade_module.CameraPreviewStopRequest(
device_session_id=service._device_session_id, # type: ignore[arg-type] # noqa: SLF001
generation=admitted_generation,
)
)
assert stop_error.value.reason_code == "acquisition-camera-evidence-owned"
after_rejected_stop = service.camera_preview.snapshot()
assert after_rejected_stop["active_source_id"] == "sensor.camera.right"
assert after_rejected_stop["recording"]["active"] is True
assert after_rejected_stop["generation"] == admitted_generation
# Every later authoritative PCL for the same lineage is idempotent.
physical_snapshot_calls_before = physical_snapshot_calls
@@ -17570,7 +17585,7 @@ def test_abort_waits_for_start_handoff_then_stops_owned_producers(
assert service._acquisition_session_lease is None # noqa: SLF001
def test_camera_selection_finishes_before_serialized_acquisition_stop(
def test_active_acquisition_camera_selection_is_rejected_before_serialized_stop(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
@@ -17591,43 +17606,8 @@ def test_camera_selection_finishes_before_serialized_acquisition_stop(
service._device_session_id = device_session_id # noqa: SLF001
out_dir = service._acquisition_out_dir # noqa: SLF001
assert out_dir is not None
entered_camera_arm = threading.Event()
release_camera_arm = threading.Event()
stop_finished = threading.Event()
worker_errors: list[BaseException] = []
events: list[str] = []
def blocked_camera_arm(_out_dir: Path, *, require_session: bool = False) -> None:
assert require_session is True
events.append("arm")
entered_camera_arm.set()
if not release_camera_arm.wait(timeout=2):
raise TimeoutError("test did not release camera arm")
def select_worker() -> None:
try:
service.select_camera_preview(
CameraPreviewSelectRequest(
source_id="sensor.camera.left",
device_session_id=device_session_id,
)
)
except BaseException as exc: # pragma: no cover - asserted below
worker_errors.append(exc)
def stop_worker() -> None:
try:
service.stop_acquisition(
_stop_request(
acquisition_id=acquisition_id,
mode="capture-only",
)
)
except BaseException as exc: # pragma: no cover - asserted below
worker_errors.append(exc)
finally:
stop_finished.set()
camera_snapshot = service.camera_preview.snapshot()
camera_snapshot.update(
{
@@ -17647,7 +17627,11 @@ def test_camera_selection_finishes_before_serialized_acquisition_stop(
},
}
)
monkeypatch.setattr(service, "_arm_camera_recording", blocked_camera_arm)
monkeypatch.setattr(
service,
"_arm_camera_recording",
lambda *_args, **_kwargs: events.append("unexpected-arm"),
)
monkeypatch.setattr(
service.camera_preview,
"snapshot",
@@ -17656,13 +17640,7 @@ def test_camera_selection_finishes_before_serialized_acquisition_stop(
monkeypatch.setattr(
service.camera_preview,
"select",
lambda source_id, _target: (
events.append("select")
or {
"active_source_id": source_id,
"recording": {"active": False},
}
),
lambda *_args, **_kwargs: events.append("unexpected-select"),
)
monkeypatch.setattr(
service.camera_preview,
@@ -17670,18 +17648,24 @@ def test_camera_selection_finishes_before_serialized_acquisition_stop(
lambda **_kwargs: events.append("camera-stop"),
)
monkeypatch.setattr(runtime, "stop", lambda: events.append("runtime-stop"))
select_thread = threading.Thread(target=select_worker)
stop_thread = threading.Thread(target=stop_worker)
select_thread.start()
assert entered_camera_arm.wait(timeout=2)
stop_thread.start()
assert not stop_finished.wait(timeout=0.05)
release_camera_arm.set()
select_thread.join(timeout=2)
stop_thread.join(timeout=2)
assert worker_errors == []
assert events == ["arm", "select", "camera-stop", "runtime-stop"]
with pytest.raises(facade_module.LocalAcquisitionLifecycleError) as selection_error:
service.select_camera_preview(
CameraPreviewSelectRequest(
source_id="sensor.camera.left",
device_session_id=device_session_id,
)
)
assert selection_error.value.reason_code == "acquisition-camera-evidence-owned"
assert events == []
service.stop_acquisition(
_stop_request(
acquisition_id=acquisition_id,
mode="capture-only",
)
)
assert events == ["camera-stop", "runtime-stop"]
assert service.state()["acquisition"]["state"] == "completed"
assert service._acquisition_session_lease is None # noqa: SLF001
@@ -21646,6 +21630,8 @@ def test_read_only_reconciliation_requires_exact_requested_mode_despite_shared_s
service,
intended_mode=intended_mode,
)
if requested_mode != "bridge":
_select_connection_mode(service, requested_mode)
async def read_current_status(*_: object, **__: object) -> dict[str, Any]:
raise AssertionError("mode mismatch must fail before a GATT read")
@@ -21820,6 +21806,13 @@ def test_durable_restart_rejects_request_target_mismatch_before_gatt(
transport_ref=DURABLE_K1_UUID,
)
restarted, _ = service_with_fake_runtime(tmp_path)
requested_mode = {
"direct-lan": "bridge",
"device-ap": "quick-connect",
"controller-hotspot": "direct-connect",
}[attestation.topology]
if requested_mode != "bridge":
_select_connection_mode(restarted, requested_mode)
reads: list[str] = []
async def forbidden_read(*_: object, **__: object) -> dict[str, Any]:
@@ -21989,6 +21982,8 @@ def test_durable_restart_previous_topology_cannot_resolve_ambiguous_write(
previous_connection=previous,
)
restarted, _ = service_with_fake_runtime(tmp_path)
if intended_mode != "bridge":
_select_connection_mode(restarted, intended_mode)
captured = _durable_status_capture()
reads = 0
pin_calls: list[tuple[object, str]] = []
@@ -22123,13 +22118,15 @@ def test_durable_restart_failed_status_keeps_ambiguity_and_does_not_pin(
def _seed_durable_quick_reconnect_evidence(
service: XgridsK1CompatibilityService,
*,
transport_ref: str = DURABLE_K1_UUID,
) -> None:
"""Persist the exact secret-free Quick topology and AP activation proof."""
topology_store = service._semantic_topology_store # noqa: SLF001
assert topology_store is not None
topology_store.commit(
transport_ref=DURABLE_K1_UUID,
transport_ref=transport_ref,
connection_mode="quick-connect",
ipv4="192.168.56.1",
compatibility_profile_id=XGRIDS_K1_COMPATIBILITY_PROFILE_ID,
@@ -22145,7 +22142,7 @@ def _seed_durable_quick_reconnect_evidence(
json.dumps(
{
"operation": "single_reviewed_quick_connect_ap_activation",
"device_macos_uuid": DURABLE_K1_UUID,
"device_macos_uuid": transport_ref,
"device_name": "XGR-A46BE7",
"outcome": "ap_ready_observed",
"ready_observed": True,
@@ -22220,6 +22217,7 @@ def test_semantic_quick_restart_restores_saved_host_profile_before_endpoint_prob
)
restarted, _ = service_with_fake_runtime(tmp_path)
_select_connection_mode(restarted, "quick-connect")
calls: list[str] = []
association_calls: list[tuple[Path, str, str, float, float]] = []
route_samples: list[HostPathProbeResult] = []
@@ -22367,6 +22365,7 @@ def test_semantic_quick_restart_skips_wifi_mutation_when_direct_route_already_ex
first, _ = service_with_fake_runtime(tmp_path)
_seed_durable_quick_reconnect_evidence(first)
restarted, _ = service_with_fake_runtime(tmp_path)
_select_connection_mode(restarted, "quick-connect")
wifi_mutations: list[str] = []
tcp_calls: list[str] = []
@@ -22444,6 +22443,7 @@ def test_semantic_quick_restart_rejects_non_direct_route_without_tcp_or_ble(
first, _ = service_with_fake_runtime(tmp_path)
_seed_durable_quick_reconnect_evidence(first)
restarted, _ = service_with_fake_runtime(tmp_path)
_select_connection_mode(restarted, "quick-connect")
device_edges: list[str] = []
tcp_calls: list[str] = []
@@ -22528,6 +22528,7 @@ def test_semantic_quick_restart_tcp_timeout_fails_without_ble_fallback(
first, _ = service_with_fake_runtime(tmp_path)
_seed_durable_quick_reconnect_evidence(first)
restarted, _ = service_with_fake_runtime(tmp_path)
_select_connection_mode(restarted, "quick-connect")
device_edges: list[str] = []
tcp_calls: list[str] = []
@@ -22606,6 +22607,7 @@ def test_semantic_quick_restart_rejects_route_change_during_single_tcp_probe(
first, _ = service_with_fake_runtime(tmp_path)
_seed_durable_quick_reconnect_evidence(first)
restarted, _ = service_with_fake_runtime(tmp_path)
_select_connection_mode(restarted, "quick-connect")
device_edges: list[str] = []
tcp_calls: list[str] = []
initial_path = _direct_host_path("192.168.56.1")
@@ -23014,6 +23016,240 @@ def test_semantic_durable_verify_rejects_association_change_during_tcp_probe(
assert state["semantic_topology_store"]["record"]["revision"] == 1
def test_quick_verify_is_rejected_while_bridge_is_selected_before_any_io(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
first, _ = service_with_fake_runtime(tmp_path)
_seed_durable_quick_reconnect_evidence(first)
restarted, _ = service_with_fake_runtime(tmp_path)
edges: list[str] = []
def forbidden_edge(*_: object, **__: object) -> object:
edges.append("io")
raise AssertionError("mode mismatch must fail before BLE, Wi-Fi or endpoint I/O")
async def forbidden_async_edge(*_: object, **__: object) -> dict[str, Any]:
forbidden_edge()
return {}
monkeypatch.setattr(facade_module, "read_wifi_status_once", forbidden_async_edge)
monkeypatch.setattr(facade_module, "provision_wifi_once", forbidden_async_edge)
monkeypatch.setattr(
facade_module,
"associate_with_wifi_profile_once",
forbidden_edge,
)
monkeypatch.setattr(facade_module, "_inspect_host_path", forbidden_edge)
before = restarted.state()
assert before["desired_connection_mode"] == "bridge"
with pytest.raises(facade_module.ConnectionVerificationError) as raised:
asyncio.run(
restarted.verify_connection(
ConnectionVerifyRequest(
device_id=DURABLE_K1_UUID,
source="durable-configured-state",
compatibility_attestation=QUICK_CONNECT_ATTESTATION,
expected_mode_revision=before[
"desired_connection_mode_revision"
],
)
)
)
assert raised.value.reason_code == "connection-mode-draft-mismatch"
assert edges == []
after = restarted.state()
assert after["desired_connection_mode"] == "bridge"
assert after["operations"] == []
def test_physical_recovery_cannot_silently_change_bridge_draft_to_quick(
tmp_path: Path,
) -> None:
original, _ = service_with_fake_runtime(tmp_path)
_persist_resolved_unclassified_stop_for_restart(
original,
connection_mode="quick-connect",
target_ipv4=facade_module.AP_FALLBACK_IPV4,
)
_seed_durable_quick_reconnect_evidence(
original,
transport_ref="test-ble-transport",
)
restarted, _ = service_with_fake_runtime(tmp_path)
before = restarted.state()
assert before["desired_connection_mode"] == "bridge"
assert before["connection_policy"]["actions"][
"observe-configured-device-network"
]["required_connection_mode"] == "quick-connect"
with pytest.raises(facade_module.NetworkProvisioningConflict) as raised:
restarted.select_connection_mode(
DesiredConnectionModeRequest(
connection_mode="quick-connect",
expected_revision=before["desired_connection_mode_revision"],
)
)
assert raised.value.reason_code == (
"connection-mode-selection-physical-recovery-reset-required"
)
after = restarted.state()
assert after["desired_connection_mode"] == "bridge"
assert after["desired_connection_mode_revision"] == before[
"desired_connection_mode_revision"
]
def test_public_quick_physical_recovery_rejoins_saved_wifi_after_live_ble_proof(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
original, _ = service_with_fake_runtime(tmp_path)
_persist_resolved_unclassified_stop_for_restart(
original,
connection_mode="quick-connect",
target_ipv4=facade_module.AP_FALLBACK_IPV4,
)
_seed_durable_quick_reconnect_evidence(
original,
transport_ref="test-ble-transport",
)
restarted, runtime = service_with_fake_runtime(tmp_path)
# Model a process whose explicit Quick draft was already authoritative
# before the unresolved physical record became visible. The public UI may
# never create this authority as a side effect of the recovery button.
with restarted._lock: # noqa: SLF001
restarted._desired_connection_mode = "quick-connect" # noqa: SLF001
restarted._desired_connection_mode_revision = 1 # noqa: SLF001
restarted._host_wifi_association_probe = FakeHostWifiAssociationProbe( # type: ignore[assignment] # noqa: SLF001
"a" * 64
)
capture = _durable_status_capture(device_id="test-ble-transport")
ordered_edges: list[str] = []
status_reads: list[tuple[str, bool, bool, float | None]] = []
route_samples: list[HostPathProbeResult] = []
network_writes: list[str] = []
async def read_current_quick_status(
device_id: str,
*,
allow_known_device_retrieval: bool = False,
rediscover: bool = False,
exact_scan_timeout_seconds: float | None = None,
on_gatt_validated: Callable[[object], None] | None = None,
**_: object,
) -> dict[str, Any]:
ordered_edges.append("ble-status-read")
status_reads.append(
(
device_id,
allow_known_device_retrieval,
rediscover,
exact_scan_timeout_seconds,
)
)
assert on_gatt_validated is not None
on_gatt_validated(capture)
result = _wifi_status_read(None, device_id=device_id)
result["status"] = _ap_ready_wifi_status()
return result
def associate_saved_profile(*_: object, **__: object) -> dict[str, Any]:
ordered_edges.append("host-wifi-association")
return _successful_saved_quick_profile_association()
def settling_quick_route(target: str) -> HostPathProbeResult:
path = (
_tunnel_host_path(target)
if len(route_samples) < 2
else _direct_host_path(target)
)
route_samples.append(path)
return path
def reachable_tcp(target: str) -> facade_module.TcpReachabilityProbeResult:
assert target == facade_module.AP_FALLBACK_IPV4
ordered_edges.append("mqtt-endpoint-probe")
return facade_module.TcpReachabilityProbeResult(reachable=True)
async def forbidden_network_write(*_: object, **__: object) -> dict[str, Any]:
network_writes.append("network-write")
raise AssertionError("saved Quick physical recovery must not write K1 network state")
monkeypatch.setattr(facade_module, "read_wifi_status_once", read_current_quick_status)
monkeypatch.setattr(
facade_module,
"associate_with_wifi_profile_once",
associate_saved_profile,
)
monkeypatch.setattr(facade_module, "provision_wifi_once", forbidden_network_write)
monkeypatch.setattr(facade_module, "pin_connected_device_handle", lambda *_a, **_k: None)
monkeypatch.setattr(facade_module, "_target_is_local_ipv4", lambda _target: False)
monkeypatch.setattr(facade_module, "_inspect_host_path", settling_quick_route)
monkeypatch.setattr(facade_module, "_probe_control_endpoint_socket", reachable_tcp)
monkeypatch.setattr(
facade_module,
"DURABLE_QUICK_CONNECT_ROUTE_SETTLE_INTERVAL_SECONDS",
0.001,
)
_install_real_coordinator_bootstrap(
restarted,
observed_session_state="ready",
)
before = restarted.state()
recovery = before["connection_policy"]["actions"][
"observe-configured-device-network"
]
assert recovery["allowed"] is True
assert recovery["required_connection_mode"] == "quick-connect"
assert recovery["requires_live_gatt_validation"] is True
verified = asyncio.run(restarted.verify_connection(ConnectionVerifyRequest()))
assert ordered_edges == [
"ble-status-read",
"host-wifi-association",
"mqtt-endpoint-probe",
]
assert status_reads == [
(
"test-ble-transport",
True,
True,
facade_module.CONNECTION_VERIFY_EXACT_UUID_SCAN_TIMEOUT_SECONDS,
)
]
assert network_writes == []
assert runtime.start_calls == []
assert runtime.stop_calls == 0
assert verified["active_connection_mode"] == "quick-connect"
assert verified["k1_ip"] == facade_module.AP_FALLBACK_IPV4
assert verified["last_operation"]["status"] == "succeeded"
assert verified["last_operation"]["result"]["physical_reconciliation"][
"performed"
] is True
reassociation_sessions = sorted(
restarted.evidence_root.glob("*viewer_k1_quick_connect_host_reassociation*")
)
assert len(reassociation_sessions) == 1
proof = json.loads(
(reassociation_sessions[0] / "host-network-proof.redacted.json").read_text(
encoding="utf-8"
)
)
assert proof["outcome"] == "direct-endpoint-reachable"
assert proof["route_after"]["route_class"] == "direct"
assert proof["tcp_probe_performed"] is True
assert proof["ble_operation_performed"] is True
assert proof["device_write_performed"] is False
assert proof["automatic_retry"] is False
@pytest.mark.parametrize("observed_session_state", ["ready", "scanning"])
def test_public_physical_recovery_refreshes_stale_dhcp_then_classifies_without_writes(
monkeypatch: pytest.MonkeyPatch,
@@ -27491,9 +27727,9 @@ def test_select_device_handoff_is_local_cancel_invalidates_candidates_and_stale_
expected_reconfiguration_revision=fresh["connection_reconfiguration"]["revision"],
expected_reconfiguration_intent_id=fresh["connection_reconfiguration"]["intent_id"],
)
with pytest.raises(facade_module.NetworkProvisioningConflict) as quick_isolated:
with pytest.raises(facade_module.ConnectionVerificationError) as quick_isolated:
asyncio.run(service.verify_connection(quick_verify_request))
assert quick_isolated.value.reason_code == "connection-reconfiguration-target-mismatch"
assert quick_isolated.value.reason_code == "connection-mode-draft-mismatch"
cancel_request = _reconfiguration_request(fresh, "cancel")
cancelled = asyncio.run(service.prepare_connection_reconfiguration(cancel_request))
@@ -30869,6 +31105,9 @@ def _persist_ambiguous_start_with_prepared_checkpoint(
def _persist_resolved_active_start_for_restart(
service: XgridsK1CompatibilityService,
*,
connection_mode: facade_module.ConnectionMode = "bridge",
target_ipv4: str = "192.168.68.52",
) -> None:
"""Persist one successful START without retaining process-local acquisition."""
@@ -30879,8 +31118,8 @@ def _persist_resolved_active_start_for_restart(
connection = PhysicalCommandConnectionBinding(
intent_id="old-start-intent",
transport_ref="test-ble-transport",
connection_mode="bridge",
target_ipv4="192.168.68.52",
connection_mode=connection_mode,
target_ipv4=target_ipv4,
target_port=facade_module.CONTROL_MQTT_PORT,
host_path_epoch=1,
control_session_id="old-start-control",
@@ -31019,10 +31258,17 @@ def _persist_same_runtime_prepared_checkpoint_for_resolved_start(
def _persist_resolved_unclassified_stop_for_restart(
service: XgridsK1CompatibilityService,
*,
connection_mode: facade_module.ConnectionMode = "bridge",
target_ipv4: str = "192.168.68.52",
) -> tuple[str, int]:
"""Persist the exact legacy startup shape without process-local owners."""
_persist_resolved_active_start_for_restart(service)
_persist_resolved_active_start_for_restart(
service,
connection_mode=connection_mode,
target_ipv4=target_ipv4,
)
ledger = service._physical_command_ledger # noqa: SLF001
start = ledger.snapshot().record
assert start is not None