142 lines
5.3 KiB
Python
142 lines
5.3 KiB
Python
from datetime import UTC, datetime
|
|
|
|
import pytest
|
|
|
|
from production_analytics.calculations.material_consumption import (
|
|
MaterialIntegrationState,
|
|
)
|
|
from production_analytics.service.material_polling import MaterialPollingState
|
|
from production_analytics.service.material_state_store import JsonMaterialStateStore
|
|
|
|
|
|
def test_json_state_store_returns_none_for_missing_state(tmp_path) -> None:
|
|
store = JsonMaterialStateStore(tmp_path)
|
|
|
|
assert store.load("machine-1", "ORDER-1") is None
|
|
|
|
|
|
def test_json_state_store_roundtrip(tmp_path) -> None:
|
|
store = JsonMaterialStateStore(tmp_path)
|
|
|
|
original = MaterialPollingState(
|
|
run_id="run-1",
|
|
integration_state=MaterialIntegrationState(
|
|
cumulative_consumption_kg=123.4,
|
|
integrated_running_seconds=456.0,
|
|
last_processed_timestamp=datetime(
|
|
2026, 9, 3, 5, 10, 50, tzinfo=UTC
|
|
),
|
|
last_material_rate_kg_per_hour=1028.5,
|
|
last_gate_value=2.1,
|
|
integration_active=True,
|
|
),
|
|
)
|
|
|
|
store.save("machine-1", "ORDER-1", original)
|
|
restored = store.load("machine-1", "ORDER-1")
|
|
|
|
assert restored == original
|
|
|
|
|
|
def test_json_state_store_separates_orders(tmp_path) -> None:
|
|
store = JsonMaterialStateStore(tmp_path)
|
|
|
|
state_a = MaterialPollingState(
|
|
run_id="run-a",
|
|
integration_state=MaterialIntegrationState(
|
|
cumulative_consumption_kg=10.0,
|
|
),
|
|
)
|
|
state_b = MaterialPollingState(
|
|
run_id="run-b",
|
|
integration_state=MaterialIntegrationState(
|
|
cumulative_consumption_kg=20.0,
|
|
),
|
|
)
|
|
|
|
store.save("machine-1", "ORDER-1", state_a)
|
|
store.save("machine-1", "ORDER-2", state_b)
|
|
|
|
assert store.load("machine-1", "ORDER-1") == state_a
|
|
assert store.load("machine-1", "ORDER-2") == state_b
|
|
|
|
|
|
def test_identifiers_are_isolated_safe_and_deterministic(tmp_path) -> None:
|
|
store = JsonMaterialStateStore(tmp_path)
|
|
pairs = [("a/b", "x"), ("a_b", "x"), ("a", "b__x"), ("a__b", "x"),
|
|
("../..", "/tmp/escape"), ("", ""), ("a/b", "y")]
|
|
for index, pair in enumerate(pairs):
|
|
store.save(*pair, MaterialPollingState(str(index), MaterialIntegrationState()))
|
|
assert len(list(tmp_path.iterdir())) == len(pairs)
|
|
for index, pair in enumerate(pairs):
|
|
assert JsonMaterialStateStore(tmp_path).load(*pair).run_id == str(index)
|
|
assert all(path.parent == tmp_path and len(path.name) == 69 for path in tmp_path.iterdir())
|
|
|
|
|
|
def test_failed_write_preserves_primary_and_cleans_temporary_file(tmp_path, monkeypatch) -> None:
|
|
import production_analytics.service.material_state_store as module
|
|
|
|
store = JsonMaterialStateStore(tmp_path)
|
|
state = MaterialPollingState("old", MaterialIntegrationState())
|
|
store.save("machine", "order", state)
|
|
path = next(tmp_path.iterdir())
|
|
original = path.read_bytes()
|
|
|
|
def fail_dump(payload, file, **kwargs):
|
|
file.write('{"partial":')
|
|
raise OSError("disk full")
|
|
|
|
monkeypatch.setattr(module.json, "dump", fail_dump)
|
|
with pytest.raises(OSError, match="disk full"):
|
|
store.save("machine", "order", MaterialPollingState("new", MaterialIntegrationState()))
|
|
assert path.read_bytes() == original
|
|
assert list(tmp_path.iterdir()) == [path]
|
|
|
|
|
|
def test_replace_failure_preserves_primary(tmp_path, monkeypatch) -> None:
|
|
import production_analytics.service.material_state_store as module
|
|
|
|
store = JsonMaterialStateStore(tmp_path)
|
|
state = MaterialPollingState("old", MaterialIntegrationState())
|
|
store.save("m", "o", state)
|
|
path = next(tmp_path.iterdir())
|
|
|
|
def fail_replace(source, target):
|
|
assert source.parent == target.parent == tmp_path
|
|
assert store.load("m", "o") == state
|
|
raise OSError("replace failed")
|
|
|
|
monkeypatch.setattr(module.os, "replace", fail_replace)
|
|
with pytest.raises(OSError, match="replace failed"):
|
|
store.save("m", "o", MaterialPollingState("new", MaterialIntegrationState()))
|
|
assert store.load("m", "o") == state
|
|
assert list(tmp_path.iterdir()) == [path]
|
|
|
|
|
|
@pytest.mark.parametrize("contents", ['{', '[]', '{}', '{"run_id": null}',
|
|
'{"run_id": "r", "integration_state": {}}'])
|
|
def test_corrupt_state_fails_clearly(tmp_path, contents) -> None:
|
|
store = JsonMaterialStateStore(tmp_path)
|
|
store.save("m", "o", MaterialPollingState("r", MaterialIntegrationState()))
|
|
next(tmp_path.iterdir()).write_text(contents)
|
|
with pytest.raises(ValueError, match="Invalid material polling state"):
|
|
store.load("m", "o")
|
|
|
|
|
|
@pytest.mark.parametrize("field,value", [
|
|
("integration_active", "false"), ("cumulative_consumption_kg", float("nan")),
|
|
("integrated_running_seconds", -1), ("last_processed_timestamp", "2026-09-03T05:00:00"),
|
|
("integration_active", True), ("last_gate_value", True),
|
|
])
|
|
def test_malformed_integration_state_fails(tmp_path, field, value) -> None:
|
|
import json
|
|
|
|
store = JsonMaterialStateStore(tmp_path)
|
|
store.save("m", "o", MaterialPollingState("r", MaterialIntegrationState()))
|
|
path = next(tmp_path.iterdir())
|
|
payload = json.loads(path.read_text())
|
|
payload["integration_state"][field] = value
|
|
path.write_text(json.dumps(payload))
|
|
with pytest.raises(ValueError, match="Invalid material polling state"):
|
|
store.load("m", "o")
|