""" Comprehensive tests for all articles. Tests every article (grey_k1_from_DBPD) for valid data and predictions. """ import pytest import pandas as pd import sys import os from datetime import datetime import json sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) class TestAllArticles: """Test suite that verifies ALL articles.""" def test_all_articles_exist(self, data_service, raw_excel_df): """Verify all articles can be queried.""" articles = raw_excel_df["OCDKE1"].dropna().unique().tolist() results = {"total_tested": 0, "found": 0, "not_found": 0, "errors": []} for article_id in articles: results["total_tested"] += 1 try: result = data_service.get_article_insights(str(article_id)) if "error" in result: results["not_found"] += 1 else: results["found"] += 1 # Verify basic structure assert "dna" in result assert "count" in result assert "data" in result except Exception as e: results["errors"].append({"article": str(article_id), "error": str(e)}) # Save results report_path = os.path.join( os.path.dirname(__file__), "reports", "articles_report.json" ) os.makedirs(os.path.dirname(report_path), exist_ok=True) with open(report_path, "w") as f: json.dump(results, f, indent=2) # At least 90% should be found found_rate = ( results["found"] / results["total_tested"] * 100 if results["total_tested"] > 0 else 0 ) assert found_rate >= 90.0, ( f"Too many articles not found: {results['not_found']}/{results['total_tested']}" ) def test_article_dna_structure(self, data_service, raw_excel_df): """Verify article DNA contains expected fields.""" articles = raw_excel_df["OCDKE1"].dropna().unique().tolist()[:50] required_dna_fields = [ "Article", "Count", "Product", "Standard_Route", "Base_Finish_Example", ] for article_id in articles: result = data_service.get_article_insights(str(article_id)) if "error" in result: continue for field in required_dna_fields: assert field in result["dna"], ( f"Missing DNA field {field} for article {article_id}" ) def test_article_data_has_required_columns(self, data_service, raw_excel_df): """Verify article data contains required columns.""" articles = raw_excel_df["OCDKE1"].dropna().unique().tolist()[:20] required_columns = [ "PO_NO", "Order Qty", "Reserver Qty as per Std Norms", "Actual Gr Opening", "Deviation", "Finish", "Route", ] for article_id in articles: result = data_service.get_article_insights(str(article_id)) if "error" in result: continue if result["data"]: first_row = result["data"][0] for col in required_columns: assert col in first_row, ( f"Missing column {col} in article data for {article_id}" ) def test_article_count_matches_data_length(self, data_service, raw_excel_df): """Verify article count matches number of data rows.""" articles = raw_excel_df["OCDKE1"].dropna().unique().tolist()[:30] for article_id in articles: result = data_service.get_article_insights(str(article_id)) if "error" in result: continue assert result["count"] == len(result["data"]), ( f"Count mismatch for article {article_id}" ) class TestArticlePredictions: """Test suite for article prediction functionality.""" def test_article_predictions_structure(self, data_service, raw_excel_df): """Verify article predictions have correct structure.""" articles = raw_excel_df["OCDKE1"].dropna().unique().tolist()[:20] for article_id in articles: try: result = data_service.get_article_predictions(str(article_id)) if result is None: continue # Verify structure assert "article_id" in result assert "details" in result assert "stats" in result assert "ai_prediction" in result assert "orders" in result # Verify ai_prediction structure pred = result["ai_prediction"] assert "historical_orders" in pred assert "yield_stats" in pred assert "recommendation" in pred assert "confidence" in pred except Exception as e: pass # Some articles may not have predictions def test_article_yield_stats_reasonable(self, data_service, raw_excel_df): """Verify yield stats are within reasonable ranges.""" articles = raw_excel_df["OCDKE1"].dropna().unique().tolist()[:30] for article_id in articles: try: result = data_service.get_article_predictions(str(article_id)) if result is None: continue yield_stats = result["ai_prediction"]["yield_stats"] # Yield should be between 0 and 200 (allowing for edge cases) assert 0 <= yield_stats["avg"] <= 200, ( f"Unreasonable yield avg for {article_id}" ) assert 0 <= yield_stats["min"] <= 200, ( f"Unreasonable yield min for {article_id}" ) assert 0 <= yield_stats["max"] <= 200, ( f"Unreasonable yield max for {article_id}" ) except Exception as e: pass def test_article_recommendation_reasonable(self, data_service, raw_excel_df): """Verify recommendation values are reasonable.""" articles = raw_excel_df["OCDKE1"].dropna().unique().tolist()[:30] for article_id in articles: try: result = data_service.get_article_predictions(str(article_id)) if result is None: continue rec = result["ai_prediction"]["recommendation"] # Suggested reservation should be between -50% and +50% assert -50 <= rec["suggested_reservation_pct"] <= 50, ( f"Unreasonable reservation suggestion for {article_id}" ) except Exception as e: pass def test_article_confidence_levels(self, data_service, raw_excel_df): """Verify confidence levels are valid.""" articles = raw_excel_df["OCDKE1"].dropna().unique().tolist()[:30] valid_confidence = ["high", "medium", "low"] for article_id in articles: try: result = data_service.get_article_predictions(str(article_id)) if result is None: continue confidence = result["ai_prediction"]["confidence"] assert confidence in valid_confidence, ( f"Invalid confidence level for {article_id}" ) except Exception as e: pass class TestArticleAggregation: """Test suite for article-level aggregation logic.""" def test_article_total_volume_matches(self, data_service, raw_excel_df): """Verify article total volume matches sum of orders.""" articles = raw_excel_df["OCDKE1"].dropna().unique().tolist()[:20] for article_id in articles: try: result = data_service.get_article_predictions(str(article_id)) if result is None: continue # Verify stats stats = result["stats"] # Total volume should be positive assert stats["total_volume"] >= 0, f"Negative volume for {article_id}" # Total orders should match orders list assert stats["total_orders"] == len(result["orders"]), ( f"Order count mismatch for {article_id}" ) except Exception as e: pass def test_article_orders_have_required_fields(self, data_service, raw_excel_df): """Verify each order in article has required fields.""" articles = raw_excel_df["OCDKE1"].dropna().unique().tolist()[:20] required_fields = ["id", "volume", "input", "output", "yield"] for article_id in articles: try: result = data_service.get_article_predictions(str(article_id)) if result is None or not result["orders"]: continue for order in result["orders"][:5]: # Check first 5 orders for field in required_fields: assert field in order, ( f"Missing field {field} in order for {article_id}" ) except Exception as e: pass