Исключение ложного переподключения к выведенному K1

This commit is contained in:
DCCONSTRUCTIONS
2026-08-23 14:12:57 +03:00
parent 6e7428a116
commit 03830cd15e
2 changed files with 129 additions and 2 deletions
@@ -33782,6 +33782,15 @@ def _connection_policy_projection(
durable_observation_reasons.append("resolved-apply-durable-target-mismatch")
if not durable_observation_ref or durable_observation_mode is None:
durable_observation_reasons.append("durable-recovery-target-unavailable")
if transport_is_retired(durable_observation_ref):
# A resolved Apply plus matching semantic topology is normally enough
# to offer the saved LAN/Quick reconnect. An operator retirement is a
# newer physical boundary, though: verify_connection rejects that same
# UUID before creating an operation or touching the host network. Do
# not publish a button whose exact server request is guaranteed to be
# rejected. The explicit fresh Scan/reopen flow remains the only path
# that can re-admit this physical target.
durable_observation_reasons.append("physical-command-target-retired")
if (
transport_is_eligible_fresh(durable_observation_ref)
and resolved_apply_durable_target is None
+120 -2
View File
@@ -28,6 +28,8 @@ from k1link.device_plugins.xgrids_k1.active_acquisition_recovery_checkpoint impo
from k1link.device_plugins.xgrids_k1.facade import (
BleScanRequest,
CompatibilityAttestationRequest,
ConnectionVerificationError,
ConnectionVerifyRequest,
ConnectRequest,
DesiredConnectionModeRequest,
NetworkProvisioningConflict,
@@ -84,13 +86,16 @@ def _seed_resettable_physical_record(
*,
closed_cycles: int = 0,
prepared_only: bool = False,
connection_mode: facade_module.ConnectionMode = "bridge",
) -> PhysicalCommandRecord:
ledger = service._physical_command_ledger # noqa: SLF001
connection = PhysicalCommandConnectionBinding(
intent_id="reset-preflight-intent-0001",
transport_ref="RESET-PREFLIGHT-K1-UUID",
connection_mode="bridge",
target_ipv4="192.168.68.51",
connection_mode=connection_mode,
target_ipv4=(
"192.168.56.1" if connection_mode == "quick-connect" else "192.168.68.51"
),
target_port=1883,
host_path_epoch=7,
control_session_id="reset-preflight-control-session-0001",
@@ -590,6 +595,119 @@ def test_scenario_reset_preserves_physical_audit_and_unrelated_plugin_state(
assert unrelated_state.stat().st_ino == unrelated_inode
def test_reset_retired_saved_quick_target_is_not_advertised_as_reconnectable(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
service = _service(monkeypatch, tmp_path)
transport_ref = "RESET-PREFLIGHT-K1-UUID"
ledger = service._network_mutation_ledger # noqa: SLF001
prepared = ledger.prepare(
operation_id="reset-saved-quick-resolved-apply",
transport_ref=transport_ref,
intended_mode="quick-connect",
write_mode="with_response",
baseline_status=NetworkStatusEvidence(
mode="WIFI_CLIENT",
ipv4="192.168.68.51",
status_code=1,
reserved=0,
),
)
dispatching = ledger.mark_dispatching(
prepared.operation_id,
expected_revision=prepared.revision,
)
observing = ledger.mark_observing(
dispatching.operation_id,
expected_revision=dispatching.revision,
write_confirmed=True,
observation=NetworkStatusEvidence(
mode="WIFI_AP",
ipv4="192.168.56.1",
status_code=1,
reserved=0,
),
)
ledger.resolve(
observing.operation_id,
expected_revision=observing.revision,
resolution="target-observed",
)
topology_store = service._semantic_topology_store # noqa: SLF001
assert topology_store is not None
topology_store.commit(
transport_ref=transport_ref,
connection_mode="quick-connect",
ipv4="192.168.56.1",
compatibility_profile_id=facade_module.XGRIDS_K1_COMPATIBILITY_PROFILE_ID,
firmware_version="3.0.2",
source="ble-post-write-status",
observed_at_utc="2026-08-23T10:32:31.583Z",
)
_seed_resettable_physical_record(
service,
connection_mode="quick-connect",
)
reset = service.select_connection_mode(
_reset(
mode="quick-connect",
revision=0,
reset_id="op-reset-retired-saved-quick-policy-01",
)
)
decision = reset["connection_policy"]["actions"][
"observe-configured-device-network"
]
assert decision["required_transport_ref"] == transport_ref
assert decision["required_connection_mode"] == "quick-connect"
assert decision["allowed"] is False
assert "physical-command-target-retired" in decision["reason_codes"]
assert (
"observe-configured-device-network"
not in reset["connection_policy"]["allowed_actions"]
)
assert reset["connection_policy"]["recommended_action"] == "scan-ble"
assert reset["connection_policy"]["actions"]["scan-ble"]["allowed"] is True
io_calls: list[str] = []
def forbidden_io(*_args: object, **_kwargs: object) -> object:
io_calls.append("called")
raise AssertionError("retired saved Quick target reached host or device I/O")
monkeypatch.setattr(
facade_module,
"associate_with_wifi_profile_once",
forbidden_io,
)
monkeypatch.setattr(facade_module, "_inspect_host_path", forbidden_io)
monkeypatch.setattr(facade_module, "_probe_control_endpoint_socket", forbidden_io)
operations_before = service._operations.snapshot(limit=128) # noqa: SLF001
with pytest.raises(ConnectionVerificationError) as denied:
asyncio.run(
service.verify_connection(
ConnectionVerifyRequest(
device_id=transport_ref,
source="durable-configured-state",
compatibility_attestation=CompatibilityAttestationRequest(
firmware_version="3.0.2",
topology="device-ap",
verification="live-device-info",
),
expected_mode_revision=1,
)
)
)
assert denied.value.reason_code == "physical-command-target-retired"
assert service._operations.snapshot(limit=128) == operations_before # noqa: SLF001
assert io_calls == []
def test_existing_reset_retires_failed_reconnect_reopen_for_manual_new_flow(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,