process-aware-ai / backend /tests /test_edge_cases.py
borndeveloper's picture
Fixed terms
8790d88
Raw
History Blame Contribute Delete
10.2 kB
"""
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"
)