Spaces:
Sleeping
Sleeping
| """ | |
| Tests for edge cases. | |
| Verifies handling of zero values, negative deviations, impossible yields, etc. | |
| """ | |
| import pytest | |
| import pandas as pd | |
| import sys | |
| import os | |
| sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) | |
| class TestEdgeCases: | |
| """Test suite for edge case handling.""" | |
| def test_zero_order_qty_handled(self, data_service, raw_excel_df): | |
| """Verify zero order qty doesn't cause division errors.""" | |
| # Find orders with zero order qty | |
| zero_orders = raw_excel_df[raw_excel_df["DORQT1"] == 0]["COPS_NO"].unique() | |
| for so_id in zero_orders[:5]: | |
| try: | |
| result = data_service.get_sale_order_details(so_id) | |
| # Should not raise an error | |
| assert "error" not in result or result.get("error") == "Order not found" | |
| except ZeroDivisionError: | |
| pytest.fail(f"ZeroDivisionError for order with zero qty: {so_id}") | |
| def test_zero_issuance_handled(self, data_service, raw_excel_df): | |
| """Verify zero issuance doesn't cause division errors.""" | |
| # Find orders with zero issuance | |
| zero_iss = raw_excel_df[raw_excel_df["ISS_QTY"] == 0]["COPS_NO"].unique() | |
| for so_id in zero_iss[:5]: | |
| try: | |
| result = data_service.get_sale_order_details(so_id) | |
| assert "error" not in result or result.get("error") == "Order not found" | |
| except ZeroDivisionError: | |
| pytest.fail(f"ZeroDivisionError for order with zero issuance: {so_id}") | |
| def test_zero_pack_fresh_handled(self, data_service, raw_excel_df): | |
| """Verify zero pack_fresh doesn't cause errors.""" | |
| zero_pack = raw_excel_df[raw_excel_df["pack_fresh"] == 0]["COPS_NO"].unique() | |
| for so_id in zero_pack[:5]: | |
| try: | |
| result = data_service.get_sale_order_details(so_id) | |
| if "error" not in result: | |
| pass | |
| except Exception as e: | |
| pytest.fail(f"Error for order with zero pack_fresh {so_id}: {e}") | |
| def test_under_issuance_negative_deviation(self, data_service, raw_excel_df): | |
| """Verify under-issuance (ISS_QTY < RES_QTY) produces negative deviation.""" | |
| under_issued = raw_excel_df[raw_excel_df["ISS_QTY"] < raw_excel_df["RES_QTY"]] | |
| if len(under_issued) > 0: | |
| sample = under_issued.iloc[0] | |
| so_id = sample["COPS_NO"] | |
| result = data_service.get_sale_order_details(so_id) | |
| if "error" not in result: | |
| # Deviation should be negative for under-issuance | |
| total_issued = result["metrics"]["Actual Issued"] | |
| total_reserved = result["metrics"]["Reserved Qty"] | |
| # For this specific PO, check deviation | |
| deviation = total_issued - total_reserved | |
| # Note: This is at order level, so might not always be negative | |
| def test_over_issuance_positive_deviation(self, data_service, raw_excel_df): | |
| """Verify over-issuance (ISS_QTY > RES_QTY) produces positive deviation.""" | |
| over_issued = raw_excel_df[raw_excel_df["ISS_QTY"] > raw_excel_df["RES_QTY"]] | |
| if len(over_issued) > 0: | |
| sample = over_issued.iloc[0] | |
| so_id = sample["COPS_NO"] | |
| result = data_service.get_sale_order_details(so_id) | |
| if "error" not in result: | |
| # At least some metrics should indicate over-issuance | |
| actual_gr_issue = result["metrics"]["Actual Gr Issue %"] | |
| # Should be positive if over-issued | |
| def test_impossible_yield_over_100(self, data_service, raw_excel_df): | |
| """Verify yield > 100% is handled (possible with reprocess data).""" | |
| # Find orders where pack_fresh > ISS_QTY | |
| impossible = raw_excel_df[raw_excel_df["pack_fresh"] > raw_excel_df["ISS_QTY"]] | |
| for idx, row in impossible.head(5).iterrows(): | |
| so_id = row["COPS_NO"] | |
| try: | |
| result = data_service.get_sale_order_details(so_id) | |
| if "error" not in result: | |
| pass | |
| except Exception as e: | |
| pytest.fail(f"Error handling yield > 100%: {e}") | |
| def test_multiple_po_types_in_order(self, data_service, raw_excel_df): | |
| """Verify orders with multiple PO types are handled correctly.""" | |
| # Find orders with multiple POs | |
| so_counts = raw_excel_df.groupby("COPS_NO")["PO_NO"].nunique() | |
| multi_po_sos = so_counts[so_counts > 2].index.tolist() | |
| for so_id in multi_po_sos[:5]: | |
| try: | |
| result = data_service.get_sale_order_details(so_id) | |
| if "error" not in result: | |
| # Verify PO breakdown exists | |
| assert "po_breakdown" in result | |
| assert len(result["po_breakdown"]) > 1 | |
| except Exception as e: | |
| pytest.fail(f"Error handling multi-PO order {so_id}: {e}") | |
| def test_order_with_reprocess(self, data_service, raw_excel_df): | |
| """Verify orders with reprocess POs are handled correctly.""" | |
| # Find orders with Reprocess PO type | |
| reprocess_orders = raw_excel_df[raw_excel_df["PO Type"] == "Reprocess"][ | |
| "COPS_NO" | |
| ].unique() | |
| for so_id in reprocess_orders[:5]: | |
| try: | |
| result = data_service.get_sale_order_details(so_id) | |
| if "error" not in result: | |
| # Should have reprocess count > 0 | |
| reprocess_count = result["metrics"].get("Reprocess Count", 0) | |
| assert reprocess_count >= 0 | |
| except Exception as e: | |
| pytest.fail(f"Error handling reprocess order {so_id}: {e}") | |
| def test_order_with_shortfall(self, data_service, raw_excel_df): | |
| """Verify orders with Short Fall PO type are handled correctly.""" | |
| shortfall_orders = raw_excel_df[raw_excel_df["PO Type"] == "Short Fall"][ | |
| "COPS_NO" | |
| ].unique() | |
| for so_id in shortfall_orders[:5]: | |
| try: | |
| result = data_service.get_sale_order_details(so_id) | |
| if "error" not in result: | |
| # Shortfall should be tracked | |
| shortfall = result["metrics"].get("Shortfall", 0) | |
| status = result["metrics"].get("Status", "") | |
| # Either Shortfall or Fulfilled | |
| assert status in ["Shortfall", "Fulfilled"] | |
| except Exception as e: | |
| pytest.fail(f"Error handling shortfall order {so_id}: {e}") | |
| def test_order_not_found(self, data_service): | |
| """Verify non-existent order returns error gracefully.""" | |
| result = data_service.get_sale_order_details("NONEXISTENT_ORDER_12345") | |
| assert "error" in result | |
| assert result["error"] == "Order not found" | |
| def test_article_not_found(self, data_service): | |
| """Verify non-existent article returns error gracefully.""" | |
| result = data_service.get_article_insights("NONEXISTENT_ARTICLE_12345") | |
| assert "error" in result | |
| assert result["error"] == "No data found" | |
| def test_single_po_order(self, data_service, raw_excel_df): | |
| """Verify orders with single PO are handled correctly.""" | |
| so_counts = raw_excel_df.groupby("COPS_NO")["PO_NO"].nunique() | |
| single_po_sos = so_counts[so_counts == 1].index.tolist() | |
| for so_id in single_po_sos[:5]: | |
| try: | |
| result = data_service.get_sale_order_details(so_id) | |
| if "error" not in result: | |
| assert len(result["po_breakdown"]) == 1 | |
| except Exception as e: | |
| pytest.fail(f"Error handling single-PO order {so_id}: {e}") | |
| class TestInputOutputClassification: | |
| """Test suite for is_input and is_output classification.""" | |
| def test_fresh_input_classification(self, data_service): | |
| """Verify Fresh Input POs have is_input=True.""" | |
| df = data_service.master_df | |
| fresh_codes = [ | |
| "F0U", | |
| "F01", | |
| "FQT", | |
| "FBT", | |
| "F0A", | |
| "F0P", | |
| "F0Q", | |
| "F0X", | |
| "F0Z", | |
| "FBY", | |
| "FFX", | |
| "FMW", | |
| "FOB", | |
| "FPT", | |
| "FPX", | |
| "FPY", | |
| ] | |
| for code in fresh_codes: | |
| rows = df[df["PO_CODE"] == code] | |
| if len(rows) > 0: | |
| assert all(rows["is_input"] == True), ( | |
| f"{code} should have is_input=True" | |
| ) | |
| def test_reprocess_output_classification(self, data_service): | |
| """Verify Reprocess POs have is_output=True but is_input=False.""" | |
| df = data_service.master_df | |
| frg_rows = df[df["PO_CODE"] == "FRG"] | |
| if len(frg_rows) > 0: | |
| assert all(frg_rows["is_input"] == False), "FRG should have is_input=False" | |
| assert all(frg_rows["is_output"] == True), "FRG should have is_output=True" | |
| frp_rows = df[df["PO_CODE"] == "FRP"] | |
| if len(frp_rows) > 0: | |
| assert all(frp_rows["is_input"] == False), "FRP should have is_input=False" | |
| assert all(frp_rows["is_output"] == True), "FRP should have is_output=True" | |
| def test_no_fresh_po_classification(self, data_service): | |
| """Verify No Fresh PO has is_input=False, is_output=False.""" | |
| df = data_service.master_df | |
| f0n_rows = df[df["PO_CODE"] == "F0N"] | |
| if len(f0n_rows) > 0: | |
| assert all(f0n_rows["is_input"] == False), "F0N should have is_input=False" | |
| assert all(f0n_rows["is_output"] == False), ( | |
| "F0N should have is_output=False" | |
| ) | |
| def test_short_fall_classification(self, data_service): | |
| """Verify Short Fall has is_input=False, is_output=False.""" | |
| df = data_service.master_df | |
| f0s_rows = df[df["PO_CODE"] == "F0S"] | |
| if len(f0s_rows) > 0: | |
| assert all(f0s_rows["is_input"] == False), "F0S should have is_input=False" | |
| assert all(f0s_rows["is_output"] == False), ( | |
| "F0S should have is_output=False" | |
| ) | |