42 lines
1.3 KiB
Python
42 lines
1.3 KiB
Python
from datetime import datetime
|
|
|
|
import pytest
|
|
|
|
from energi_bridge import AuditLog, ContractRegistry, LocalProblem, PrivacyBudget
|
|
|
|
|
|
def test_contract_registry_tracks_versions() -> None:
|
|
registry = ContractRegistry()
|
|
|
|
first = registry.register(
|
|
contract_id="solar_meter.site-1",
|
|
version="0.1.0",
|
|
adapter_id="SolarMeterAdapter",
|
|
schema_name="SolarMeterReading",
|
|
)
|
|
second = registry.register(
|
|
contract_id="solar_meter.site-1",
|
|
version="0.2.0",
|
|
adapter_id="SolarMeterAdapter",
|
|
schema_name="SolarMeterReading",
|
|
)
|
|
|
|
assert registry.latest_version("solar_meter.site-1") == "0.2.0"
|
|
assert registry.resolve("solar_meter.site-1").version == "0.2.0"
|
|
assert first.previous_version is None
|
|
assert second.previous_version == "0.1.0"
|
|
assert [item.contract_version for item in registry.manifest()] == ["0.1.0", "0.2.0"]
|
|
|
|
|
|
def test_schema_validations_and_governance_helpers() -> None:
|
|
with pytest.raises(ValueError):
|
|
LocalProblem(id="", domain="solar", assets=["panel-1"])
|
|
|
|
budget = PrivacyBudget(budget=1.0)
|
|
budget.consume(0.25)
|
|
assert budget.spent == 0.25
|
|
|
|
log = AuditLog()
|
|
log.record("cycle_completed", ts=datetime.utcnow().isoformat())
|
|
assert log.entries[0]["event"] == "cycle_completed"
|