187 lines
7.6 KiB
Python
187 lines
7.6 KiB
Python
from dataclasses import FrozenInstanceError, replace
|
|
from datetime import UTC, datetime
|
|
from unittest.mock import Mock
|
|
from zoneinfo import ZoneInfo
|
|
|
|
import pytest
|
|
|
|
from production_analytics.erp import CurrentWorkplaceStatus
|
|
from production_analytics.service.material_efficiency import MaterialEfficiencyService
|
|
from production_analytics.service.postgres_material import MaterialConsumptionSnapshot
|
|
|
|
CALC = 'k7-fiber-consumption'
|
|
MACHINE = 'c220f95c-a65e-4cb7-99b7-0626d6c7508c'
|
|
START = datetime(2026, 9, 4, 10, tzinfo=UTC)
|
|
FEEDBACK = START.replace(minute=5)
|
|
STATUS = CurrentWorkplaceStatus(
|
|
'K7', '12026000815', '212520', 'Stex R 1501 C (PR) 5,80 x 50 m',
|
|
FEEDBACK, None, 800, None, None, None,
|
|
)
|
|
MATERIAL = MaterialConsumptionSnapshot(START, CALC, MACHINE, 'K 7-12026000815', 'run', 1000)
|
|
|
|
|
|
def service(repository=None, **config):
|
|
return MaterialEfficiencyService(
|
|
Mock(get_current_workplace_status=Mock(return_value=STATUS)),
|
|
repository if repository is not None else Mock(latest_at_or_before=Mock(
|
|
return_value=MATERIAL,
|
|
)),
|
|
**dict(workplace='K7', machine_id=MACHINE, calculation_id=CALC,
|
|
format_template='K 7-{production_order}', **config),
|
|
)
|
|
|
|
|
|
def test_valid_current_feedback():
|
|
subject = service()
|
|
result = subject.evaluate_current()
|
|
assert result.material_consumption_kg_per_m2 == 1.25
|
|
assert result.material_consumption_g_per_m2 == 1250
|
|
assert result.nominal_width_m == 5.8
|
|
assert result.good_quantity_m2 == 800
|
|
assert result.material_consumption_kg == 1000
|
|
assert result.erp_feedback_timestamp == FEEDBACK
|
|
assert result.material_snapshot_timestamp == START
|
|
assert (result.workplace, result.machine_id, result.calculation_id) == ('K7', MACHINE, CALC)
|
|
assert (result.erp_production_order, result.enlyze_production_order) == (
|
|
'12026000815', 'K 7-12026000815',
|
|
)
|
|
assert (result.article_number, result.article_description) == (
|
|
STATUS.article_number, STATUS.article_description,
|
|
)
|
|
subject.erp.get_current_workplace_status.assert_called_once_with('K7')
|
|
subject.materials.latest_at_or_before.assert_called_once_with(
|
|
calculation_id=CALC, machine_id=MACHINE, production_order='K 7-12026000815',
|
|
timestamp=FEEDBACK,
|
|
)
|
|
with pytest.raises(FrozenInstanceError):
|
|
result.good_quantity_m2 = 5
|
|
|
|
|
|
@pytest.mark.parametrize('quantity', [None, 0, -1, float('nan'), float('inf'), -float('inf')])
|
|
def test_invalid_quantity(quantity):
|
|
subject = service()
|
|
assert subject.evaluate(replace(STATUS, good_quantity_m2=quantity)) is None
|
|
subject.materials.latest_at_or_before.assert_not_called()
|
|
|
|
|
|
@pytest.mark.parametrize('description', [None, 'unstructured article'])
|
|
def test_width_is_optional_context(description):
|
|
result = service().evaluate(replace(STATUS, article_description=description))
|
|
assert result.nominal_width_m is None
|
|
assert result.material_consumption_kg_per_m2 == 1.25
|
|
|
|
|
|
def test_unavailable_sources():
|
|
subject = service()
|
|
subject.erp.get_current_workplace_status.return_value = None
|
|
assert subject.evaluate_current() is None
|
|
subject.materials.latest_at_or_before.return_value = None
|
|
assert subject.evaluate(STATUS) is None
|
|
|
|
|
|
@pytest.mark.parametrize('consumption, quantity', [
|
|
(float('nan'), 800), (float('inf'), 800), (-1, 800),
|
|
(1e308, 1e-308), (1e308, 1),
|
|
])
|
|
def test_invalid_consumption_and_overflow(consumption, quantity):
|
|
subject = service()
|
|
subject.materials.latest_at_or_before.return_value = replace(
|
|
MATERIAL, consumption_kg=consumption,
|
|
)
|
|
assert subject.evaluate(replace(STATUS, good_quantity_m2=quantity)) is None
|
|
|
|
|
|
def test_zero_consumption_is_valid():
|
|
subject = service()
|
|
subject.materials.latest_at_or_before.return_value = replace(MATERIAL, consumption_kg=0)
|
|
assert subject.evaluate(STATUS).material_consumption_kg_per_m2 == 0
|
|
|
|
|
|
@pytest.mark.parametrize('changes', [
|
|
dict(timestamp=START.replace(minute=10)), dict(timestamp=START.replace(tzinfo=None)),
|
|
dict(production_order='K 7-12026000815-K 7-12026000816'),
|
|
dict(machine_id='other'), dict(calculation_id='other'),
|
|
])
|
|
def test_repository_contract_is_checked(changes):
|
|
subject = service()
|
|
subject.materials.latest_at_or_before.return_value = replace(MATERIAL, **changes)
|
|
with pytest.raises(ValueError, match='unaligned'):
|
|
subject.evaluate(STATUS)
|
|
|
|
|
|
def test_other_explicit_configuration():
|
|
repository = Mock()
|
|
repository.latest_at_or_before.return_value = replace(
|
|
MATERIAL, machine_id='example-machine', production_order='ORDER/00123',
|
|
)
|
|
subject = MaterialEfficiencyService(
|
|
Mock(), repository, workplace='example', machine_id='example-machine',
|
|
calculation_id=CALC, format_template='ORDER/{production_order}',
|
|
)
|
|
result = subject.evaluate(replace(STATUS, workplace='example', production_order=' 00123 '))
|
|
assert result.enlyze_production_order == 'ORDER/00123'
|
|
|
|
|
|
def test_wrong_workplace_and_combined_order_rejected():
|
|
with pytest.raises(ValueError, match='workplace'):
|
|
service().evaluate(replace(STATUS, workplace='other'))
|
|
with pytest.raises(ValueError, match='ERP production order'):
|
|
service().evaluate(replace(STATUS, production_order='K 7-123-K 7-456'))
|
|
|
|
|
|
def test_temporal_regression_and_successive_feedback():
|
|
later = replace(
|
|
MATERIAL, timestamp=START.replace(minute=10), consumption_kg=1100, run_id='run2',
|
|
)
|
|
repository = Mock()
|
|
repository.latest_at_or_before.side_effect = lambda **kw: max(
|
|
(row for row in [MATERIAL, later] if row.timestamp <= kw['timestamp']),
|
|
key=lambda row: row.timestamp, default=None,
|
|
)
|
|
subject = service(repository)
|
|
first = subject.evaluate(STATUS)
|
|
second = subject.evaluate(replace(
|
|
STATUS, feedback_timestamp=later.timestamp, good_quantity_m2=1000,
|
|
))
|
|
assert first.material_snapshot_timestamp == START
|
|
assert first.material_consumption_kg == 1000
|
|
assert first.material_consumption_kg_per_m2 == 1.25
|
|
assert second.material_snapshot_timestamp == later.timestamp
|
|
assert second.material_consumption_kg == 1100
|
|
assert second.material_consumption_kg_per_m2 == 1.1
|
|
assert first.good_quantity_m2 == 800
|
|
|
|
|
|
def test_naive_feedback_requires_explicit_timezone_and_returns_instant():
|
|
naive = datetime(2026, 9, 4, 12, 5)
|
|
status = replace(STATUS, feedback_timestamp=naive)
|
|
with pytest.raises(ValueError, match='explicit erp_timezone'):
|
|
service().evaluate(status)
|
|
subject = service(erp_timezone=ZoneInfo('Europe/Berlin'))
|
|
result = subject.evaluate(status)
|
|
assert result.erp_feedback_timestamp == FEEDBACK
|
|
assert status.feedback_timestamp == naive
|
|
assert subject.materials.latest_at_or_before.call_args.kwargs['timestamp'] == FEEDBACK
|
|
|
|
|
|
@pytest.mark.parametrize('timestamp', [datetime(2026, 3, 29, 2, 30), datetime(2026, 10, 25, 2, 30)])
|
|
def test_dst_gap_and_overlap_rejected(timestamp):
|
|
with pytest.raises(ValueError, match='ambiguous or nonexistent'):
|
|
service(erp_timezone=ZoneInfo('Europe/Berlin')).evaluate(
|
|
replace(STATUS, feedback_timestamp=timestamp),
|
|
)
|
|
|
|
|
|
def test_aware_feedback_returns_utc():
|
|
timestamp = FEEDBACK.astimezone(ZoneInfo('Europe/Berlin'))
|
|
result = service().evaluate(replace(STATUS, feedback_timestamp=timestamp))
|
|
assert result.erp_feedback_timestamp == timestamp
|
|
assert result.erp_feedback_timestamp.tzinfo is UTC
|
|
|
|
|
|
def test_database_failure_propagates():
|
|
subject = service()
|
|
subject.materials.latest_at_or_before.side_effect = RuntimeError('unavailable')
|
|
with pytest.raises(RuntimeError, match='unavailable'):
|
|
subject.evaluate(STATUS)
|