139 lines
4.1 KiB
Python
139 lines
4.1 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from k1link.device_plugins.xgrids_k1.protocol.application_bootstrap import (
|
|
ApplicationControlAuthority,
|
|
)
|
|
from k1link.device_plugins.xgrids_k1.protocol.application_execution import (
|
|
ApplicationAuthorityLease,
|
|
ApplicationControlExecutionError,
|
|
DormantApplicationControlCoordinator,
|
|
)
|
|
from k1link.device_plugins.xgrids_k1.protocol.application_publish import (
|
|
WriteDisabledOneShotPublisher,
|
|
)
|
|
|
|
PRIVATE_AUTHORITY = "11111111-2222-3333-4444-555555555555"
|
|
|
|
|
|
class FakeClock:
|
|
def __init__(self) -> None:
|
|
self.now = 100.0
|
|
|
|
def __call__(self) -> float:
|
|
return self.now
|
|
|
|
|
|
class FakeLoader:
|
|
def __init__(self) -> None:
|
|
self.calls = 0
|
|
|
|
def load(self) -> ApplicationControlAuthority:
|
|
self.calls += 1
|
|
return ApplicationControlAuthority(openapi_key=PRIVATE_AUTHORITY)
|
|
|
|
|
|
class ForbiddenSink:
|
|
def publish(
|
|
self,
|
|
*,
|
|
topic: str,
|
|
payload: bytes,
|
|
qos: int,
|
|
retain: bool,
|
|
) -> None:
|
|
del topic, payload, qos, retain
|
|
raise AssertionError("write-disabled coordinator touched its sink")
|
|
|
|
|
|
def test_authority_lease_expires_and_drops_its_cached_reference() -> None:
|
|
clock = FakeClock()
|
|
loader = FakeLoader()
|
|
lease = ApplicationAuthorityLease(loader, ttl_seconds=30.0, monotonic=clock)
|
|
|
|
assert loader.calls == 1
|
|
assert lease.snapshot().state == "armed"
|
|
assert lease.snapshot().authority_cached
|
|
assert PRIVATE_AUTHORITY not in repr(lease.snapshot())
|
|
|
|
clock.now = 130.0
|
|
snapshot = lease.snapshot()
|
|
assert snapshot.state == "expired"
|
|
assert not snapshot.authority_cached
|
|
assert snapshot.remaining_seconds == 0.0
|
|
with pytest.raises(ApplicationControlExecutionError, match="not armed"):
|
|
lease.authority()
|
|
|
|
|
|
@pytest.mark.parametrize("ttl_seconds", [14.999, 300.001, True])
|
|
def test_authority_lease_rejects_unreviewed_ttl(ttl_seconds: object) -> None:
|
|
with pytest.raises(ApplicationControlExecutionError):
|
|
ApplicationAuthorityLease( # type: ignore[arg-type]
|
|
FakeLoader(),
|
|
ttl_seconds=ttl_seconds,
|
|
)
|
|
|
|
|
|
def test_dormant_coordinator_arms_once_and_exposes_no_emission_path() -> None:
|
|
clock = FakeClock()
|
|
loader = FakeLoader()
|
|
coordinator = DormantApplicationControlCoordinator(
|
|
loader,
|
|
WriteDisabledOneShotPublisher(ForbiddenSink()),
|
|
monotonic=clock,
|
|
)
|
|
|
|
armed = coordinator.arm(
|
|
ttl_seconds=60.0,
|
|
epoch_seconds=1_752_680_000,
|
|
timezone_name="Europe/Moscow",
|
|
)
|
|
public = armed.as_dict()
|
|
assert public["state"] == "armed-shadow-only"
|
|
assert public["can_emit_requests"] is False
|
|
assert public["live_transport_installed"] is False
|
|
assert public["publisher"] == {
|
|
"mode": "write-disabled",
|
|
"denied_attempts": 0,
|
|
"transport_calls": 0,
|
|
"vendor_writes_enabled": False,
|
|
"publisher_armed": False,
|
|
"automatic_retry": False,
|
|
}
|
|
assert PRIVATE_AUTHORITY not in repr(armed)
|
|
assert PRIVATE_AUTHORITY not in str(public)
|
|
with pytest.raises(ApplicationControlExecutionError, match="already armed"):
|
|
coordinator.arm(
|
|
ttl_seconds=60.0,
|
|
epoch_seconds=1_752_680_000,
|
|
timezone_name="Europe/Moscow",
|
|
)
|
|
|
|
disarmed = coordinator.disarm().as_dict()
|
|
assert disarmed["state"] == "disarmed"
|
|
assert disarmed["lease"] is None
|
|
assert disarmed["orchestrator"] is None
|
|
assert loader.calls == 1
|
|
|
|
|
|
def test_coordinator_expiry_drops_lease_and_orchestrator_together() -> None:
|
|
clock = FakeClock()
|
|
coordinator = DormantApplicationControlCoordinator(
|
|
FakeLoader(),
|
|
WriteDisabledOneShotPublisher(ForbiddenSink()),
|
|
monotonic=clock,
|
|
)
|
|
coordinator.arm(
|
|
ttl_seconds=15.0,
|
|
epoch_seconds=1_752_680_000,
|
|
timezone_name="Europe/Moscow",
|
|
)
|
|
|
|
clock.now = 115.0
|
|
snapshot = coordinator.snapshot().as_dict()
|
|
assert snapshot["state"] == "expired"
|
|
assert snapshot["lease"] is None
|
|
assert snapshot["orchestrator"] is None
|
|
assert snapshot["can_emit_requests"] is False
|