import re import traceback from dataclasses import FrozenInstanceError, replace from datetime import UTC, datetime from decimal import Decimal from unittest.mock import MagicMock import pytest from production_analytics.enlyze.exploration import ConfigurationError from production_analytics.erp import ErpReadError, ErpSettings, ErpWorkplaceStatusGateway ENV = dict(ERP_DB_HOST='localhost', ERP_DB_PORT='49601', ERP_DB_NAME='NV_DWH', ERP_DB_USER='private-user', ERP_DB_PASSWORD='private-password') NOW = datetime(2026, 9, 4, 21, 20, 50) ROW = ('K7 ', '12026000815 ', '212520 ', 'Stex R 1501 C (PR) 5,80 x 50 m ', NOW, Decimal('80040.000'), Decimal('69281.000'), Decimal('10759.000'), 17, 38) def test_valid_settings(): settings = ErpSettings.from_environment(ENV) assert (settings.host, settings.port, settings.database) == ('localhost', 49601, 'NV_DWH') assert settings.user == ENV['ERP_DB_USER'] assert settings.password == ENV['ERP_DB_PASSWORD'] assert settings.user not in repr(settings) assert settings.password not in repr(settings) @pytest.mark.parametrize('key', ENV) @pytest.mark.parametrize('value', [None, '', ' ', '\x00']) def test_required_settings(key, value): environment = dict(ENV) if value is None: del environment[key] else: environment[key] = value with pytest.raises(ConfigurationError, match=key): ErpSettings.from_environment(environment) @pytest.mark.parametrize('port', ['0', '-1', '65536', '1.5', 'private-password']) def test_invalid_port(port): with pytest.raises(ConfigurationError, match='ERP_DB_PORT') as error: ErpSettings.from_environment(dict(ENV, ERP_DB_PORT=port)) assert 'private-password' not in ''.join(traceback.format_exception(error.value)) @pytest.mark.parametrize('port', [1, 65535]) def test_port_boundaries(port): assert ErpSettings.from_environment(dict(ENV, ERP_DB_PORT=str(port))).port == port @pytest.mark.parametrize('changes', [{'port': True}, {'port': 1.5}, {'host': ''}, {'database': ''}, {'user': ''}, {'password': ''}]) def test_direct_settings_validation(changes): with pytest.raises(ConfigurationError): replace(ErpSettings.from_environment(ENV), **changes) def test_secret_file(tmp_path): path = tmp_path / 'erp.env' path.write_text("ERP_DB_PASSWORD='file password'\n") assert ErpSettings.from_secret_file(path, environment=ENV).password == 'file password' assert ErpSettings.from_secret_file(tmp_path / 'missing', environment=ENV).port == 49601 path.write_text("ERP_DB_PASSWORD='private-password\n") with pytest.raises(ConfigurationError) as error: ErpSettings.from_secret_file(path, environment=ENV) assert 'private-password' not in ''.join(traceback.format_exception(error.value)) @pytest.fixture def db(monkeypatch): connect = MagicMock() monkeypatch.setattr('production_analytics.erp.gateway.pymssql.connect', connect) connection = connect.return_value.__enter__.return_value cursor = connection.cursor.return_value.__enter__.return_value cursor.fetchmany.return_value = [ROW] return connect, connection, cursor def read(): return ErpWorkplaceStatusGateway(ErpSettings.from_environment(ENV)) def test_valid_row_and_resource_lifecycle(db): connect, connection, cursor = db status = read().get_current_workplace_status('K7') assert status.workplace == 'K7' assert status.production_order == '12026000815' assert status.article_number == '212520' assert status.article_description == 'Stex R 1501 C (PR) 5,80 x 50 m' assert status.feedback_timestamp is NOW assert status.feedback_timestamp.tzinfo is None assert (status.order_quantity_m2, status.good_quantity_m2, status.remaining_quantity_m2, status.remaining_time_hours, status.remaining_rolls) == ( 80040., 69281., 10759., 17., 38., ) assert type(status.order_quantity_m2) is float with pytest.raises(FrozenInstanceError): status.workplace = 'other' cursor.fetchmany.assert_called_once_with(2) connection.cursor.return_value.__exit__.assert_called_once() connect.return_value.__exit__.assert_called_once() connect.assert_called_once_with(server='localhost', port=49601, database='NV_DWH', user='private-user', password='private-password', login_timeout=10, timeout=10) connection.commit.assert_not_called() def test_leading_whitespace_and_timezone_preserved(db): row = list(ROW) row[:4] = [' K7 ', ' K 7-001 ', ' 001 ', ' description '] row[4] = NOW.replace(tzinfo=UTC) db[2].fetchmany.return_value = [row] status = read().get_current_workplace_status(' K7') assert (status.workplace, status.production_order, status.article_number, status.article_description) == (' K7', ' K 7-001', ' 001', ' description') assert status.feedback_timestamp is row[4] def test_nullable_fields(db): db[2].fetchmany.return_value = [('K7', '001', None, None, NOW, *([None] * 5))] status = read().get_current_workplace_status('K7') assert all(getattr(status, key) is None for key in ( 'article_number', 'article_description', 'order_quantity_m2', 'good_quantity_m2', 'remaining_quantity_m2', 'remaining_time_hours', 'remaining_rolls', )) @pytest.mark.parametrize('value', [Decimal('NaN'), Decimal('Infinity'), Decimal('1e999'), float('nan'), True, 'private-password']) def test_invalid_numbers(db, value): row = list(ROW) row[5] = value db[2].fetchmany.return_value = [row] with pytest.raises(ErpReadError, match='invalid row') as error: read().get_current_workplace_status('K7') assert 'private-password' not in ''.join(traceback.format_exception(error.value)) def test_fractional_decimal(db): row = list(ROW) row[5:] = [Decimal('12.125')] * 5 db[2].fetchmany.return_value = [row] status = read().get_current_workplace_status('K7') assert status.order_quantity_m2 == status.remaining_rolls == 12.125 def test_zero_rows(db): db[2].fetchmany.return_value = [] assert read().get_current_workplace_status('K7') is None def test_multiple_rows(db): db[2].fetchmany.return_value = [ROW, ROW] with pytest.raises(ErpReadError, match='multiple rows'): read().get_current_workplace_status('K7') def test_parameterized_read_only_query(db): workplace = "K7'; DELETE FROM anything; --" read().get_current_workplace_status(workplace) db[2].execute.assert_called_once() sql, params = db[2].execute.call_args.args assert params == (workplace,) assert workplace not in sql assert 'WHERE [Arbeitsplatz] = %s' in sql assert re.findall(r'FROM\s+(\S+)', sql) == ['[dbo].[GRAFANA_WORKPLACE_STATUS]'] assert not re.search(r'\b(INSERT|UPDATE|DELETE|MERGE|EXEC|TOP|ORDER BY)\b', sql, re.I) assert sql.lstrip().startswith('SELECT') @pytest.mark.parametrize('stage', ['connect', 'execute', 'fetch', 'cursor_close', 'close']) def test_driver_errors_are_safe(db, stage): connect, connection, cursor = db target = {'connect': connect, 'execute': cursor.execute, 'fetch': cursor.fetchmany, 'cursor_close': connection.cursor.return_value.__exit__, 'close': connect.return_value.__exit__}[stage] target.side_effect = RuntimeError('private-user private-password') with pytest.raises(ErpReadError, match='read failed') as error: read().get_current_workplace_status('K7') rendered = ''.join(traceback.format_exception(error.value)) assert 'private-user' not in rendered assert 'private-password' not in rendered if stage != 'connect': connect.return_value.__exit__.assert_called_once()