Create conftest.py

Adding Tests for MLOps Infrastructure Enhancement

tests/conftest.py

@@ -0,0 +1,493 @@
+# tests/conftest.py
+# Shared test configuration and fixtures
+
+import pytest
+import numpy as np
+import pandas as pd
+import tempfile
+import sys
+import os
+from pathlib import Path
+from unittest.mock import patch
+
+# Add project root to Python path
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
+
+@pytest.fixture(scope="session")
+def test_data_dir():
+    """Create temporary directory for test data"""
+    with tempfile.TemporaryDirectory() as temp_dir:
+        yield Path(temp_dir)
+
+@pytest.fixture(scope="session") 
+def sample_fake_news_data():
+    """Generate realistic fake news dataset for testing"""
+    np.random.seed(42)
+    
+    # Realistic fake news patterns
+    fake_texts = [
+        "BREAKING: Scientists discover shocking truth about vaccines that doctors don't want you to know!",
+        "EXCLUSIVE: Celebrity caught in major scandal - you won't believe what happened next!",
+        "ALERT: Government secretly planning massive operation - leaked documents reveal everything!",
+        "AMAZING: Local mom discovers one weird trick that makes millions - experts hate her!",
+        "URGENT: New study proves everything you know about nutrition is completely wrong!",
+    ] * 20
+    
+    # Realistic real news patterns
+    real_texts = [
+        "Local city council approves new infrastructure budget for road maintenance and repairs.",
+        "University researchers publish peer-reviewed study on climate change impacts in regional ecosystems.",
+        "Stock market shows mixed results following quarterly earnings reports from major corporations.",
+        "Public health officials recommend updated vaccination schedules based on recent clinical trials.",
+        "Municipal government announces new public transportation routes to improve city connectivity.",
+    ] * 20
+    
+    # Combine and create DataFrame
+    all_texts = fake_texts + real_texts
+    all_labels = [1] * len(fake_texts) + [0] * len(real_texts)
+    
+    df = pd.DataFrame({
+        'text': all_texts,
+        'label': all_labels
+    })
+    
+    return df.sample(frac=1, random_state=42).reset_index(drop=True)
+
+@pytest.fixture
+def mock_enhanced_features():
+    """Mock enhanced feature engineering when not available"""
+    with patch('model.retrain.ENHANCED_FEATURES_AVAILABLE', True):
+        with patch('model.retrain.AdvancedFeatureEngineer') as mock_fe:
+            # Configure mock to behave like real feature engineer
+            mock_instance = mock_fe.return_value
+            mock_instance.get_feature_metadata.return_value = {
+                'total_features': 5000,
+                'feature_types': {
+                    'tfidf_features': 3000,
+                    'sentiment_features': 10,
+                    'readability_features': 15,
+                    'entity_features': 25,
+                    'linguistic_features': 50
+                },
+                'configuration': {'test': True}
+            }
+            mock_instance.get_feature_importance.return_value = {
+                'feature_1': 0.15,
+                'feature_2': 0.12,
+                'feature_3': 0.10
+            }
+            mock_instance.get_feature_names.return_value = [f'feature_{i}' for i in range(5000)]
+            
+            yield mock_fe
+
+# tests/test_data_processing.py
+# Test data processing and validation components
+
+import pytest
+import pandas as pd
+import numpy as np
+from pathlib import Path
+import tempfile
+
+from data.data_validator import DataValidator
+from data.prepare_datasets import DatasetPreparer
+
+class TestDataValidation:
+    """Test data validation functionality"""
+    
+    def test_validate_text_column(self, sample_fake_news_data):
+        """Test text column validation"""
+        validator = DataValidator()
+        
+        # Valid data should pass
+        is_valid, issues = validator.validate_dataframe(sample_fake_news_data)
+        assert is_valid == True
+        assert len(issues) == 0
+        
+        # Test with invalid data
+        invalid_data = pd.DataFrame({
+            'text': ['', 'x', None, 'Valid text here'],
+            'label': [0, 1, 0, 2]  # Invalid label
+        })
+        
+        is_valid, issues = validator.validate_dataframe(invalid_data)
+        assert is_valid == False
+        assert len(issues) > 0
+    
+    def test_text_quality_validation(self):
+        """Test text quality validation rules"""
+        validator = DataValidator()
+        
+        # Test minimum length requirement
+        short_texts = pd.DataFrame({
+            'text': ['hi', 'ok', 'This is a proper length text for validation'],
+            'label': [0, 1, 0]
+        })
+        
+        is_valid, issues = validator.validate_dataframe(short_texts)
+        assert is_valid == False
+        assert any('length' in str(issue).lower() for issue in issues)
+
+
+# tests/test_train_integration.py
+# Test integration with train.py to ensure compatibility
+
+import pytest
+import tempfile
+from pathlib import Path
+from unittest.mock import patch
+
+class TestTrainRetrainCompatibility:
+    """Test compatibility between train.py and retrain.py"""
+    
+    def test_metadata_compatibility(self):
+        """Test metadata format compatibility between train and retrain"""
+        from model.train import EnhancedModelTrainer
+        from model.retrain import EnhancedModelRetrainer
+        
+        with tempfile.TemporaryDirectory() as temp_dir:
+            temp_path = Path(temp_dir)
+            
+            # Mock trainer to avoid full training
+            trainer = EnhancedModelTrainer(use_enhanced_features=False)
+            trainer.base_dir = temp_path
+            trainer.setup_paths()
+            
+            # Create sample metadata as train.py would
+            sample_metadata = {
+                'model_version': 'v1.0',
+                'model_type': 'enhanced_pipeline_cv',
+                'feature_engineering': {'type': 'standard'},
+                'test_f1': 0.85,
+                'cross_validation': {
+                    'test_scores': {'f1': {'mean': 0.82, 'std': 0.03}}
+                }
+            }
+            
+            # Save metadata
+            import json
+            with open(trainer.metadata_path, 'w') as f:
+                json.dump(sample_metadata, f)
+            
+            # Test retrainer can read it
+            retrainer = EnhancedModelRetrainer()
+            retrainer.base_dir = temp_path
+            retrainer.setup_paths()
+            
+            metadata = retrainer.load_existing_metadata()
+            assert metadata is not None
+            assert metadata['model_version'] == 'v1.0'
+            assert metadata['feature_engineering']['type'] == 'standard'
+    
+    def test_model_file_compatibility(self):
+        """Test model file format compatibility"""
+        # Both train.py and retrain.py should save/load models consistently
+        from model.retrain import EnhancedModelRetrainer
+        
+        with tempfile.TemporaryDirectory() as temp_dir:
+            temp_path = Path(temp_dir)
+            
+            retrainer = EnhancedModelRetrainer()
+            retrainer.base_dir = temp_path
+            retrainer.setup_paths()
+            
+            # Create mock pipeline as train.py would save
+            from sklearn.pipeline import Pipeline
+            from sklearn.linear_model import LogisticRegression
+            from sklearn.feature_extraction.text import TfidfVectorizer
+            
+            mock_pipeline = Pipeline([
+                ('vectorize', TfidfVectorizer(max_features=1000)),
+                ('model', LogisticRegression())
+            ])
+            
+            import joblib
+            joblib.dump(mock_pipeline, retrainer.prod_pipeline_path)
+            
+            # Test retrainer can load it
+            success, model, message = retrainer.load_production_model()
+            assert success == True
+            assert model is not None
+
+
+# tests/pytest.ini
+# Pytest configuration file
+[tool:pytest]
+testpaths = tests
+python_files = test_*.py
+python_classes = Test*
+python_functions = test_*
+addopts = 
+    -v
+    --tb=short
+    --strict-markers
+    --disable-warnings
+    --color=yes
+markers =
+    slow: marks tests as slow (deselect with '-m "not slow"')
+    integration: marks tests as integration tests
+    unit: marks tests as unit tests
+    cpu_constraint: marks tests that verify CPU constraint compliance
+filterwarnings =
+    ignore::UserWarning
+    ignore::FutureWarning
+    ignore::DeprecationWarning
+
+
+# tests/test_lightgbm_integration.py
+# Specific tests for LightGBM integration
+
+import pytest
+import numpy as np
+from unittest.mock import patch
+import lightgbm as lgb
+
+class TestLightGBMIntegration:
+    """Test LightGBM-specific functionality"""
+    
+    def test_lightgbm_model_configuration(self):
+        """Test LightGBM model is properly configured for CPU constraints"""
+        from model.retrain import EnhancedModelRetrainer
+        
+        retrainer = EnhancedModelRetrainer()
+        lgb_config = retrainer.models['lightgbm']
+        lgb_model = lgb_config['model']
+        
+        # Verify CPU-friendly configuration
+        assert isinstance(lgb_model, lgb.LGBMClassifier)
+        assert lgb_model.n_jobs == 1
+        assert lgb_model.verbose == -1
+        assert lgb_model.n_estimators <= 100
+        assert lgb_model.num_leaves <= 31
+        
+        # Verify parameter grid is reasonable for CPU
+        param_grid = lgb_config['param_grid']
+        assert all(est <= 100 for est in param_grid['model__n_estimators'])
+        assert all(leaves <= 31 for leaves in param_grid['model__num_leaves'])
+    
+    def test_lightgbm_training_integration(self):
+        """Test LightGBM integrates properly in training pipeline"""
+        from model.retrain import EnhancedModelRetrainer
+        
+        # Create minimal dataset
+        X = np.random.randn(50, 10)
+        y = np.random.randint(0, 2, 50)
+        
+        retrainer = EnhancedModelRetrainer()
+        retrainer.use_enhanced_features = False
+        
+        # Test hyperparameter tuning works with LightGBM
+        pipeline = retrainer.create_preprocessing_pipeline()
+        
+        try:
+            best_model, results = retrainer.hyperparameter_tuning_with_cv(
+                pipeline, X, y, 'lightgbm'
+            )
+            
+            # Should complete without errors
+            assert best_model is not None
+            assert 'cross_validation' in results or 'error' in results
+            
+        except Exception as e:
+            # If tuning fails, should fall back gracefully
+            assert 'fallback' in str(e).lower() or 'error' in str(e).lower()
+    
+    def test_lightgbm_cpu_performance(self):
+        """Test LightGBM performance is acceptable under CPU constraints"""
+        import time
+        from model.retrain import EnhancedModelRetrainer
+        
+        # Create reasonably sized dataset
+        X = np.random.randn(200, 20)
+        y = np.random.randint(0, 2, 200)
+        
+        retrainer = EnhancedModelRetrainer()
+        pipeline = retrainer.create_preprocessing_pipeline()
+        lgb_model = retrainer.models['lightgbm']['model']
+        pipeline.set_params(model=lgb_model)
+        
+        # Time the training
+        start_time = time.time()
+        pipeline.fit(X, y)
+        training_time = time.time() - start_time
+        
+        # Should complete reasonably quickly on CPU
+        assert training_time < 30  # Should take less than 30 seconds
+        
+        # Should produce valid predictions
+        predictions = pipeline.predict(X[:10])
+        assert len(predictions) == 10
+        assert all(pred in [0, 1] for pred in predictions)
+
+
+# tests/test_ensemble_statistical_validation.py  
+# Test ensemble statistical validation logic
+
+import pytest
+import numpy as np
+from scipy import stats
+from unittest.mock import Mock, patch
+
+class TestEnsembleStatisticalValidation:
+    """Test statistical validation for ensemble selection"""
+    
+    def test_paired_ttest_ensemble_selection(self):
+        """Test paired t-test logic for ensemble vs individual models"""
+        from model.retrain import CVModelComparator
+        
+        comparator = CVModelComparator(cv_folds=5, random_state=42)
+        
+        # Create mock CV scores where ensemble is significantly better
+        individual_scores = [0.75, 0.74, 0.76, 0.73, 0.75]
+        ensemble_scores = [0.80, 0.81, 0.79, 0.78, 0.82]
+        
+        # Test metric comparison
+        comparison = comparator._compare_metric_scores(
+            individual_scores, ensemble_scores, 'f1', 'individual', 'ensemble'
+        )
+        
+        assert 'tests' in comparison
+        assert 'paired_ttest' in comparison['tests']
+        
+        # Should detect significant improvement
+        t_test_result = comparison['tests']['paired_ttest']
+        assert 'p_value' in t_test_result
+        assert 'significant' in t_test_result
+        
+        # With this clear difference, should be significant
+        if t_test_result['p_value'] is not None:
+            assert t_test_result['significant'] == True
+    
+    def test_ensemble_not_selected_when_not_significant(self):
+        """Test ensemble is not selected when improvement is not significant"""
+        from model.retrain import CVModelComparator
+        
+        comparator = CVModelComparator(cv_folds=5, random_state=42)
+        
+        # Create mock CV scores where ensemble is only marginally better
+        individual_scores = [0.75, 0.74, 0.76, 0.73, 0.75]
+        ensemble_scores = [0.751, 0.741, 0.761, 0.731, 0.751]  # Tiny improvement
+        
+        comparison = comparator._compare_metric_scores(
+            individual_scores, ensemble_scores, 'f1', 'individual', 'ensemble'
+        )
+        
+        # Should not show significant improvement
+        assert comparison['significant_improvement'] == False
+    
+    def test_effect_size_calculation(self):
+        """Test Cohen's d effect size calculation"""
+        from model.retrain import CVModelComparator
+        
+        comparator = CVModelComparator(cv_folds=5, random_state=42)
+        
+        # Create scores with known effect size
+        individual_scores = [0.70, 0.71, 0.69, 0.72, 0.70]
+        ensemble_scores = [0.80, 0.81, 0.79, 0.82, 0.80]  # Large effect
+        
+        comparison = comparator._compare_metric_scores(
+            individual_scores, ensemble_scores, 'f1', 'individual', 'ensemble'
+        )
+        
+        assert 'effect_size' in comparison
+        effect_size = comparison['effect_size']
+        
+        # Should detect large effect size
+        assert abs(effect_size) > 0.5  # Large effect by Cohen's standards
+    
+    def test_promotion_decision_with_feature_upgrade(self):
+        """Test promotion decision considers feature engineering upgrades"""
+        from model.retrain import CVModelComparator
+        
+        comparator = CVModelComparator()
+        
+        # Mock comparison results with feature upgrade
+        mock_results = {
+            'metric_comparisons': {
+                'f1': {
+                    'improvement': 0.008,  # Small improvement
+                    'significant_improvement': False
+                },
+                'accuracy': {
+                    'improvement': 0.005,
+                    'significant_improvement': False
+                }
+            },
+            'feature_engineering_comparison': {
+                'feature_upgrade': {
+                    'is_upgrade': True,
+                    'upgrade_type': 'standard_to_enhanced'
+                }
+            }
+        }
+        
+        decision = comparator._make_enhanced_promotion_decision(mock_results)
+        
+        # Should promote despite small improvement due to feature upgrade
+        assert decision['promote_candidate'] == True
+        assert decision['feature_engineering_factor'] == True
+        assert 'feature' in decision['reason'].lower()
+
+
+# tests/run_tests.py
+# Test runner script with different test categories
+
+import pytest
+import sys
+from pathlib import Path
+
+def run_unit_tests():
+    """Run fast unit tests"""
+    return pytest.main([
+        "tests/", 
+        "-m", "not slow and not integration",
+        "-v", 
+        "--tb=short"
+    ])
+
+def run_integration_tests():
+    """Run slower integration tests"""
+    return pytest.main([
+        "tests/",
+        "-m", "integration", 
+        "-v",
+        "--tb=short"
+    ])
+
+def run_cpu_constraint_tests():
+    """Run tests that verify CPU constraint compliance"""
+    return pytest.main([
+        "tests/",
+        "-m", "cpu_constraint",
+        "-v",
+        "--tb=short"
+    ])
+
+def run_all_tests():
+    """Run complete test suite"""
+    return pytest.main([
+        "tests/",
+        "-v",
+        "--tb=short",
+        "--cov=model",
+        "--cov-report=html"
+    ])
+
+if __name__ == "__main__":
+    if len(sys.argv) > 1:
+        test_type = sys.argv[1]
+        if test_type == "unit":
+            exit_code = run_unit_tests()
+        elif test_type == "integration":  
+            exit_code = run_integration_tests()
+        elif test_type == "cpu":
+            exit_code = run_cpu_constraint_tests()
+        elif test_type == "all":
+            exit_code = run_all_tests()
+        else:
+            print("Usage: python run_tests.py [unit|integration|cpu|all]")
+            exit_code = 1
+    else:
+        exit_code = run_unit_tests()  # Default to unit tests
+    
+    sys.exit(exit_code)