Spaces:
Sleeping
Sleeping
| """ | |
| Unit tests for ML feature extraction module | |
| """ | |
| import pytest | |
| import numpy as np | |
| from datetime import datetime, timedelta | |
| import sys | |
| import os | |
| sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) | |
| from features import FeatureExtractor, HistoricalFeatureAggregator, FeatureScaler | |
| class TestFeatureExtractor: | |
| """Tests for FeatureExtractor class""" | |
| def extractor(self): | |
| return FeatureExtractor() | |
| def test_extract_task_features_basic(self, extractor): | |
| """Test basic task feature extraction""" | |
| task_data = { | |
| "title": "Complete project report", | |
| "category": "WORK", | |
| "priority": "HIGH", | |
| "complexity": 4, | |
| "estimated_duration": 120, | |
| } | |
| features = extractor.extract_task_features(task_data) | |
| # Check for category-based features | |
| assert "cat_cognitive_load" in features | |
| assert "cat_time_sensitivity" in features | |
| # Check for priority-based features | |
| assert "pri_urgency_score" in features | |
| # Check for complexity features | |
| assert "complexity_normalized" in features | |
| def test_extract_task_features_with_due_date(self, extractor): | |
| """Test feature extraction with due date""" | |
| due_date = datetime.now() + timedelta(days=3) | |
| task_data = { | |
| "title": "Urgent task", | |
| "category": "WORK", | |
| "priority": "URGENT", | |
| "complexity": 5, | |
| "due_date": due_date.isoformat(), | |
| } | |
| features = extractor.extract_task_features(task_data) | |
| assert "days_until_due" in features | |
| assert features["days_until_due"] >= 2 # Approximately 3 days | |
| def test_extract_personality_features(self, extractor): | |
| """Test personality feature extraction""" | |
| personality = { | |
| "openness": 75, | |
| "conscientiousness": 85, | |
| "extraversion": 50, | |
| "agreeableness": 60, | |
| "neuroticism": 40, | |
| } | |
| features = extractor.extract_personality_features(personality) | |
| # Check for trait features | |
| assert "trait_openness" in features | |
| assert "trait_conscientiousness" in features | |
| # Check for derived features | |
| assert "emotional_stability" in features | |
| assert "productivity_potential" in features | |
| assert 0 <= features["trait_openness"] <= 1 | |
| def test_priority_score_mapping(self, extractor): | |
| """Test priority urgency score mapping""" | |
| tasks = [ | |
| {"title": "T1", "category": "WORK", "priority": "LOW", "complexity": 1}, | |
| {"title": "T2", "category": "WORK", "priority": "MEDIUM", "complexity": 1}, | |
| {"title": "T3", "category": "WORK", "priority": "HIGH", "complexity": 1}, | |
| {"title": "T4", "category": "WORK", "priority": "URGENT", "complexity": 1}, | |
| ] | |
| features = [extractor.extract_task_features(t) for t in tasks] | |
| scores = [f["pri_urgency_score"] for f in features] | |
| # Scores should increase with priority | |
| assert scores[0] < scores[1] < scores[2] < scores[3] | |
| def test_category_features_present(self, extractor): | |
| """Test category features are present""" | |
| categories = ["WORK", "ACADEMIC", "PERSONAL", "HEALTH"] | |
| for cat in categories: | |
| task = {"title": "Test", "category": cat, "priority": "MEDIUM", "complexity": 3} | |
| features = extractor.extract_task_features(task) | |
| assert "cat_cognitive_load" in features | |
| assert "cat_time_sensitivity" in features | |
| def test_complexity_normalization(self, extractor): | |
| """Test complexity is normalized correctly""" | |
| task = {"title": "Test", "category": "WORK", "priority": "MEDIUM", "complexity": 5} | |
| features = extractor.extract_task_features(task) | |
| assert features["complexity_normalized"] == 1.0 # 5/5 = 1.0 | |
| task["complexity"] = 1 | |
| features = extractor.extract_task_features(task) | |
| assert features["complexity_normalized"] == 0.2 # 1/5 = 0.2 | |
| class TestHistoricalFeatureAggregator: | |
| """Tests for HistoricalFeatureAggregator class""" | |
| def aggregator(self): | |
| return HistoricalFeatureAggregator() | |
| def test_aggregate_empty_history(self, aggregator): | |
| """Test aggregation with no historical data""" | |
| features = aggregator.aggregate_task_history([]) | |
| # Should return default features | |
| assert "completion_rate" in features | |
| assert isinstance(features["completion_rate"], (int, float)) | |
| def test_aggregate_with_history(self, aggregator): | |
| """Test aggregation with historical data""" | |
| history = [ | |
| {"status": "COMPLETED", "completed_at": datetime.now().isoformat()}, | |
| {"status": "COMPLETED", "completed_at": datetime.now().isoformat()}, | |
| {"status": "TODO"}, | |
| ] | |
| features = aggregator.aggregate_task_history(history) | |
| assert "completion_rate" in features | |
| assert features["completion_rate"] == pytest.approx(2/3, rel=0.01) | |
| class TestFeatureScaler: | |
| """Tests for FeatureScaler class""" | |
| def scaler(self): | |
| return FeatureScaler() | |
| def test_transform_features(self, scaler): | |
| """Test feature transformation""" | |
| features = { | |
| "days_until_due": 15, # Use known range feature | |
| "duration_minutes": 60, | |
| "complexity_raw": 3, | |
| } | |
| transformed = scaler.transform(features) | |
| # Transformed values should be normalized | |
| assert "days_until_due" in transformed | |
| assert 0 <= transformed["days_until_due"] <= 1 | |
| def test_transform_empty_features(self, scaler): | |
| """Test transforming with empty features""" | |
| transformed = scaler.transform({}) | |
| assert transformed == {} | |
| def test_fit_and_transform(self, scaler): | |
| """Test fitting and transforming feature sets""" | |
| feature_sets = [ | |
| {"complexity_normalized": 0.6, "score": 75}, | |
| {"complexity_normalized": 0.8, "score": 85}, | |
| ] | |
| transformed = scaler.fit_transform(feature_sets) | |
| assert len(transformed) == 2 | |
| for fs in transformed: | |
| assert "complexity_normalized" in fs | |
| if __name__ == "__main__": | |
| pytest.main([__file__, "-v"]) | |