"""Comprehensive validation test suite for V2 extraction pipeline. This test suite processes all 58 Excel files in CustomerDocument/ folder and validates: 1. V2 extraction success with V1 fallback 2. Overall success rate meets 95% target 3. Metrics tracking and reporting """ import pytest from pathlib import Path from typing import List import structlog from src.extraction_v2.pipeline import ExtractionPipelineV2 from src.extraction.pipeline import ExtractionPipeline logger = structlog.get_logger() class MetricsTracker: """Track extraction success metrics across all test runs.""" def __init__(self): self.v2_success: List[Path] = [] self.v1_fallback: List[tuple[Path, str]] = [] self.failed: List[tuple[Path, str]] = [] def record_success(self, version: str, file_path: Path): """Record successful extraction.""" if version == 'v2': self.v2_success.append(file_path) elif version == 'v1': # V1 succeeded after V2 failed pass # Already recorded in v1_fallback def record_fallback(self, file_path: Path, reason: str): """Record V2 failure that triggered V1 fallback.""" self.v1_fallback.append((file_path, reason)) def record_failure(self, file_path: Path, reason: str): """Record complete failure (both V2 and V1 failed).""" self.failed.append((file_path, reason)) def calculate_success_rate(self, total_files: int) -> float: """Calculate overall success rate (V2 + V1 combined).""" # Success includes both V2 direct success and V1 fallback success v1_fallback_success = sum(1 for _, r in self.v1_fallback if "V1 succeeded" in r or "fallback" in r.lower()) total_success = len(self.v2_success) + v1_fallback_success return (total_success / total_files) * 100 if total_files > 0 else 0.0 def generate_report(self) -> dict: """Generate comprehensive metrics report.""" v1_fallback_success = [f for f, r in self.v1_fallback if "succeeded" in r.lower()] total = len(self.v2_success) + len(v1_fallback_success) + len(self.failed) return { "total_files": total, "v2_success": len(self.v2_success), "v1_fallback": len(v1_fallback_success), "failed": len(self.failed), "success_rate": self.calculate_success_rate(total), "v2_success_rate": (len(self.v2_success) / total * 100) if total > 0 else 0.0, "failure_details": [ {"file": f.name, "reason": r} for f, r in self.failed ], "fallback_details": [ {"file": f.name, "reason": r} for f, r in self.v1_fallback ] } @pytest.fixture(scope="session") def customer_document_files(): """Discover all Excel files in CustomerDocument/ folder.""" folder = Path("CustomerDocument/") if not folder.exists(): pytest.skip("CustomerDocument/ folder not found") patterns = ["**/*.xlsx", "**/*.xls", "**/*.xlsm"] files = [] for pattern in patterns: files.extend(folder.glob(pattern)) return sorted(files) @pytest.fixture(scope="session") def metrics_tracker(): """Session-scoped metrics tracker for all tests.""" return MetricsTracker() @pytest.fixture def v2_pipeline(): """V2 extraction pipeline instance.""" return ExtractionPipelineV2() @pytest.fixture def v1_pipeline(): """V1 extraction pipeline for fallback.""" return ExtractionPipeline() @pytest.mark.parametrize("file_path", [ pytest.param(f, id=f.name) for f in sorted(Path("CustomerDocument/").rglob("*.xl*")) if f.is_file() ]) async def test_extract_file_v2_with_fallback( file_path: Path, v2_pipeline, v1_pipeline, metrics_tracker ): """Test V2 extraction with V1 fallback for each file. This test: 1. Attempts V2 extraction first 2. Falls back to V1 if V2 fails 3. Tracks metrics for reporting 4. Passes if either V2 or V1 succeeds """ logger.info("Testing extraction", file=file_path.name) # Try V2 first try: result = await v2_pipeline.extract_from_file(str(file_path)) if result and len(result) > 0: metrics_tracker.record_success('v2', file_path) logger.info("V2 extraction successful", file=file_path.name, records=len(result)) assert True, f"V2 extraction successful: {len(result)} records" return else: logger.warning("V2 returned empty result", file=file_path.name) metrics_tracker.record_fallback(file_path, "Empty result from V2") except Exception as e: logger.warning("V2 extraction failed", file=file_path.name, error=str(e)) metrics_tracker.record_fallback(file_path, f"V2 error: {str(e)[:100]}") # Fallback to V1 try: logger.info("Falling back to V1", file=file_path.name) result_v1 = await v1_pipeline.extract_from_file(str(file_path)) if result_v1 and len(result_v1) > 0: # Record successful V1 fallback metrics_tracker.record_fallback(file_path, f"V1 succeeded with {len(result_v1)} records") logger.info("V1 fallback successful", file=file_path.name, records=len(result_v1)) assert True, f"V1 fallback successful: {len(result_v1)} records" else: metrics_tracker.record_failure(file_path, "Both V2 and V1 returned empty results") pytest.fail(f"Both V2 and V1 extraction failed for {file_path.name}") except Exception as e: metrics_tracker.record_failure(file_path, f"V1 fallback error: {str(e)[:100]}") pytest.fail(f"V1 fallback failed: {str(e)}") def test_overall_success_rate(metrics_tracker, customer_document_files): """Validate that overall success rate meets 95% target. This test runs after all parametrized tests and validates: 1. All 58 files were processed 2. Success rate >= 95% 3. Generates validation report """ total_files = len(customer_document_files) report = metrics_tracker.generate_report() logger.info("Comprehensive validation results", **report) # Save report to JSON import json with open("validation_report.json", "w") as f: json.dump(report, f, indent=2, default=str) logger.info("Validation report saved", path="validation_report.json") # Assert 95% success rate assert report["success_rate"] >= 95.0, ( f"Success rate {report['success_rate']:.1f}% is below 95% target. " f"Failed files: {len(report['failure_details'])}. " f"Check validation_report.json for details." ) # Assert all 58 files processed assert total_files == 58, f"Expected 58 files, found {total_files}" logger.info( "SUCCESS: All acceptance criteria met", total=total_files, success_rate=f"{report['success_rate']:.1f}%" )