174 lines
6.0 KiB
Python
174 lines
6.0 KiB
Python
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _make_pg_factory(session_mock: MagicMock) -> MagicMock:
|
|
factory = MagicMock()
|
|
factory.return_value.__enter__ = MagicMock(return_value=session_mock)
|
|
factory.return_value.__exit__ = MagicMock(return_value=False)
|
|
return factory
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# AW persistence layer
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestAWPersist:
|
|
def test_persist_forecast_writes_record(self) -> None:
|
|
from app.domain.aw import analytics
|
|
|
|
session = MagicMock()
|
|
factory = _make_pg_factory(session)
|
|
|
|
data = [
|
|
{"date": "2025-01-01", "predicted_revenue": 1000.0,
|
|
"lower_bound": 900.0, "upper_bound": 1100.0},
|
|
]
|
|
|
|
with patch.object(analytics, "append_audit"):
|
|
analytics.persist_forecast(factory, data, horizon_days=30, trigger_source="test")
|
|
|
|
session.add.assert_called_once()
|
|
model = session.add.call_args[0][0]
|
|
assert model.horizon_days == 30
|
|
assert model.point_count == 1
|
|
|
|
def test_persist_rep_scores_writes_record(self) -> None:
|
|
from app.domain.aw import analytics
|
|
|
|
session = MagicMock()
|
|
factory = _make_pg_factory(session)
|
|
|
|
data = [
|
|
{"rep_name": "Alice", "total_revenue": 100_000.0,
|
|
"total_orders": 50, "performance_score": 0.92},
|
|
]
|
|
|
|
with patch.object(analytics, "append_audit"):
|
|
analytics.persist_rep_scores(factory, data, top_n=10, trigger_source="test")
|
|
|
|
session.add.assert_called_once()
|
|
model = session.add.call_args[0][0]
|
|
assert model.rep_count == 1
|
|
|
|
def test_persist_product_demand_writes_record(self) -> None:
|
|
from app.domain.aw import analytics
|
|
|
|
session = MagicMock()
|
|
factory = _make_pg_factory(session)
|
|
|
|
data = [{"product_name": "Widget", "total_quantity": 500.0}]
|
|
|
|
with patch.object(analytics, "append_audit"):
|
|
analytics.persist_product_demand(factory, data, top_n=10, trigger_source="test")
|
|
|
|
session.add.assert_called_once()
|
|
|
|
def test_persist_anomaly_run_writes_record(self) -> None:
|
|
from app.domain.aw import analytics
|
|
|
|
session = MagicMock()
|
|
factory = _make_pg_factory(session)
|
|
|
|
data = [
|
|
{"is_anomaly": True, "date": "2025-01-01", "revenue": 50.0},
|
|
{"is_anomaly": False, "date": "2025-01-02", "revenue": 1000.0},
|
|
]
|
|
|
|
with patch.object(analytics, "append_audit"):
|
|
analytics.persist_anomaly_run(factory, data, trigger_source="test")
|
|
|
|
session.add.assert_called_once()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# WWI persistence layer
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestWWIPersist:
|
|
def test_persist_reorder_recommendations_writes_record(self) -> None:
|
|
from app.domain.wwi import analytics
|
|
|
|
session = MagicMock()
|
|
factory = _make_pg_factory(session)
|
|
|
|
data = [
|
|
{"stock_item_key": 1, "stock_item_name": "Widget",
|
|
"current_stock": 10.0, "urgency": "HIGH"},
|
|
]
|
|
|
|
with patch.object(analytics, "append_audit"):
|
|
analytics.persist_reorder_recommendations(factory, data, trigger_source="test")
|
|
|
|
session.add.assert_called_once()
|
|
model = session.add.call_args[0][0]
|
|
assert model.item_count == 1
|
|
|
|
def test_persist_supplier_scores_writes_record(self) -> None:
|
|
from app.domain.wwi import analytics
|
|
|
|
session = MagicMock()
|
|
factory = _make_pg_factory(session)
|
|
|
|
data = [
|
|
{"supplier_name": "Acme", "performance_score": 0.87},
|
|
]
|
|
|
|
with patch.object(analytics, "append_audit"):
|
|
analytics.persist_supplier_scores(factory, data, top_n=10, trigger_source="test")
|
|
|
|
session.add.assert_called_once()
|
|
model = session.add.call_args[0][0]
|
|
assert model.supplier_count == 1
|
|
|
|
def test_generate_stock_events_skips_non_high_urgency(self) -> None:
|
|
from app.domain.wwi import analytics
|
|
from app.domain.wwi.models import WWIBusinessEvent
|
|
|
|
session = MagicMock()
|
|
session.query.return_value.filter.return_value.first.return_value = None
|
|
factory = _make_pg_factory(session)
|
|
|
|
items = [
|
|
{"stock_item_key": 1, "stock_item_name": "Widget", "urgency": "LOW",
|
|
"current_stock": 100.0, "avg_daily_demand": 5.0,
|
|
"days_until_stockout": 20.0, "recommended_reorder_qty": 50},
|
|
{"stock_item_key": 2, "stock_item_name": "Gadget", "urgency": "MEDIUM",
|
|
"current_stock": 50.0, "avg_daily_demand": 3.0,
|
|
"days_until_stockout": 16.0, "recommended_reorder_qty": 30},
|
|
]
|
|
|
|
analytics.generate_stock_events(factory, items)
|
|
|
|
# No events should have been added since neither item is HIGH urgency
|
|
session.add.assert_not_called()
|
|
|
|
def test_generate_stock_events_creates_event_for_high_urgency(self) -> None:
|
|
from app.domain.wwi import analytics
|
|
|
|
session = MagicMock()
|
|
# Simulate no existing event in the 24h window
|
|
session.query.return_value.filter.return_value.first.return_value = None
|
|
factory = _make_pg_factory(session)
|
|
|
|
items = [
|
|
{"stock_item_key": 42, "stock_item_name": "Critical Part", "urgency": "HIGH",
|
|
"current_stock": 2.0, "avg_daily_demand": 5.0,
|
|
"days_until_stockout": None, "recommended_reorder_qty": 100},
|
|
]
|
|
|
|
analytics.generate_stock_events(factory, items)
|
|
|
|
session.add.assert_called_once()
|
|
event = session.add.call_args[0][0]
|
|
assert event.event_type == "LOW_STOCK"
|
|
assert event.entity_key == "42"
|
|
assert "immediately" in event.message
|