62 lines
2.6 KiB
Python
62 lines
2.6 KiB
Python
"""Regression coverage for SQLite initialization from production DDL."""
|
|
|
|
import json
|
|
import sqlite3
|
|
|
|
import pytest
|
|
|
|
|
|
def test_schema_initialization_and_power_meter_json(sqlite_schema):
|
|
with sqlite3.connect(":memory:") as database:
|
|
database.executescript(sqlite_schema)
|
|
database.executescript(sqlite_schema)
|
|
tables = {
|
|
row[0] for row in database.execute("SELECT name FROM sqlite_master WHERE type='table'")
|
|
}
|
|
assert {
|
|
"material_consumption_snapshots",
|
|
"material_efficiency_snapshots",
|
|
"material_application_snapshots",
|
|
"channel_material_consumption_snapshots",
|
|
"production_downtime_events",
|
|
"production_order_attribution_state",
|
|
"power_meter_readings",
|
|
"power_meter_monthly_reports",
|
|
} <= tables
|
|
|
|
insert = """INSERT INTO power_meter_readings
|
|
(meter, timestamp, energy_total_kwh, energy_delta_kwh,
|
|
active_power_total_kw, grid_frequency_hz)
|
|
VALUES (?, ?, 100, 1, 4, 50)"""
|
|
database.execute(insert, ("B2", "2026-09-01T00:00:00Z"))
|
|
phases = {"L1": 1.25, "L2": 2.75}
|
|
database.execute(insert, ("B2", "2026-09-01T00:01:00Z"))
|
|
database.execute(
|
|
"UPDATE power_meter_readings SET phase_active_power_kw=? WHERE timestamp=?",
|
|
(json.dumps(phases), "2026-09-01T00:01:00Z"),
|
|
)
|
|
rows = database.execute(
|
|
"SELECT phase_active_power_kw FROM power_meter_readings ORDER BY timestamp"
|
|
).fetchall()
|
|
assert [json.loads(row[0]) for row in rows] == [{}, phases]
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
database.execute(insert, ("B2", "2026-09-01T00:00:00Z"))
|
|
|
|
summary = {"consumption_kwh": 42.5, "first_timestamp": None, "outages": 0}
|
|
database.execute(
|
|
"""INSERT INTO power_meter_monthly_reports (meter, month, report_path, summary)
|
|
VALUES (?, ?, ?, ?)""",
|
|
("B2", "2026-09", "/reports/B2.pdf", json.dumps(summary)),
|
|
)
|
|
path, stored_summary, generated_at = database.execute(
|
|
"SELECT report_path, summary, generated_at FROM power_meter_monthly_reports"
|
|
).fetchone()
|
|
assert path == "/reports/B2.pdf"
|
|
assert json.loads(stored_summary) == summary
|
|
assert generated_at is not None
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
database.execute(
|
|
"""INSERT INTO power_meter_monthly_reports (meter, month, report_path, summary)
|
|
VALUES ('B2', '2026-10', '/reports/B2.pdf', NULL)"""
|
|
)
|