""" Tests for Supplier Performance Predictor """ import pytest import pandas as pd import numpy as np from datetime import datetime, timedelta from app.ml.supplier_performance_predictor import SupplierPerformancePredictor @pytest.fixture def sample_order_history_good_supplier(): """Generate sample order history for a reliable supplier.""" dates = pd.date_range(start='2024-01-01', end='2024-12-31', freq='W') orders = [] for i, date in enumerate(dates): expected_delivery = date + timedelta(days=3) # Good supplier: 95% on-time, occasional 1-day delay if np.random.random() < 0.95: actual_delivery = expected_delivery else: actual_delivery = expected_delivery + timedelta(days=1) # Good quality: 98% no issues quality_issues = np.random.random() > 0.98 quality_score = np.random.uniform(90, 100) if not quality_issues else np.random.uniform(70, 85) # Good quantity accuracy: 99% accurate quantity_accuracy = np.random.uniform(0.98, 1.02) orders.append({ 'order_id': f'order-{i}', 'order_date': date, 'expected_delivery_date': expected_delivery, 'actual_delivery_date': actual_delivery, 'order_quantity': 100, 'received_quantity': int(100 * quantity_accuracy), 'quality_issues': quality_issues, 'quality_score': quality_score, 'order_value': 500.0 }) return pd.DataFrame(orders) @pytest.fixture def sample_order_history_poor_supplier(): """Generate sample order history for an unreliable supplier.""" dates = pd.date_range(start='2024-01-01', end='2024-12-31', freq='W') orders = [] for i, date in enumerate(dates): expected_delivery = date + timedelta(days=3) # Poor supplier: 60% on-time, frequent delays of 2-5 days if np.random.random() < 0.60: actual_delivery = expected_delivery else: actual_delivery = expected_delivery + timedelta(days=np.random.randint(2, 6)) # Poor quality: 20% issues quality_issues = np.random.random() > 0.80 quality_score = np.random.uniform(85, 100) if not quality_issues else np.random.uniform(50, 75) # Poor quantity accuracy: frequent short deliveries if np.random.random() < 0.25: quantity_accuracy = np.random.uniform(0.75, 0.95) # Short delivery else: quantity_accuracy = np.random.uniform(0.95, 1.05) orders.append({ 'order_id': f'order-{i}', 'order_date': date, 'expected_delivery_date': expected_delivery, 'actual_delivery_date': actual_delivery, 'order_quantity': 100, 'received_quantity': int(100 * quantity_accuracy), 'quality_issues': quality_issues, 'quality_score': quality_score, 'order_value': 500.0 }) return pd.DataFrame(orders) @pytest.mark.asyncio async def test_analyze_good_supplier(sample_order_history_good_supplier): """Test analysis of a reliable supplier.""" predictor = SupplierPerformancePredictor() results = await predictor.analyze_supplier_performance( tenant_id='test-tenant', supplier_id='good-supplier', order_history=sample_order_history_good_supplier, min_orders=10 ) # Check structure assert 'tenant_id' in results assert 'supplier_id' in results assert 'reliability_score' in results assert 'metrics' in results assert 'predictions' in results assert 'risk_assessment' in results assert 'insights' in results # Check metrics calculated metrics = results['metrics'] assert metrics['total_orders'] == len(sample_order_history_good_supplier) assert 'on_time_rate' in metrics assert 'quality_issue_rate' in metrics assert 'avg_quantity_accuracy' in metrics # Good supplier should have high reliability score reliability_score = results['reliability_score'] assert reliability_score >= 85, f"Expected high reliability, got {reliability_score}" # Risk should be low risk_assessment = results['risk_assessment'] assert risk_assessment['risk_level'] in ['low', 'medium'] @pytest.mark.asyncio async def test_analyze_poor_supplier(sample_order_history_poor_supplier): """Test analysis of an unreliable supplier.""" predictor = SupplierPerformancePredictor() results = await predictor.analyze_supplier_performance( tenant_id='test-tenant', supplier_id='poor-supplier', order_history=sample_order_history_poor_supplier, min_orders=10 ) # Poor supplier should have low reliability score reliability_score = results['reliability_score'] assert reliability_score < 75, f"Expected low reliability, got {reliability_score}" # Risk should be high or critical risk_assessment = results['risk_assessment'] assert risk_assessment['risk_level'] in ['medium', 'high', 'critical'] # Should have risk factors assert len(risk_assessment['risk_factors']) > 0 # Should generate insights insights = results['insights'] assert len(insights) > 0 # Should have at least one alert or prediction alert_insights = [i for i in insights if i['type'] in ['alert', 'prediction']] assert len(alert_insights) > 0 @pytest.mark.asyncio async def test_performance_metrics_calculation(sample_order_history_good_supplier): """Test detailed metrics calculation.""" predictor = SupplierPerformancePredictor() results = await predictor.analyze_supplier_performance( tenant_id='test-tenant', supplier_id='test-supplier', order_history=sample_order_history_good_supplier ) metrics = results['metrics'] # Check all key metrics present required_metrics = [ 'total_orders', 'on_time_orders', 'delayed_orders', 'on_time_rate', 'avg_delivery_delay_days', 'avg_quantity_accuracy', 'short_deliveries', 'short_delivery_rate', 'quality_issues', 'quality_issue_rate', 'avg_quality_score', 'delivery_consistency', 'quantity_consistency' ] for metric in required_metrics: assert metric in metrics, f"Missing metric: {metric}" # Check metrics are reasonable assert 0 <= metrics['on_time_rate'] <= 100 assert 0 <= metrics['avg_quantity_accuracy'] <= 200 # Allow up to 200% over-delivery assert 0 <= metrics['quality_issue_rate'] <= 100 assert 0 <= metrics['avg_quality_score'] <= 100 @pytest.mark.asyncio async def test_reliability_score_calculation(): """Test reliability score calculation with known inputs.""" predictor = SupplierPerformancePredictor() # Perfect metrics perfect_metrics = { 'on_time_rate': 100.0, 'avg_quantity_accuracy': 100.0, 'avg_quality_score': 100.0, 'delivery_consistency': 100.0, 'quantity_consistency': 100.0, 'quality_issue_rate': 0.0, 'short_delivery_rate': 0.0 } perfect_score = predictor._calculate_reliability_score(perfect_metrics) assert perfect_score >= 95, f"Expected perfect score ~100, got {perfect_score}" # Poor metrics poor_metrics = { 'on_time_rate': 50.0, 'avg_quantity_accuracy': 85.0, 'avg_quality_score': 70.0, 'delivery_consistency': 50.0, 'quantity_consistency': 60.0, 'quality_issue_rate': 20.0, # Should apply penalty 'short_delivery_rate': 25.0 # Should apply penalty } poor_score = predictor._calculate_reliability_score(poor_metrics) assert poor_score < 70, f"Expected poor score <70, got {poor_score}" @pytest.mark.asyncio async def test_delay_probability_prediction(sample_order_history_poor_supplier): """Test delay probability prediction.""" predictor = SupplierPerformancePredictor() results = await predictor.analyze_supplier_performance( tenant_id='test-tenant', supplier_id='test-supplier', order_history=sample_order_history_poor_supplier ) predictions = results['predictions'] # Should have delay probability assert 'next_order_delay_probability' in predictions assert 0 <= predictions['next_order_delay_probability'] <= 1.0 # Poor supplier should have higher delay probability assert predictions['next_order_delay_probability'] > 0.3 # Should have confidence score assert 'confidence' in predictions assert 0 <= predictions['confidence'] <= 100 @pytest.mark.asyncio async def test_risk_assessment(sample_order_history_poor_supplier): """Test procurement risk assessment.""" predictor = SupplierPerformancePredictor() results = await predictor.analyze_supplier_performance( tenant_id='test-tenant', supplier_id='test-supplier', order_history=sample_order_history_poor_supplier ) risk_assessment = results['risk_assessment'] # Check structure assert 'risk_level' in risk_assessment assert 'risk_score' in risk_assessment assert 'risk_factors' in risk_assessment assert 'recommendation' in risk_assessment # Risk level should be valid assert risk_assessment['risk_level'] in ['low', 'medium', 'high', 'critical'] # Risk score should be 0-100 assert 0 <= risk_assessment['risk_score'] <= 100 # Should have risk factors for poor supplier assert len(risk_assessment['risk_factors']) > 0 # Recommendation should be string assert isinstance(risk_assessment['recommendation'], str) assert len(risk_assessment['recommendation']) > 0 @pytest.mark.asyncio async def test_insight_generation_low_reliability(sample_order_history_poor_supplier): """Test insight generation for low reliability supplier.""" predictor = SupplierPerformancePredictor() results = await predictor.analyze_supplier_performance( tenant_id='test-tenant', supplier_id='poor-supplier', order_history=sample_order_history_poor_supplier ) insights = results['insights'] # Should generate insights assert len(insights) > 0 # Check for low reliability alert reliability_insights = [i for i in insights if 'reliability' in i.get('title', '').lower()] if reliability_insights: insight = reliability_insights[0] assert insight['type'] in ['alert', 'recommendation'] assert insight['priority'] in ['high', 'critical'] assert 'actionable' in insight assert insight['actionable'] is True assert 'recommendation_actions' in insight assert len(insight['recommendation_actions']) > 0 @pytest.mark.asyncio async def test_insight_generation_high_delay_risk(sample_order_history_poor_supplier): """Test insight generation for high delay probability.""" predictor = SupplierPerformancePredictor() results = await predictor.analyze_supplier_performance( tenant_id='test-tenant', supplier_id='poor-supplier', order_history=sample_order_history_poor_supplier ) insights = results['insights'] # Check for delay risk prediction delay_insights = [i for i in insights if 'delay' in i.get('title', '').lower()] if delay_insights: insight = delay_insights[0] assert 'confidence' in insight assert 'metrics_json' in insight assert 'recommendation_actions' in insight @pytest.mark.asyncio async def test_insight_generation_excellent_supplier(sample_order_history_good_supplier): """Test that excellent suppliers get positive insights.""" predictor = SupplierPerformancePredictor() results = await predictor.analyze_supplier_performance( tenant_id='test-tenant', supplier_id='excellent-supplier', order_history=sample_order_history_good_supplier ) insights = results['insights'] # Should have positive insight for excellent performance positive_insights = [i for i in insights if 'excellent' in i.get('title', '').lower()] if positive_insights: insight = positive_insights[0] assert insight['type'] == 'insight' assert insight['impact_type'] == 'positive_performance' def test_compare_suppliers(): """Test supplier comparison functionality.""" predictor = SupplierPerformancePredictor() # Mock analysis results suppliers_analysis = [ { 'supplier_id': 'supplier-1', 'reliability_score': 95, 'risk_assessment': {'risk_level': 'low', 'risk_score': 10} }, { 'supplier_id': 'supplier-2', 'reliability_score': 60, 'risk_assessment': {'risk_level': 'high', 'risk_score': 75} }, { 'supplier_id': 'supplier-3', 'reliability_score': 80, 'risk_assessment': {'risk_level': 'medium', 'risk_score': 40} } ] comparison = predictor.compare_suppliers(suppliers_analysis) # Check structure assert 'suppliers_compared' in comparison assert 'top_supplier' in comparison assert 'top_supplier_score' in comparison assert 'bottom_supplier' in comparison assert 'bottom_supplier_score' in comparison assert 'ranked_suppliers' in comparison assert 'recommendations' in comparison # Check ranking assert comparison['suppliers_compared'] == 3 assert comparison['top_supplier'] == 'supplier-1' assert comparison['top_supplier_score'] == 95 assert comparison['bottom_supplier'] == 'supplier-2' assert comparison['bottom_supplier_score'] == 60 # Ranked suppliers should be in order ranked = comparison['ranked_suppliers'] assert ranked[0]['supplier_id'] == 'supplier-1' assert ranked[-1]['supplier_id'] == 'supplier-2' # Should have recommendations assert len(comparison['recommendations']) > 0 @pytest.mark.asyncio async def test_insufficient_data_handling(): """Test handling of insufficient order history.""" predictor = SupplierPerformancePredictor() # Only 5 orders (less than min_orders=10) small_history = pd.DataFrame([ { 'order_date': datetime(2024, 1, i), 'expected_delivery_date': datetime(2024, 1, i+3), 'actual_delivery_date': datetime(2024, 1, i+3), 'order_quantity': 100, 'received_quantity': 100, 'quality_issues': False, 'quality_score': 95.0, 'order_value': 500.0 } for i in range(1, 6) ]) results = await predictor.analyze_supplier_performance( tenant_id='test-tenant', supplier_id='new-supplier', order_history=small_history, min_orders=10 ) # Should return insufficient data response assert results['orders_analyzed'] == 0 assert results['reliability_score'] is None assert results['risk_assessment']['risk_level'] == 'unknown' assert 'Insufficient' in results['risk_assessment']['risk_factors'][0] def test_get_supplier_reliability_score(): """Test getting cached reliability scores.""" predictor = SupplierPerformancePredictor() # Initially no score assert predictor.get_supplier_reliability_score('supplier-1') is None # Set a score predictor.reliability_scores['supplier-1'] = 85 # Should retrieve it assert predictor.get_supplier_reliability_score('supplier-1') == 85 @pytest.mark.asyncio async def test_metrics_no_nan_values(sample_order_history_good_supplier): """Test that metrics never contain NaN values.""" predictor = SupplierPerformancePredictor() results = await predictor.analyze_supplier_performance( tenant_id='test-tenant', supplier_id='test-supplier', order_history=sample_order_history_good_supplier ) metrics = results['metrics'] # Check no NaN values for key, value in metrics.items(): if isinstance(value, float): assert not np.isnan(value), f"Metric {key} is NaN"