102 lines
4.3 KiB
Python
102 lines
4.3 KiB
Python
import sqlite3
|
|
import sys
|
|
from datetime import UTC, datetime
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from production_analytics.service.postgres_material import (
|
|
PostgresMaterialSnapshotRepository,
|
|
PostgresSettings,
|
|
)
|
|
|
|
START = datetime(2026, 9, 4, 10, tzinfo=UTC)
|
|
ORDER = " K 7-123'; -- "
|
|
SETTINGS = PostgresSettings('localhost', 5432, 'analytics', 'reader', 'secret')
|
|
|
|
|
|
@pytest.fixture
|
|
def database():
|
|
# Execute the actual portable SELECT in SQLite; only adapt driver placeholders
|
|
# and datetime transport. This tests SQL semantics without a live PostgreSQL server.
|
|
with sqlite3.connect(':memory:') as database:
|
|
database.execute('''CREATE TABLE material_consumption_snapshots (
|
|
timestamp TEXT, calculation_id TEXT, machine_id TEXT, production_order TEXT,
|
|
run_id TEXT, consumption_kg REAL
|
|
)''')
|
|
rows = [
|
|
(START.isoformat(), 'calc', 'machine', ORDER, 'run1', 1000),
|
|
(START.replace(minute=10).isoformat(), 'calc', 'machine', ORDER, 'run2', 1100),
|
|
(START.replace(minute=4).isoformat(), 'other', 'machine', ORDER, 'run', 9999),
|
|
(START.replace(minute=4).isoformat(), 'calc', 'other', ORDER, 'run', 9999),
|
|
(START.replace(minute=4).isoformat(), 'calc', 'machine', ORDER.strip(), 'run', 9999),
|
|
(START.replace(minute=4).isoformat(), 'calc', 'machine', ORDER + '-other', 'run', 9999),
|
|
]
|
|
database.executemany(
|
|
'INSERT INTO material_consumption_snapshots VALUES (?,?,?,?,?,?)', rows,
|
|
)
|
|
yield database
|
|
|
|
|
|
@pytest.mark.parametrize('minute, expected_minute, consumption', [
|
|
(-1, None, None), (0, 0, 1000), (5, 0, 1000), (10, 10, 1100), (15, 10, 1100),
|
|
])
|
|
def test_aligned_lookup_sql(database, minute, expected_minute, consumption):
|
|
cutoff = START.replace(minute=minute) if minute >= 0 else START.replace(hour=9, minute=59)
|
|
driver = MagicMock()
|
|
connection = driver.connect.return_value.__enter__.return_value
|
|
|
|
def execute(sql, parameters):
|
|
assert parameters == ('calc', 'machine', ORDER, cutoff)
|
|
assert sql.count('%s') == 4
|
|
assert ORDER not in sql
|
|
assert 'ORDER BY timestamp DESC' in sql
|
|
assert 'LIMIT 1' in sql
|
|
row = database.execute(
|
|
sql.replace('%s', '?'), (*parameters[:3], parameters[3].isoformat()),
|
|
).fetchone()
|
|
if row is not None:
|
|
row = (datetime.fromisoformat(row[0]), *row[1:])
|
|
return MagicMock(fetchone=MagicMock(return_value=row))
|
|
|
|
connection.execute.side_effect = execute
|
|
with patch.dict(sys.modules, psycopg=driver):
|
|
result = PostgresMaterialSnapshotRepository(SETTINGS).latest_at_or_before(
|
|
calculation_id='calc', machine_id='machine', production_order=ORDER, timestamp=cutoff,
|
|
)
|
|
if expected_minute is None:
|
|
assert result is None
|
|
else:
|
|
assert result.timestamp == START.replace(minute=expected_minute)
|
|
assert result.timestamp <= cutoff
|
|
assert result.consumption_kg == consumption
|
|
assert (result.calculation_id, result.machine_id, result.production_order) == (
|
|
'calc', 'machine', ORDER,
|
|
)
|
|
assert result.run_id == ('run1' if expected_minute == 0 else 'run2')
|
|
connection.execute.assert_called_once()
|
|
driver.connect.assert_called_once_with(
|
|
host='localhost', port=5432, dbname='analytics', user='reader', password='secret',
|
|
connect_timeout=10, options='-c statement_timeout=10000',
|
|
)
|
|
driver.connect.return_value.__exit__.assert_called_once_with(None, None, None)
|
|
|
|
|
|
def test_naive_cutoff_rejected_before_connection():
|
|
driver = MagicMock()
|
|
with patch.dict(sys.modules, psycopg=driver), pytest.raises(ValueError, match='timezone-aware'):
|
|
PostgresMaterialSnapshotRepository(SETTINGS).latest_at_or_before(
|
|
calculation_id='calc', machine_id='machine', production_order=ORDER,
|
|
timestamp=START.replace(tzinfo=None),
|
|
)
|
|
driver.connect.assert_not_called()
|
|
|
|
|
|
def test_database_failure_propagates():
|
|
driver = MagicMock()
|
|
driver.connect.side_effect = RuntimeError('connection unavailable')
|
|
with patch.dict(sys.modules, psycopg=driver), pytest.raises(RuntimeError):
|
|
PostgresMaterialSnapshotRepository(SETTINGS).latest_at_or_before(
|
|
calculation_id='calc', machine_id='machine', production_order=ORDER, timestamp=START,
|
|
)
|