188 lines
7.7 KiB
Python
188 lines
7.7 KiB
Python
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()
|