process-aware-ai / backend /tests /conftest.py
borndeveloper's picture
Fixed terms
8790d88
Raw
History Blame Contribute Delete
4.2 kB
"""
Pytest configuration and fixtures for data service tests.
"""
import pandas as pd
import sys
import os
from datetime import datetime
# Add parent directory to path for imports
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from app.services.data_service import DataService, DATA_PATH
# pytest is optional - only needed for pytest-based tests
try:
import pytest
PYTEST_AVAILABLE = True
except ImportError:
PYTEST_AVAILABLE = False
# Create dummy pytest.fixture decorator
class pytest_dummy:
@staticmethod
def fixture(*args, **kwargs):
def decorator(func):
return func
return decorator
pytest = pytest_dummy()
@pytest.fixture(scope="session")
def data_service():
"""Create a DataService instance and load data once for all tests."""
service = DataService()
service.load_data()
return service
@pytest.fixture(scope="session")
def raw_excel_df():
"""Load raw Excel data for manual verification."""
df = pd.read_excel(DATA_PATH, sheet_name="Detail", header=2)
return df
@pytest.fixture(scope="session")
def po_type_df():
"""Load PO Type lookup table."""
return pd.read_excel(DATA_PATH, sheet_name="PO Type")
@pytest.fixture(scope="session")
def all_sale_orders(raw_excel_df):
"""Get list of all unique sale orders."""
return raw_excel_df["COPS_NO"].unique().tolist()
@pytest.fixture(scope="session")
def all_articles(raw_excel_df):
"""Get list of all unique articles."""
articles = raw_excel_df["OCDKE1"].dropna().unique().tolist() if "OCDKE1" in raw_excel_df.columns else raw_excel_df["grey_k1_from_DBPD"].dropna().unique().tolist()
return [str(a) for a in articles]
@pytest.fixture(scope="session")
def po_type_map(po_type_df):
"""Create PO type to is_input/is_output mapping."""
po_type_df.columns = [c.strip() for c in po_type_df.columns]
po_type_df["is_input"] = (
po_type_df.iloc[:, 2].astype(str).str.upper().apply(lambda x: "YES" in x)
)
po_type_df["is_output"] = (
po_type_df.iloc[:, 3].astype(str).str.upper().apply(lambda x: "YES" in x)
)
return po_type_df.set_index(po_type_df.columns[0])[
["is_input", "is_output"]
].to_dict("index")
@pytest.fixture
def test_results():
"""Fixture to store test results for reporting."""
return {
"timestamp": datetime.now().isoformat(),
"sale_orders": {"passed": 0, "failed": 0, "errors": []},
"articles": {"passed": 0, "failed": 0, "errors": []},
"calculations": {"passed": 0, "failed": 0, "errors": []},
"edge_cases": {"passed": 0, "failed": 0, "errors": []},
}
class ManualCalculator:
"""
Manual calculation class to verify formulas against Excel logic.
Formulas from Excel "Eg, Calculation" sheet:
- G18: =(G14-G13)/G13 (Reserved vs PO %)
- G19: =(G15-G13)/G13 (Gr Opening vs PO %)
- G20: =(G15-G16)/G15 (Loss %)
- G21: =G17/G16 (Fresh Packing %)
- G22: =G17/G13 (Yield %)
"""
@staticmethod
def extra_gr_reserved_pct(reserved_qty, po_qty):
"""(Reserved - PO_Qty) / PO_Qty × 100"""
if po_qty == 0:
return 0.0
return (reserved_qty - po_qty) / po_qty * 100
@staticmethod
def actual_gr_issue_pct(issued_qty, po_qty):
"""(Issued - PO_Qty) / PO_Qty × 100"""
if po_qty == 0:
return 0.0
return (issued_qty - po_qty) / po_qty * 100
@staticmethod
def shrinkage_pct(issued_qty, total_packing):
"""(Issued - Total Packing) / Issued × 100"""
if issued_qty == 0:
return 0.0
return (issued_qty - total_packing) / issued_qty * 100
@staticmethod
def fresh_pkg_pct(pack_fresh, total_packing):
"""Pack Fresh / Total Packing × 100"""
if total_packing == 0:
return 0.0
return pack_fresh / total_packing * 100
@staticmethod
def fresh_to_order_pct(pack_fresh, order_qty):
"""Pack Fresh / Order Qty × 100"""
if order_qty == 0:
return 0.0
return pack_fresh / order_qty * 100