41 lines
1.4 KiB
Python
41 lines
1.4 KiB
Python
from __future__ import annotations
|
|
|
|
import time
|
|
|
|
from pulsemesh_open_telemetry_visualization_a.delta import Delta
|
|
from pulsemesh_open_telemetry_visualization_a.protocol import TelemetryContract
|
|
from pulsemesh_open_telemetry_visualization_a.protocol import DeltaEnvelope
|
|
from pulsemesh_open_telemetry_visualization_a.catopt_bridge import map_delta_to_catopt
|
|
|
|
|
|
def _sample_contract() -> TelemetryContract:
|
|
return TelemetryContract(
|
|
contract_id="test-contract-1",
|
|
device_type="test_device",
|
|
metrics=["telemetry.sample"],
|
|
units={"telemetry.sample": "unit"},
|
|
quality_flags=["ok"],
|
|
version=1,
|
|
)
|
|
|
|
|
|
def test_catopt_bridge_basic_mapping():
|
|
# Build a minimal delta
|
|
delta = Delta(delta_id="d1", timestamp=time.time(), items=[{"dummy": True}])
|
|
envelope = DeltaEnvelope(protocol_version="PulseMesh/0.1.0", contract=_sample_contract(), delta=delta)
|
|
|
|
catopt = map_delta_to_catopt(envelope)
|
|
|
|
# Ensure the top-level keys exist
|
|
assert isinstance(catopt, dict)
|
|
assert "Object" in catopt and "Morphism" in catopt and "PlanDelta" in catopt
|
|
|
|
# Validate some shapes inside
|
|
obj = catopt["Object"]
|
|
assert obj["type"] == "TelemetryContext"
|
|
assert obj["contract_id"] == envelope.contract.contract_id
|
|
|
|
morph = catopt["Morphism"]
|
|
assert morph["type"] == "AnomalySignalChannel"
|
|
assert isinstance(morph["delta"], dict)
|