482 lines
16 KiB
Python
482 lines
16 KiB
Python
"""
|
|
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"
|