75 lines
2.7 KiB
Python
75 lines
2.7 KiB
Python
"""Tests for hameter.cost_state persistence."""
|
|
|
|
import json
|
|
|
|
from hameter.cost_state import load_cost_state, save_cost_state
|
|
|
|
|
|
class TestCostStatePersistence:
|
|
"""Tests for load/save of cost state."""
|
|
|
|
def test_save_and_load(self, tmp_path):
|
|
path = str(tmp_path / "cost_state.json")
|
|
states = {
|
|
"123": {
|
|
"cumulative_cost": 156.78,
|
|
"last_calibrated_reading": 59830.5,
|
|
"billing_period_start": "2026-03-01T00:00:00Z",
|
|
"last_updated": "2026-03-05T14:30:00Z",
|
|
"fixed_charges_applied": 9.65,
|
|
}
|
|
}
|
|
save_cost_state(states, path)
|
|
loaded = load_cost_state(path)
|
|
assert loaded["123"]["cumulative_cost"] == 156.78
|
|
assert loaded["123"]["last_calibrated_reading"] == 59830.5
|
|
assert loaded["123"]["fixed_charges_applied"] == 9.65
|
|
|
|
def test_load_nonexistent_returns_empty(self, tmp_path):
|
|
path = str(tmp_path / "nonexistent.json")
|
|
result = load_cost_state(path)
|
|
assert result == {}
|
|
|
|
def test_load_corrupt_returns_empty(self, tmp_path):
|
|
path = str(tmp_path / "cost_state.json")
|
|
with open(path, "w") as f:
|
|
f.write("not json")
|
|
result = load_cost_state(path)
|
|
assert result == {}
|
|
|
|
def test_creates_parent_directory(self, tmp_path):
|
|
path = str(tmp_path / "subdir" / "cost_state.json")
|
|
save_cost_state({"1": {"cumulative_cost": 0}}, path)
|
|
loaded = load_cost_state(path)
|
|
assert "1" in loaded
|
|
|
|
def test_no_temp_file_left(self, tmp_path):
|
|
path = str(tmp_path / "cost_state.json")
|
|
save_cost_state({"1": {"cumulative_cost": 10.0}}, path)
|
|
files = list(tmp_path.iterdir())
|
|
assert len(files) == 1
|
|
assert files[0].name == "cost_state.json"
|
|
|
|
def test_multiple_meters(self, tmp_path):
|
|
path = str(tmp_path / "cost_state.json")
|
|
states = {
|
|
"111": {"cumulative_cost": 50.0, "last_calibrated_reading": 1000.0},
|
|
"222": {"cumulative_cost": 25.0, "last_calibrated_reading": 500.0},
|
|
}
|
|
save_cost_state(states, path)
|
|
loaded = load_cost_state(path)
|
|
assert len(loaded) == 2
|
|
assert loaded["111"]["cumulative_cost"] == 50.0
|
|
assert loaded["222"]["cumulative_cost"] == 25.0
|
|
|
|
def test_atomic_write_format(self, tmp_path):
|
|
path = str(tmp_path / "cost_state.json")
|
|
states = {"123": {"cumulative_cost": 42.0}}
|
|
save_cost_state(states, path)
|
|
with open(path) as f:
|
|
raw = f.read()
|
|
# Should be pretty-printed JSON with trailing newline
|
|
assert raw.endswith("\n")
|
|
parsed = json.loads(raw)
|
|
assert parsed == states
|