from datetime import UTC, datetime, timedelta from pathlib import Path from unittest.mock import MagicMock, Mock import psycopg import pytest import yaml from production_analytics.calculations.config import ( CalculationConfigError, load_material_calculation, ) from production_analytics.calculations.material_application import rotational_application_g_m2 from production_analytics.calculations.material_consumption import MaterialSample from production_analytics.enlyze.gateway import EnlyzeApiGateway, EnlyzeProductionRun from production_analytics.service.material_polling import MaterialPollingService from production_analytics.service.postgres_material import PostgresSettings from production_analytics.service.postgres_material_application import ( PostgresMaterialApplicationWriter, ) NOW = datetime(2026, 9, 7, tzinfo=UTC) def test_realistic_application_and_generic_factor(): assert rotational_application_g_m2([3.739, 3.956], 3.12, 8, 0.3) == pytest.approx(3001.05) assert rotational_application_g_m2([2, 3], 1.5, 10, 0.5) == 750 @pytest.mark.parametrize('speed', [-10, 0, 1e-300, 0.299999999, 0.3]) def test_inactive_and_near_zero(speed): assert rotational_application_g_m2([3.739, 3.956], 3.12, speed, 0.3) is None @pytest.mark.parametrize('speeds,factor,speed,threshold', [ ([], 3.12, 8, 0.3), ([float('nan')], 3.12, 8, 0.3), ([1], 0, 8, 0.3), ([1], float('inf'), 8, 0.3), ([1], 3.12, float('nan'), 0.3), ([1], 3.12, 8, 0), ([1], 3.12, 8, float('inf')), ([1e308], 3.12, 8, 0.3), ]) def test_invalid_inputs(speeds, factor, speed, threshold): with pytest.raises(ValueError): rotational_application_g_m2(speeds, factor, speed, threshold) @pytest.mark.parametrize('width', [1, 4.85, 5, 100]) def test_gateway_width_cancels_and_only_act_signals_requested(width): config = load_material_calculation('config/bento1-material-consumption.yaml') refs = [*config.rotational_speed_signal_refs, config.gate_signal_ref] client = Mock() client.post_json.return_value.body = {'data': { 'columns': ['time', *refs], 'records': [[NOW.isoformat(), 3.739, 3.956, 8], [(NOW + timedelta(seconds=10)).isoformat(), 3.739, 3.956, 0]], }} samples = EnlyzeApiGateway(client).get_material_samples( machine_id=config.machine_ref, rate_variable_id='unused', gate_variable_id=config.gate_signal_ref, start=NOW, end=NOW + timedelta(seconds=10), rotational_speed_variable_ids=config.rotational_speed_signal_refs, nominal_width_m=width, specific_discharge_kg_per_rev_m=3.12, process_application_gate_threshold=0.3, ) assert samples[0].application_g_m2 == pytest.approx(3001.05) assert samples[1].application_g_m2 is None assert samples[0].material_rate_kg_per_hour == pytest.approx(7.695 * width * 3.12 * 60) assert client.post_json.call_args.args[1]['variables'] == [{'uuid': ref} for ref in refs] @pytest.mark.parametrize('changes', [ {'process_application_calculation_id': ''}, {'process_application_calculation_id': None}, {'process_application_calculation_id': 3}, {'process_application_calculation_id': 'x\x00'}, {'process_application_calculation_id': 'bento1-fresh-bentonite-consumption'}, {'gate_signal_ref': None}, {'gate_threshold': 0}, {'gate_threshold': -0.3}, ]) def test_process_config_validation(tmp_path, changes): document = yaml.safe_load(Path('config/bento1-material-consumption.yaml').read_text()) document['calculations'][0].update(changes) path = tmp_path / 'config.yaml' path.write_text(yaml.safe_dump(document)) with pytest.raises(CalculationConfigError): load_material_calculation(path) def test_persistence_before_checkpoint_retries_and_disjoint_runs(): old = EnlyzeProductionRun('old', 'machine', None, 'order', NOW, NOW + timedelta(seconds=10)) current = EnlyzeProductionRun( 'new', 'machine', None, 'order', NOW + timedelta(seconds=15), None, ) gateway = Mock() gateway.get_open_production_run.return_value = current gateway.get_production_runs.return_value = [old, current] def samples(**kwargs): start = kwargs['start'] assert kwargs['process_application_gate_threshold'] == 0.3 return [MaterialSample(start, 7202.52, 8, 3001.05), MaterialSample(start + timedelta(seconds=10), 7202.52, 8, 3001.05)] gateway.get_material_samples.side_effect = samples store = Mock(load=Mock(return_value=None)) writer = Mock() writer.write.side_effect = [None, psycopg.OperationalError(), None, None] service = MaterialPollingService( gateway=gateway, state_store=store, machine_id='machine', rate_variable_id='unused', gate_variable_id='speed', gate_threshold=0.3, max_sample_gap_seconds=20, rotational_speed_variable_ids=('right', 'left'), specific_discharge_kg_per_rev_m=3.12, nominal_width_provider=lambda order: 5, process_application_calculation_id='application', application_writer=writer, ) with pytest.raises(psycopg.OperationalError): service.poll_once(now=NOW + timedelta(seconds=25)) store.save.assert_not_called() result = service.poll_once(now=NOW + timedelta(seconds=25)) assert result.state.cumulative_consumption_kg == pytest.approx(40.014) assert result.state.integrated_running_seconds == 20 assert [c.kwargs['run_id'] for c in writer.write.call_args_list] == ['old', 'new', 'old', 'new'] store.save.assert_called_once() def test_sql_persistence_idempotent_and_inactive_omitted(monkeypatch): import sqlite3 database = sqlite3.connect(':memory:') database.executescript(Path('db/schema.sql').read_text()) connection = MagicMock() cursor = connection.__enter__.return_value.cursor.return_value.__enter__.return_value cursor.executemany.side_effect = lambda sql, rows: database.executemany( sql.replace('%s', '?'), [(row[0].isoformat(), *row[1:]) for row in rows], ) connect = Mock(return_value=connection) monkeypatch.setattr(psycopg, 'connect', connect) writer = PostgresMaterialApplicationWriter(PostgresSettings('host', 5432, 'db', 'u', 'p')) kwargs = dict(calculation_id='application', machine_id='machine', production_order='order', run_id='run', samples=[MaterialSample(NOW, 7202.52, 8, 3001.05), MaterialSample(NOW + timedelta(seconds=10), 7202.52, 0)]) writer.write(**kwargs) writer.write(**kwargs) assert database.execute('SELECT * FROM material_application_snapshots').fetchall() == [ (NOW.isoformat(), 'application', 'machine', 'order', 'run', 3001.05), ] writer.write(**(kwargs | {'samples': [MaterialSample(NOW, 1, 0)]})) assert connect.call_count == 2 database.close() @pytest.mark.parametrize('collision', ['output', 'consumption']) def test_output_ids_unique_across_config_entries(tmp_path, collision): document = yaml.safe_load(Path('config/bento1-material-consumption.yaml').read_text()) first = document['calculations'][0] second = dict(first, id='second-consumption') if collision == 'consumption': second['process_application_calculation_id'] = first['id'] document['calculations'].append(second) path = tmp_path / 'config.yaml' path.write_text(yaml.safe_dump(document)) with pytest.raises(CalculationConfigError, match='unique'): load_material_calculation(path, first['id'])