89 lines
2.5 KiB
Python
89 lines
2.5 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from k1link.device_plugins.xgrids_k1.protocol.application_bootstrap import (
|
|
ApplicationControlAuthority,
|
|
LiveDeviceControlBinding,
|
|
build_shadow_bootstrap,
|
|
)
|
|
from k1link.device_plugins.xgrids_k1.protocol.application_publish import (
|
|
OneShotPublishEnvelope,
|
|
VendorWritesDisabledError,
|
|
WriteDisabledOneShotPublisher,
|
|
)
|
|
|
|
PRIVATE_AUTHORITY = "11111111-2222-3333-4444-555555555555"
|
|
|
|
|
|
class SpySink:
|
|
def __init__(self) -> None:
|
|
self.calls = 0
|
|
|
|
def publish(
|
|
self,
|
|
*,
|
|
topic: str,
|
|
payload: bytes,
|
|
qos: int,
|
|
retain: bool,
|
|
) -> None:
|
|
del topic, payload, qos, retain
|
|
self.calls += 1
|
|
|
|
|
|
def _envelope() -> OneShotPublishEnvelope:
|
|
authority = ApplicationControlAuthority(openapi_key=PRIVATE_AUTHORITY)
|
|
binding = LiveDeviceControlBinding(
|
|
vendor_device_id="aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee",
|
|
device_serial="K1SERIAL01",
|
|
software_version="V3.0.2-build.1",
|
|
system_version="V3.0.2",
|
|
device_model="LixelKity K1",
|
|
device_type="K1",
|
|
is_activated=True,
|
|
)
|
|
request = build_shadow_bootstrap(
|
|
authority,
|
|
binding,
|
|
epoch_seconds=1_752_680_000,
|
|
timezone_name="Europe/Moscow",
|
|
).requests[0]
|
|
return OneShotPublishEnvelope.from_bootstrap_request(request)
|
|
|
|
|
|
def test_write_disabled_publisher_never_touches_injected_transport() -> None:
|
|
sink = SpySink()
|
|
publisher = WriteDisabledOneShotPublisher(sink)
|
|
envelope = _envelope()
|
|
|
|
for _ in range(2):
|
|
with pytest.raises(VendorWritesDisabledError, match="cannot publish"):
|
|
publisher.publish_once(envelope)
|
|
|
|
assert sink.calls == 0
|
|
assert publisher.snapshot().as_dict() == {
|
|
"mode": "write-disabled",
|
|
"denied_attempts": 2,
|
|
"transport_calls": 0,
|
|
"vendor_writes_enabled": False,
|
|
"publisher_armed": False,
|
|
"automatic_retry": False,
|
|
}
|
|
|
|
|
|
def test_publish_envelope_redacts_payload_and_rejects_retry_semantics() -> None:
|
|
envelope = _envelope()
|
|
|
|
assert PRIVATE_AUTHORITY not in repr(envelope)
|
|
assert PRIVATE_AUTHORITY not in str(envelope.as_dict())
|
|
with pytest.raises(ValueError, match="one-shot contract"):
|
|
OneShotPublishEnvelope(
|
|
topic=envelope.topic,
|
|
payload=envelope.payload,
|
|
payload_sha256=envelope.payload_sha256,
|
|
payload_bytes=envelope.payload_bytes,
|
|
qos=1,
|
|
retain=False,
|
|
)
|