""" test_enhanced_stance_system.py ============================== Comprehensive test suite for the enhanced stance classification system. This script tests all the improvements made to the stance classification system: 1. Enhanced text preprocessing 2. Multi-template NLI layer 3. Domain-specific semantic rules 4. Quality-weighted aggregation 5. Confidence-based decision making """ import logging import time from typing import List, Dict, Any import json # Set up logging logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s — %(message)s" ) logger = logging.getLogger("test_enhanced_stance") def test_enhanced_stance_classifier(): """Test the enhanced stance classifier with various claim types.""" print("=" * 60) print("TESTING ENHANCED STANCE CLASSIFIER") print("=" * 60) try: from enhanced_stance_classifier import EnhancedStanceClassifier classifier = EnhancedStanceClassifier() # Test cases covering different domains test_cases = [ { "claim": "COVID-19 vaccines cause autism", "title": "CDC Study on Vaccine Safety", "text": "A comprehensive study by the CDC involving over 95,000 children found no evidence linking COVID-19 vaccines to autism. The research, published in a peer-reviewed journal, debunked claims about vaccine-autism connections. Clinical trials and post-market surveillance data consistently show vaccines are safe and effective.", "source_reliability": 0.9, "expected_stance": "contradicting", "domain": "health" }, { "claim": "5G towers spread coronavirus", "title": "WHO Statement on 5G and COVID-19", "text": "The World Health Organization has stated that there is no link between 5G networks and COVID-19 transmission. Viruses cannot spread through radio waves or mobile networks. COVID-19 has spread in many countries that do not have 5G networks. Scientists confirm this is scientifically impossible.", "source_reliability": 0.95, "expected_stance": "contradicting", "domain": "technology" }, { "claim": "Climate change is a natural phenomenon", "title": "IPCC Climate Report 2023", "text": "The Intergovernmental Panel on Climate Change confirms that current climate change is primarily driven by human activities. Scientific consensus shows 97% of climate scientists agree that human activities are the main cause of recent climate change. Temperature records and data analysis support this conclusion.", "source_reliability": 0.92, "expected_stance": "contradicting", "domain": "climate" }, { "claim": "Government announces new healthcare policy", "title": "Official Government Press Release", "text": "The Ministry of Health officially announced a new healthcare policy today. The press release from the government website confirms the implementation of universal healthcare coverage. Official statement from the health minister validates this policy change.", "source_reliability": 0.88, "expected_stance": "supporting", "domain": "politics" }, { "claim": "New study shows coffee reduces heart disease risk", "title": "Harvard Medical School Research", "text": "A peer-reviewed study published in the Journal of the American Heart Association found that moderate coffee consumption is associated with reduced risk of heart disease. The research involved 500,000 participants over 10 years. Clinical evidence shows coffee's antioxidants may protect cardiovascular health.", "source_reliability": 0.85, "expected_stance": "supporting", "domain": "health" } ] results = [] for i, test_case in enumerate(test_cases, 1): print(f"\nTest Case {i}: {test_case['domain'].upper()} DOMAIN") print(f"Claim: {test_case['claim']}") print(f"Expected: {test_case['expected_stance']}") start_time = time.time() stance, confidence, layer = classifier.classify_detailed( title=test_case['title'], text=test_case['text'], claim=test_case['claim'], source_reliability=test_case['source_reliability'] ) end_time = time.time() correct = stance == test_case['expected_stance'] results.append({ "test_case": i, "domain": test_case['domain'], "stance": stance, "confidence": confidence, "layer": layer, "expected": test_case['expected_stance'], "correct": correct, "processing_time": end_time - start_time }) print(f"Result: {stance} (confidence: {confidence:.3f}, layer: {layer})") print(f"Correct: {'✓' if correct else '✗'}") print(f"Processing time: {end_time - start_time:.3f}s") # Summary correct_count = sum(1 for r in results if r['correct']) total_count = len(results) accuracy = correct_count / total_count avg_confidence = sum(r['confidence'] for r in results) / total_count avg_time = sum(r['processing_time'] for r in results) / total_count print(f"\n{'='*40}") print("ENHANCED STANCE CLASSIFIER RESULTS") print(f"{'='*40}") print(f"Accuracy: {correct_count}/{total_count} ({accuracy:.1%})") print(f"Average Confidence: {avg_confidence:.3f}") print(f"Average Processing Time: {avg_time:.3f}s") # Layer usage statistics layer_usage = {} for result in results: layer = result['layer'] layer_usage[layer] = layer_usage.get(layer, 0) + 1 print(f"\nLayer Usage:") for layer, count in layer_usage.items(): print(f" {layer}: {count} ({count/total_count:.1%})") return results except ImportError as e: print(f"Enhanced stance classifier not available: {e}") return [] def test_quality_weighted_aggregator(): """Test the quality-weighted aggregation system.""" print("\n" + "=" * 60) print("TESTING QUALITY-WEIGHTED AGGREGATOR") print("=" * 60) try: from quality_weighted_aggregator import ( QualityWeightedAggregator, StanceEvidence, AggregationStrategy ) # Create test evidence evidence_list = [ StanceEvidence( stance="contradicting", confidence=0.9, source_reliability=0.95, source_type="official", method="semantic", text_length=500, claim_relevance=0.95 ), StanceEvidence( stance="contradicting", confidence=0.8, source_reliability=0.85, source_type="academic", method="nli", text_length=800, claim_relevance=0.9 ), StanceEvidence( stance="supporting", confidence=0.6, source_reliability=0.4, source_type="blog", method="keyword", text_length=200, claim_relevance=0.7 ), StanceEvidence( stance="neutral", confidence=0.5, source_reliability=0.6, source_type="news", method="fallback", text_length=300, claim_relevance=0.8 ) ] # Test different aggregation strategies strategies = [ AggregationStrategy.WEIGHTED_MAJORITY, AggregationStrategy.CONFIDENCE_THRESHOLD, AggregationStrategy.RELIABILITY_WEIGHTED, AggregationStrategy.BAYESIAN_UPDATE, AggregationStrategy.ENSEMBLE ] results = {} for strategy in strategies: aggregator = QualityWeightedAggregator(strategy=strategy) start_time = time.time() stance, confidence, metadata = aggregator.aggregate_stance_evidence( evidence_list, claim_text="Test claim about vaccine safety" ) end_time = time.time() # Assess uncertainty uncertainty = aggregator.assess_uncertainty(evidence_list) results[strategy.value] = { "stance": stance, "confidence": confidence, "metadata": metadata, "uncertainty": uncertainty, "processing_time": end_time - start_time } print(f"\n{strategy.value.upper()}:") print(f" Stance: {stance}") print(f" Confidence: {confidence:.3f}") print(f" High Uncertainty: {uncertainty['high_uncertainty']}") print(f" Processing Time: {end_time - start_time:.4f}s") # Compare results print(f"\n{'='*40}") print("AGGREGATION STRATEGY COMPARISON") print(f"{'='*40}") for strategy_name, result in results.items(): print(f"{strategy_name:20} | {result['stance']:12} | {result['confidence']:.3f} | {result['processing_time']:.4f}s") return results except ImportError as e: print(f"Quality-weighted aggregator not available: {e}") return {} def test_text_processing(): """Test the enhanced text processing capabilities.""" print("\n" + "=" * 60) print("TESTING ENHANCED TEXT PROCESSING") print("=" * 60) try: from enhanced_stance_classifier import TextProcessor processor = TextProcessor() # Test text with various noise patterns noisy_text = """ This is a test article about vaccines. [Editor's note: This is important] You can read more at https://example.com/article. @DrSmith says vaccines are safe. #VaccinesSafe #HealthNews. The study (published in 2023) shows clear evidence. Multiple spaces and formatting issues here. This is another sentence. And another one! What about questions? """ print("Original text:") print(repr(noisy_text[:200] + "...")) # Test cleaning cleaned = processor.clean_text(noisy_text) print(f"\nCleaned text:") print(repr(cleaned[:200] + "...")) # Test intelligent chunking long_text = "This is a sentence. " * 50 # Create long text chunks = processor.intelligent_chunk(long_text, max_length=100, overlap=20) print(f"\nChunking test:") print(f"Original length: {len(long_text)} characters") print(f"Number of chunks: {len(chunks)}") print(f"Chunk lengths: {[len(chunk) for chunk in chunks]}") # Test entity extraction entity_text = "Dr. John Smith from Harvard Medical School published a study showing 95% effectiveness. The FDA approved this treatment in 2023." entities = processor.extract_key_entities(entity_text) print(f"\nEntity extraction test:") print(f"Text: {entity_text}") print(f"Entities: {entities}") return { "cleaned_length": len(cleaned), "original_length": len(noisy_text), "chunk_count": len(chunks), "entities_found": len(entities) } except ImportError as e: print(f"Text processor not available: {e}") return {} def test_data_collector_integration(): """Test the integration with the enhanced data collector.""" print("\n" + "=" * 60) print("TESTING DATA COLLECTOR INTEGRATION") print("=" * 60) try: from data_collector import DataCollector, CollectionRequest # Create a simple test request request = CollectionRequest( content="A post claiming that vaccines cause autism", claims=["Vaccines cause autism"], difficulty="medium" ) collector = DataCollector() print("Testing data collector with enhanced stance classification...") print(f"Claim: {request.claims[0]}") print(f"Available backends: {[b.name for b in collector._backends]}") # Note: This would require API keys to actually collect data # For testing purposes, we'll just verify the collector is properly configured print(f"Stance classifier type: {type(collector._stancer).__name__}") print(f"Has classify_detailed method: {hasattr(collector._stancer, 'classify_detailed')}") return { "collector_configured": True, "enhanced_classifier": hasattr(collector._stancer, 'classify_detailed'), "backend_count": len(collector._backends) } except ImportError as e: print(f"Data collector not available: {e}") return {} def run_performance_benchmark(): """Run performance benchmarks on the enhanced system.""" print("\n" + "=" * 60) print("PERFORMANCE BENCHMARK") print("=" * 60) try: from enhanced_stance_classifier import EnhancedStanceClassifier classifier = EnhancedStanceClassifier() # Benchmark data benchmark_cases = [ ("Short claim", "Vaccines are safe", "Short text about vaccine safety from CDC", 0.8), ("Medium claim", "Climate change is caused by human activities and poses significant risks", "A comprehensive study by climate scientists shows overwhelming evidence that human activities, particularly greenhouse gas emissions, are the primary driver of current climate change. The research, published in multiple peer-reviewed journals, demonstrates clear correlations between industrial activity and global temperature rise." * 2, 0.9), ("Long claim", "The effectiveness of various COVID-19 vaccines in preventing severe illness and hospitalization", "Multiple large-scale clinical trials and real-world studies have demonstrated the effectiveness of COVID-19 vaccines in preventing severe illness, hospitalization, and death. Data from health agencies worldwide consistently show that vaccinated individuals have significantly lower rates of severe COVID-19 outcomes compared to unvaccinated individuals. The vaccines have undergone rigorous testing and continue to be monitored for safety and efficacy." * 5, 0.85) ] results = [] for case_name, claim, text, reliability in benchmark_cases: times = [] for _ in range(5): # Run each test 5 times start_time = time.time() stance, confidence, layer = classifier.classify_detailed( title="Test Title", text=text, claim=claim, source_reliability=reliability ) end_time = time.time() times.append(end_time - start_time) avg_time = sum(times) / len(times) min_time = min(times) max_time = max(times) results.append({ "case": case_name, "claim_length": len(claim), "text_length": len(text), "avg_time": avg_time, "min_time": min_time, "max_time": max_time, "stance": stance, "confidence": confidence, "layer": layer }) print(f"{case_name}:") print(f" Claim length: {len(claim)} chars") print(f" Text length: {len(text)} chars") print(f" Avg time: {avg_time:.4f}s") print(f" Range: {min_time:.4f}s - {max_time:.4f}s") print(f" Result: {stance} ({confidence:.3f}, {layer})") print() return results except ImportError as e: print(f"Performance benchmark not available: {e}") return [] def main(): """Run all tests and generate a comprehensive report.""" print("ENHANCED STANCE CLASSIFICATION SYSTEM TEST SUITE") print("=" * 60) print(f"Test started at: {time.strftime('%Y-%m-%d %H:%M:%S')}") # Run all tests test_results = {} # Test 1: Enhanced Stance Classifier test_results['stance_classifier'] = test_enhanced_stance_classifier() # Test 2: Quality-Weighted Aggregator test_results['aggregator'] = test_quality_weighted_aggregator() # Test 3: Text Processing test_results['text_processing'] = test_text_processing() # Test 4: Data Collector Integration test_results['data_collector'] = test_data_collector_integration() # Test 5: Performance Benchmark test_results['performance'] = run_performance_benchmark() # Generate summary report print("\n" + "=" * 60) print("COMPREHENSIVE TEST SUMMARY") print("=" * 60) if test_results['stance_classifier']: stance_results = test_results['stance_classifier'] accuracy = sum(1 for r in stance_results if r['correct']) / len(stance_results) avg_confidence = sum(r['confidence'] for r in stance_results) / len(stance_results) print(f"✓ Enhanced Stance Classifier: {accuracy:.1%} accuracy, {avg_confidence:.3f} avg confidence") else: print("✗ Enhanced Stance Classifier: Not available") if test_results['aggregator']: print("✓ Quality-Weighted Aggregator: All strategies tested successfully") else: print("✗ Quality-Weighted Aggregator: Not available") if test_results['text_processing']: tp_results = test_results['text_processing'] print(f"✓ Text Processing: {tp_results.get('chunk_count', 0)} chunks, {tp_results.get('entities_found', 0)} entities extracted") else: print("✗ Text Processing: Not available") if test_results['data_collector']: dc_results = test_results['data_collector'] enhanced = "Enhanced" if dc_results.get('enhanced_classifier') else "Basic" print(f"✓ Data Collector Integration: {enhanced} classifier, {dc_results.get('backend_count', 0)} backends") else: print("✗ Data Collector Integration: Not available") if test_results['performance']: perf_results = test_results['performance'] avg_time = sum(r['avg_time'] for r in perf_results) / len(perf_results) print(f"✓ Performance Benchmark: {avg_time:.4f}s average processing time") else: print("✗ Performance Benchmark: Not available") print(f"\nTest completed at: {time.strftime('%Y-%m-%d %H:%M:%S')}") # Save detailed results to file try: with open('test_results.json', 'w') as f: json.dump(test_results, f, indent=2, default=str) print("Detailed results saved to test_results.json") except Exception as e: print(f"Could not save results: {e}") return test_results if __name__ == "__main__": main()