379 lines
11 KiB
Python
379 lines
11 KiB
Python
"""
|
||
Tests for Energy Calculator Module.
|
||
|
||
This module tests the energy score calculation functionality including:
|
||
- Multi-source weighted calculation
|
||
- Temporal weighting
|
||
- Degraded mode (when sources are unavailable)
|
||
- Score normalization (0-100)
|
||
- Confidence calculation
|
||
"""
|
||
|
||
import pytest
|
||
from datetime import datetime, timedelta
|
||
from app.ml.energy_calculator import (
|
||
calculate_energy_score,
|
||
apply_temporal_weighting,
|
||
calculate_confidence,
|
||
normalize_score,
|
||
apply_source_weights,
|
||
adjust_weights_for_degraded_mode
|
||
)
|
||
|
||
|
||
class TestCalculateEnergyScore:
|
||
"""Test the main energy score calculation function."""
|
||
|
||
def test_calculate_energy_score_with_all_sources(self):
|
||
"""Test energy calculation with all three sources available."""
|
||
# Arrange
|
||
match_id = 1
|
||
team_id = 1
|
||
twitter_sentiments = [
|
||
{'compound': 0.5, 'positive': 0.6, 'negative': 0.2, 'neutral': 0.2, 'sentiment': 'positive'}
|
||
]
|
||
reddit_sentiments = [
|
||
{'compound': 0.3, 'positive': 0.4, 'negative': 0.3, 'neutral': 0.3, 'sentiment': 'positive'}
|
||
]
|
||
rss_sentiments = [
|
||
{'compound': 0.4, 'positive': 0.5, 'negative': 0.2, 'neutral': 0.3, 'sentiment': 'positive'}
|
||
]
|
||
|
||
# Act
|
||
result = calculate_energy_score(
|
||
match_id=match_id,
|
||
team_id=team_id,
|
||
twitter_sentiments=twitter_sentiments,
|
||
reddit_sentiments=reddit_sentiments,
|
||
rss_sentiments=rss_sentiments
|
||
)
|
||
|
||
# Assert
|
||
assert 'score' in result
|
||
assert 'confidence' in result
|
||
assert 'sources_used' in result
|
||
assert 0 <= result['score'] <= 100
|
||
assert len(result['sources_used']) == 3
|
||
assert result['confidence'] > 0.6 # High confidence with all sources
|
||
|
||
def test_calculate_energy_score_with_twitter_only(self):
|
||
"""Test energy calculation with only Twitter source available (degraded mode)."""
|
||
# Arrange
|
||
match_id = 1
|
||
team_id = 1
|
||
twitter_sentiments = [
|
||
{'compound': 0.5, 'positive': 0.6, 'negative': 0.2, 'neutral': 0.2, 'sentiment': 'positive'}
|
||
]
|
||
reddit_sentiments = []
|
||
rss_sentiments = []
|
||
|
||
# Act
|
||
result = calculate_energy_score(
|
||
match_id=match_id,
|
||
team_id=team_id,
|
||
twitter_sentiments=twitter_sentiments,
|
||
reddit_sentiments=reddit_sentiments,
|
||
rss_sentiments=rss_sentiments
|
||
)
|
||
|
||
# Assert
|
||
assert 'score' in result
|
||
assert 'confidence' in result
|
||
assert 'sources_used' in result
|
||
assert 0 <= result['score'] <= 100
|
||
assert len(result['sources_used']) == 1
|
||
assert 'twitter' in result['sources_used']
|
||
assert result['confidence'] < 0.6 # Lower confidence in degraded mode
|
||
|
||
def test_calculate_energy_score_no_sentiment_data(self):
|
||
"""Test energy calculation with no sentiment data."""
|
||
# Arrange
|
||
match_id = 1
|
||
team_id = 1
|
||
|
||
# Act
|
||
result = calculate_energy_score(
|
||
match_id=match_id,
|
||
team_id=team_id,
|
||
twitter_sentiments=[],
|
||
reddit_sentiments=[],
|
||
rss_sentiments=[]
|
||
)
|
||
|
||
# Assert
|
||
assert result['score'] == 0.0
|
||
assert result['confidence'] == 0.0
|
||
assert len(result['sources_used']) == 0
|
||
|
||
|
||
class TestApplySourceWeights:
|
||
"""Test source weight application and adjustment."""
|
||
|
||
def test_apply_source_weights_all_sources(self):
|
||
"""Test applying weights with all sources available."""
|
||
# Arrange
|
||
twitter_score = 50.0
|
||
reddit_score = 40.0
|
||
rss_score = 30.0
|
||
available_sources = ['twitter', 'reddit', 'rss']
|
||
|
||
# Act
|
||
weighted_score = apply_source_weights(
|
||
twitter_score=twitter_score,
|
||
reddit_score=reddit_score,
|
||
rss_score=rss_score,
|
||
available_sources=available_sources
|
||
)
|
||
|
||
# Assert
|
||
expected = (50.0 * 0.60) + (40.0 * 0.25) + (30.0 * 0.15)
|
||
assert weighted_score == expected
|
||
|
||
def test_apply_source_weights_twitter_only(self):
|
||
"""Test applying weights with only Twitter available."""
|
||
# Arrange
|
||
twitter_score = 50.0
|
||
reddit_score = 0.0
|
||
rss_score = 0.0
|
||
available_sources = ['twitter']
|
||
|
||
# Act
|
||
weighted_score = apply_source_weights(
|
||
twitter_score=twitter_score,
|
||
reddit_score=reddit_score,
|
||
rss_score=rss_score,
|
||
available_sources=available_sources
|
||
)
|
||
|
||
# Assert
|
||
assert weighted_score == 50.0 # 100% of weight goes to Twitter
|
||
|
||
def test_adjust_weights_for_degraded_mode(self):
|
||
"""Test weight adjustment in degraded mode."""
|
||
# Arrange
|
||
original_weights = {
|
||
'twitter': 0.60,
|
||
'reddit': 0.25,
|
||
'rss': 0.15
|
||
}
|
||
available_sources = ['twitter', 'reddit']
|
||
|
||
# Act
|
||
adjusted_weights = adjust_weights_for_degraded_mode(
|
||
original_weights=original_weights,
|
||
available_sources=available_sources
|
||
)
|
||
|
||
# Assert
|
||
assert len(adjusted_weights) == 2
|
||
assert 'twitter' in adjusted_weights
|
||
assert 'reddit' in adjusted_weights
|
||
assert 'rss' not in adjusted_weights
|
||
# Total should be 1.0
|
||
total_weight = sum(adjusted_weights.values())
|
||
assert abs(total_weight - 1.0) < 0.001
|
||
|
||
|
||
class TestApplyTemporalWeighting:
|
||
"""Test temporal weighting of sentiment scores."""
|
||
|
||
def test_temporal_weighting_recent_tweets(self):
|
||
"""Test temporal weighting with recent tweets (within 1 hour)."""
|
||
# Arrange
|
||
base_score = 50.0
|
||
now = datetime.utcnow()
|
||
tweets_with_timestamps = [
|
||
{
|
||
'compound': 0.5,
|
||
'created_at': now - timedelta(minutes=30) # 30 minutes ago
|
||
},
|
||
{
|
||
'compound': 0.6,
|
||
'created_at': now - timedelta(minutes=15) # 15 minutes ago
|
||
}
|
||
]
|
||
|
||
# Act
|
||
weighted_score = apply_temporal_weighting(
|
||
base_score=base_score,
|
||
tweets_with_timestamps=tweets_with_timestamps
|
||
)
|
||
|
||
# Assert
|
||
# Recent tweets should have higher weight (close to 1.0)
|
||
# Score should be close to original but weighted by recency
|
||
assert 0 <= weighted_score <= 100
|
||
|
||
def test_temporal_weighting_old_tweets(self):
|
||
"""Test temporal weighting with old tweets (24+ hours)."""
|
||
# Arrange
|
||
base_score = 50.0
|
||
now = datetime.utcnow()
|
||
tweets_with_timestamps = [
|
||
{
|
||
'compound': 0.5,
|
||
'created_at': now - timedelta(hours=30) # 30 hours ago
|
||
},
|
||
{
|
||
'compound': 0.6,
|
||
'created_at': now - timedelta(hours=25) # 25 hours ago
|
||
}
|
||
]
|
||
|
||
# Act
|
||
weighted_score = apply_temporal_weighting(
|
||
base_score=base_score,
|
||
tweets_with_timestamps=tweets_with_timestamps
|
||
)
|
||
|
||
# Assert
|
||
# Old tweets should have lower weight (around 0.5)
|
||
assert 0 <= weighted_score <= 100
|
||
|
||
|
||
class TestNormalizeScore:
|
||
"""Test score normalization to 0-100 range."""
|
||
|
||
def test_normalize_score_negative(self):
|
||
"""Test normalization of negative scores."""
|
||
# Act
|
||
normalized = normalize_score(-1.5)
|
||
|
||
# Assert
|
||
assert normalized == 0.0
|
||
|
||
def test_normalize_score_positive(self):
|
||
"""Test normalization of positive scores."""
|
||
# Act
|
||
normalized = normalize_score(150.0)
|
||
|
||
# Assert
|
||
assert normalized == 100.0
|
||
|
||
def test_normalize_score_in_range(self):
|
||
"""Test normalization of scores within range."""
|
||
# Arrange & Act
|
||
normalized_0 = normalize_score(0.0)
|
||
normalized_50 = normalize_score(50.0)
|
||
normalized_100 = normalize_score(100.0)
|
||
|
||
# Assert
|
||
assert normalized_0 == 0.0
|
||
assert normalized_50 == 50.0
|
||
assert normalized_100 == 100.0
|
||
|
||
|
||
class TestCalculateConfidence:
|
||
"""Test confidence level calculation."""
|
||
|
||
def test_confidence_all_sources(self):
|
||
"""Test confidence with all sources available."""
|
||
# Arrange
|
||
available_sources = ['twitter', 'reddit', 'rss']
|
||
total_weight = 1.0
|
||
|
||
# Act
|
||
confidence = calculate_confidence(
|
||
available_sources=available_sources,
|
||
total_weight=total_weight
|
||
)
|
||
|
||
# Assert
|
||
assert confidence > 0.6 # High confidence
|
||
assert confidence <= 1.0
|
||
|
||
def test_confidence_single_source(self):
|
||
"""Test confidence with only one source available."""
|
||
# Arrange
|
||
available_sources = ['twitter']
|
||
total_weight = 0.6
|
||
|
||
# Act
|
||
confidence = calculate_confidence(
|
||
available_sources=available_sources,
|
||
total_weight=total_weight
|
||
)
|
||
|
||
# Assert
|
||
assert confidence < 0.6 # Lower confidence
|
||
assert confidence >= 0.3
|
||
|
||
def test_confidence_no_sources(self):
|
||
"""Test confidence with no sources available."""
|
||
# Arrange
|
||
available_sources = []
|
||
total_weight = 0.0
|
||
|
||
# Act
|
||
confidence = calculate_confidence(
|
||
available_sources=available_sources,
|
||
total_weight=total_weight
|
||
)
|
||
|
||
# Assert
|
||
assert confidence == 0.0
|
||
|
||
|
||
class TestEnergyFormula:
|
||
"""Test the core energy formula: Score = (Positive - Negative) × Volume × Virality."""
|
||
|
||
def test_formula_positive_sentiment(self):
|
||
"""Test energy formula with positive sentiment."""
|
||
# Arrange
|
||
positive_ratio = 0.7
|
||
negative_ratio = 0.2
|
||
volume = 100
|
||
virality = 1.5
|
||
|
||
# Act
|
||
score = (positive_ratio - negative_ratio) * volume * virality
|
||
|
||
# Assert
|
||
assert score > 0 # Positive energy
|
||
|
||
def test_formula_negative_sentiment(self):
|
||
"""Test energy formula with negative sentiment."""
|
||
# Arrange
|
||
positive_ratio = 0.2
|
||
negative_ratio = 0.7
|
||
volume = 100
|
||
virality = 1.5
|
||
|
||
# Act
|
||
score = (positive_ratio - negative_ratio) * volume * virality
|
||
|
||
# Assert
|
||
assert score < 0 # Negative energy
|
||
|
||
def test_formula_neutral_sentiment(self):
|
||
"""Test energy formula with neutral sentiment."""
|
||
# Arrange
|
||
positive_ratio = 0.5
|
||
negative_ratio = 0.5
|
||
volume = 100
|
||
virality = 1.5
|
||
|
||
# Act
|
||
score = (positive_ratio - negative_ratio) * volume * virality
|
||
|
||
# Assert
|
||
assert score == 0.0 # Neutral energy
|
||
|
||
def test_formula_zero_volume(self):
|
||
"""Test energy formula with zero volume."""
|
||
# Arrange
|
||
positive_ratio = 0.7
|
||
negative_ratio = 0.2
|
||
volume = 0
|
||
virality = 1.5
|
||
|
||
# Act
|
||
score = (positive_ratio - negative_ratio) * volume * virality
|
||
|
||
# Assert
|
||
assert score == 0.0 # No volume means no energy
|
||
|
||
|
||
if __name__ == '__main__':
|
||
pytest.main([__file__, '-v'])
|