Spaces:
Sleeping
Sleeping
| """ | |
| Tests for data loading functionality. | |
| Verifies that data is loaded correctly and all columns are mapped properly. | |
| """ | |
| import pytest | |
| import pandas as pd | |
| import sys | |
| import os | |
| sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) | |
| from conftest import ManualCalculator | |
| class TestDataLoading: | |
| """Test suite for data loading verification.""" | |
| def test_data_service_loads_successfully(self, data_service): | |
| """Verify data service loads without errors.""" | |
| assert data_service.is_loaded is True | |
| assert data_service.master_df is not None | |
| def test_master_df_has_correct_columns(self, data_service): | |
| """Verify all required columns exist in master_df.""" | |
| required_columns = [ | |
| "PO_NO", | |
| "COPS_NO", | |
| "DORQT1", | |
| "RES_QTY", | |
| "ISS_QTY", | |
| "pack_fresh", | |
| "pack_qty", | |
| "Order Qty", | |
| "Actual Gr Opening", | |
| "Reserver Qty as per Std Norms", | |
| "Deviation", | |
| "Deviation_Percent", | |
| "Article", | |
| "Sale Order", | |
| "Finish", | |
| "Route", | |
| "is_input", | |
| "is_output", | |
| ] | |
| for col in required_columns: | |
| assert col in data_service.master_df.columns, f"Missing column: {col}" | |
| def test_master_df_has_data(self, data_service): | |
| """Verify master_df contains expected number of rows.""" | |
| # Original file has 4613 rows | |
| assert len(data_service.master_df) > 4000, "Too few rows loaded" | |
| def test_po_type_flags_set_correctly(self, data_service, po_type_map): | |
| """Verify is_input and is_output flags match PO Type mapping.""" | |
| df = data_service.master_df | |
| # Sample check: F0U should be input=True, output=True | |
| f0u_rows = df[df["PO_CODE"] == "F0U"] | |
| if len(f0u_rows) > 0: | |
| assert all(f0u_rows["is_input"] == True), "F0U should have is_input=True" | |
| assert all(f0u_rows["is_output"] == True), "F0U should have is_output=True" | |
| # F0N should be input=False, output=False | |
| 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" | |
| ) | |
| # FRG (Reprocess) should be input=False, output=True | |
| 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" | |
| def test_numeric_columns_are_numeric(self, data_service): | |
| """Verify numeric columns have correct data types.""" | |
| df = data_service.master_df | |
| numeric_cols = [ | |
| "DORQT1", | |
| "RES_QTY", | |
| "ISS_QTY", | |
| "pack_fresh", | |
| "pack_qty", | |
| "Order Qty", | |
| "Actual Gr Opening", | |
| "Reserver Qty as per Std Norms", | |
| ] | |
| for col in numeric_cols: | |
| assert pd.api.types.is_numeric_dtype(df[col]), f"{col} should be numeric" | |
| def test_deviation_calculated_correctly(self, data_service): | |
| """Verify Deviation column = ISS_QTY - RES_QTY.""" | |
| df = data_service.master_df | |
| sample = df.head(100) | |
| for idx, row in sample.iterrows(): | |
| expected = row["ISS_QTY"] - row["RES_QTY"] | |
| actual = row["Deviation"] | |
| assert abs(expected - actual) < 0.01, f"Deviation mismatch at {idx}" | |
| def test_deviation_percent_calculated_correctly(self, data_service): | |
| """Verify Deviation_Percent = (Deviation / RES_QTY) * 100.""" | |
| df = data_service.master_df | |
| sample = df.head(100) | |
| for idx, row in sample.iterrows(): | |
| if row["RES_QTY"] > 0: | |
| expected = (row["Deviation"] / row["RES_QTY"]) * 100 | |
| actual = row["Deviation_Percent"] | |
| assert abs(expected - actual) < 0.1, ( | |
| f"Deviation_Percent mismatch at {idx}" | |
| ) | |
| def test_article_column_mapped(self, data_service): | |
| """Verify Article column is mapped from grey_k1_from_DBPD.""" | |
| df = data_service.master_df | |
| # Check that Article column has values | |
| non_null = df["Article"].notna().sum() | |
| assert non_null > 4000, "Too many null Article values" | |
| def test_sale_order_column_mapped(self, data_service): | |
| """Verify Sale Order column is mapped from COPS_NO.""" | |
| df = data_service.master_df | |
| # Check unique sale orders | |
| unique_orders = df["Sale Order"].nunique() | |
| assert unique_orders > 900, f"Expected ~970 sale orders, got {unique_orders}" | |
| def test_finish_column_exists(self, data_service): | |
| """Verify Finish column is properly created.""" | |
| df = data_service.master_df | |
| # Check that Finish column has values | |
| assert "Finish" in df.columns | |
| # Check expected values | |
| unique_finishes = df["Finish"].unique() | |
| # Should have values like 'Soft', 'Peach', etc. | |
| assert len(unique_finishes) > 0 | |
| class TestDataConsistency: | |
| """Test suite for data consistency checks.""" | |
| def test_no_duplicate_columns(self, data_service): | |
| """Verify no duplicate column names.""" | |
| cols = data_service.master_df.columns.tolist() | |
| assert len(cols) == len(set(cols)), "Duplicate column names found" | |
| def test_po_code_extracted_correctly(self, data_service): | |
| """Verify PO_CODE is first 3 characters of PO_NO.""" | |
| df = data_service.master_df | |
| sample = df.head(100) | |
| for idx, row in sample.iterrows(): | |
| expected = str(row["PO_NO"])[:3] | |
| actual = row["PO_CODE"] | |
| assert actual == expected, f"PO_CODE mismatch at {idx}" | |
| def test_order_qty_equals_dorqt1(self, data_service): | |
| """Verify Order Qty is mapped from DORQT1.""" | |
| df = data_service.master_df | |
| sample = df.head(100) | |
| for idx, row in sample.iterrows(): | |
| assert row["Order Qty"] == row["DORQT1"], f"Order Qty mismatch at {idx}" | |
| def test_actual_gr_opening_equals_iss_qty(self, data_service): | |
| """Verify Actual Gr Opening is mapped from ISS_QTY.""" | |
| df = data_service.master_df | |
| sample = df.head(100) | |
| for idx, row in sample.iterrows(): | |
| assert row["Actual Gr Opening"] == row["ISS_QTY"], ( | |
| f"Actual Gr Opening mismatch at {idx}" | |
| ) | |
| def test_reserved_qty_equals_res_qty(self, data_service): | |
| """Verify Reserver Qty is mapped from RES_QTY.""" | |
| df = data_service.master_df | |
| sample = df.head(100) | |
| for idx, row in sample.iterrows(): | |
| assert row["Reserver Qty as per Std Norms"] == row["RES_QTY"], ( | |
| f"Reserved Qty mismatch at {idx}" | |
| ) | |