diff --git a/README.md b/README.md index 821fdc2..5bdfdf5 100644 --- a/README.md +++ b/README.md @@ -217,6 +217,42 @@ start times, not source-sample timestamps. Schema application is manual. Bento 1 bentonite source and gate selection remain intentionally undefined pending process validation; the generic integration core is unchanged and reusable. +## ERP current workplace status + +The read-only ERP adapter reads production-order/article context from MSSQL +`NV_DWH.dbo.GRAFANA_WORKPLACE_STATUS`, for later live material KPIs. Credentials +belong in ignored `secrets/erp.env`, using `ERP_DB_HOST`, `ERP_DB_PORT`, +`ERP_DB_NAME`, `ERP_DB_USER`, and `ERP_DB_PASSWORD`. The verified endpoint is +`192.168.111.24:49601`, database `NV_DWH`, using SQL Server authentication and +a read-only account. All settings are required; ports must be integers 1–65535. + +```python +from production_analytics.erp import ErpSettings, ErpWorkplaceStatusGateway + +settings = ErpSettings.from_secret_file() # File values override environment values. +status = ErpWorkplaceStatusGateway(settings).get_current_workplace_status("K7") +``` + +Alternatively, use `ErpSettings.from_environment(mapping)`. The existing simple +dotenv loader is reused; no shell content is executed. Each call opens and closes +one connection, with 10-second login/query timeouts, and performs a parameterized +SELECT filtered by workplace. Zero rows returns `None`, one row returns an +immutable `CurrentWorkplaceStatus`, and multiple rows raise `ErpReadError`. +Driver failures and malformed rows also raise `ErpReadError` with safe messages. + +Only trailing whitespace is removed from workplace, production order, article +number, and description; identifiers otherwise remain unchanged. Nullable fields +stay `None`. Decimal/integer quantities are explicitly converted to finite floats +(with normal floating-point precision); non-finite/overflowing values are rejected. +Quantity fields use m² and remaining time uses hours, following the supplied source +semantics. `feedback_timestamp` is preserved exactly, including a naive timezone. +It represents the latest ERP feedback, and the ERP export is delayed relative to +live process data; the adapter makes no freshness inference. + +This milestone adds no kg/m² or material-efficiency calculation, ERP-to-ENLYZE +order mapping, persistence, polling, CLI, or Grafana integration. Automated ERP +tests use mocks and require no live connectivity. + ## Peak-cycle detection `PeakCycleDetector` is a pure calculation-domain component for roll length, diff --git a/pyproject.toml b/pyproject.toml index 17ca6a9..b81cd83 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,7 +10,7 @@ readme = "README.md" requires-python = ">=3.11" license = { text = "Proprietary" } authors = [{ name = "Production Analytics Team" }] -dependencies = ["PyYAML>=6.0", "psycopg[binary]>=3.2,<4"] +dependencies = ["PyYAML>=6.0", "psycopg[binary]>=3.2,<4", "pymssql>=2.4.0,<3"] [project.optional-dependencies] dev = [ diff --git a/src/production_analytics/erp/__init__.py b/src/production_analytics/erp/__init__.py new file mode 100644 index 0000000..01d508c --- /dev/null +++ b/src/production_analytics/erp/__init__.py @@ -0,0 +1,10 @@ +"""Read-only ERP current workplace context.""" + +from production_analytics.erp.gateway import ( + CurrentWorkplaceStatus, + ErpReadError, + ErpSettings, + ErpWorkplaceStatusGateway, +) + +__all__ = ["CurrentWorkplaceStatus", "ErpReadError", "ErpSettings", "ErpWorkplaceStatusGateway"] diff --git a/src/production_analytics/erp/gateway.py b/src/production_analytics/erp/gateway.py new file mode 100644 index 0000000..0c87e13 --- /dev/null +++ b/src/production_analytics/erp/gateway.py @@ -0,0 +1,148 @@ +"""Focused MSSQL boundary for dbo.GRAFANA_WORKPLACE_STATUS.""" + +import os +from collections.abc import Mapping +from dataclasses import dataclass, field +from datetime import datetime +from decimal import Decimal +from math import isfinite + +import pymssql + +from production_analytics.enlyze.exploration import ConfigurationError, load_secret_file + + +@dataclass(frozen=True, slots=True) +class ErpSettings: + host: str + port: int + database: str + user: str = field(repr=False) + password: str = field(repr=False) + + def __post_init__(self) -> None: + for attribute, suffix in ( + ("host", "HOST"), ("database", "NAME"), ("user", "USER"), ("password", "PASSWORD"), + ): + value = getattr(self, attribute) + if not isinstance(value, str) or not value.strip() or "\x00" in value: + raise ConfigurationError(f"ERP_DB_{suffix} must be non-empty without NUL bytes") + if type(self.port) is not int or not 1 <= self.port <= 65535: + raise ConfigurationError("ERP_DB_PORT must be an integer from 1 to 65535") + + @classmethod + def from_environment(cls, environment: Mapping[str, str]) -> "ErpSettings": + raw_port = environment.get("ERP_DB_PORT", "") + try: + if not isinstance(raw_port, str): + raise ValueError + port = int(raw_port) + except ValueError: + raise ConfigurationError("ERP_DB_PORT must be an integer from 1 to 65535") from None + return cls( + host=environment.get("ERP_DB_HOST", ""), port=port, + database=environment.get("ERP_DB_NAME", ""), + user=environment.get("ERP_DB_USER", ""), + password=environment.get("ERP_DB_PASSWORD", ""), + ) + + @classmethod + def from_secret_file( + cls, path: str | os.PathLike[str] = "secrets/erp.env", + *, environment: Mapping[str, str] | None = None, + ) -> "ErpSettings": + """Load dotenv values over the environment, following the existing convention.""" + try: + values = load_secret_file(path) + except Exception: + raise ConfigurationError("ERP secret file could not be loaded") from None + base = os.environ if environment is None else environment + return cls.from_environment({**base, **values}) + + +@dataclass(frozen=True, slots=True) +class CurrentWorkplaceStatus: + """Latest ERP feedback; timestamp and timezone are preserved, with no freshness claim.""" + + workplace: str + production_order: str + article_number: str | None + article_description: str | None + feedback_timestamp: datetime + order_quantity_m2: float | None + good_quantity_m2: float | None + remaining_quantity_m2: float | None + remaining_time_hours: float | None + remaining_rolls: float | None + + +class ErpReadError(RuntimeError): + """Safe operator-facing ERP read or result error.""" + + +_CURRENT_STATUS_SQL = """SELECT + [Arbeitsplatz], [Fertigungsauftragsnummer], [Artikelnummer], [Artikelbezeichnung], + [Zeitstempel], [Auftragsmenge], [Gutmenge], [Restmenge], + [Verbleibende Zeit], [Verbleibende Rollen] +FROM [dbo].[GRAFANA_WORKPLACE_STATUS] +WHERE [Arbeitsplatz] = %s""" + + +class ErpWorkplaceStatusGateway: + def __init__(self, settings: ErpSettings) -> None: + self._settings = settings + + def get_current_workplace_status(self, workplace: str) -> CurrentWorkplaceStatus | None: + if not isinstance(workplace, str) or not workplace.strip() or "\x00" in workplace: + raise ValueError("workplace must be a non-empty string without NUL bytes") + try: + with pymssql.connect( + server=self._settings.host, port=self._settings.port, + database=self._settings.database, user=self._settings.user, + password=self._settings.password, login_timeout=10, timeout=10, + ) as connection: + with connection.cursor() as cursor: + cursor.execute(_CURRENT_STATUS_SQL, (workplace,)) + # Two rows suffice to detect a violation without selecting an arbitrary row. + rows = cursor.fetchmany(2) + except Exception: + # Driver messages may include credentials; suppress their traceback chain too. + raise ErpReadError("ERP current workplace status read failed") from None + if not rows: + return None + if len(rows) > 1: + raise ErpReadError("ERP current workplace status returned multiple rows") + try: + row = rows[0] + if len(row) != 10 or not isinstance(row[4], datetime): + raise ValueError + return CurrentWorkplaceStatus( + workplace=_text(row[0]), production_order=_text(row[1]), + article_number=_text(row[2], nullable=True), + article_description=_text(row[3], nullable=True), + feedback_timestamp=row[4], + order_quantity_m2=_number(row[5]), good_quantity_m2=_number(row[6]), + remaining_quantity_m2=_number(row[7]), remaining_time_hours=_number(row[8]), + remaining_rolls=_number(row[9]), + ) + except Exception: + raise ErpReadError("ERP current workplace status returned an invalid row") from None + + +def _text(value: object, *, nullable: bool = False) -> str | None: + if value is None and nullable: + return None + if not isinstance(value, str): + raise ValueError + return value.rstrip() + + +def _number(value: object) -> float | None: + if value is None: + return None + if isinstance(value, bool) or not isinstance(value, (Decimal, int, float)): + raise ValueError + result = float(value) + if not isfinite(result): + raise ValueError + return result diff --git a/tests/test_erp_gateway.py b/tests/test_erp_gateway.py new file mode 100644 index 0000000..9e6ad4a --- /dev/null +++ b/tests/test_erp_gateway.py @@ -0,0 +1,187 @@ +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()