44 lines
1.4 KiB
Python
44 lines
1.4 KiB
Python
import sys
|
|
import os
|
|
import time
|
|
|
|
# Ensure package source is importable during tests
|
|
SRC_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "src"))
|
|
if SRC_DIR not in sys.path:
|
|
sys.path.insert(0, SRC_DIR)
|
|
|
|
from pulsemesh_open_telemetry_visualization_a.delta import Delta
|
|
from pulsemesh_open_telemetry_visualization_a.protocol import TelemetryContract, DeltaEnvelope
|
|
from pulsemesh_open_telemetry_visualization_a.catopt_bridge import map_delta_to_catopt
|
|
|
|
|
|
def test_poetic_cert_roundtrip_and_mapping():
|
|
contract = TelemetryContract(
|
|
contract_id="c1",
|
|
device_type="test_device",
|
|
metrics=["m1"],
|
|
units={"m1": "unit"},
|
|
quality_flags=["ok"],
|
|
version=1,
|
|
)
|
|
|
|
d = Delta(delta_id="d1", timestamp=time.time(), items=[{"m1": 1.0}])
|
|
|
|
poetic = {
|
|
"model_version": "v1",
|
|
"training_data_digest": "sha256:deadbeef",
|
|
"confidence": 0.95,
|
|
"signer": "edge-operator",
|
|
}
|
|
|
|
env = DeltaEnvelope(protocol_version="PulseMesh/0.1.0", contract=contract, delta=d, poetic_cert=poetic)
|
|
|
|
# Ensure envelope preserves poetic_cert
|
|
ed = env.to_dict()
|
|
assert "poetic_cert" in ed and ed["poetic_cert"]["model_version"] == "v1"
|
|
|
|
cat = map_delta_to_catopt(env)
|
|
assert isinstance(cat, dict)
|
|
assert "Metadata" in cat and cat["Metadata"].get("poetic_cert") is not None
|
|
assert cat["Metadata"]["poetic_cert"]["signer"] == "edge-operator"
|