Spaces:
Sleeping
Sleeping
| """ | |
| 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: | |
| def fixture(*args, **kwargs): | |
| def decorator(func): | |
| return func | |
| return decorator | |
| pytest = pytest_dummy() | |
| def data_service(): | |
| """Create a DataService instance and load data once for all tests.""" | |
| service = DataService() | |
| service.load_data() | |
| return service | |
| def raw_excel_df(): | |
| """Load raw Excel data for manual verification.""" | |
| df = pd.read_excel(DATA_PATH, sheet_name="Detail", header=2) | |
| return df | |
| def po_type_df(): | |
| """Load PO Type lookup table.""" | |
| return pd.read_excel(DATA_PATH, sheet_name="PO Type") | |
| def all_sale_orders(raw_excel_df): | |
| """Get list of all unique sale orders.""" | |
| return raw_excel_df["COPS_NO"].unique().tolist() | |
| 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] | |
| 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") | |
| 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 %) | |
| """ | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |