395 lines
13 KiB
Python
395 lines
13 KiB
Python
"""
|
|
Tests for Energy Service.
|
|
|
|
This module tests the energy service business logic including:
|
|
- Energy score calculation and storage
|
|
- Retrieval of energy scores
|
|
- Updating and deleting energy scores
|
|
- Filtering and querying
|
|
"""
|
|
|
|
import pytest
|
|
from datetime import datetime
|
|
from app.services.energy_service import (
|
|
calculate_and_store_energy_score,
|
|
get_energy_score,
|
|
get_energy_scores_by_match,
|
|
get_energy_scores_by_team,
|
|
get_energy_score_by_match_and_team,
|
|
update_energy_score,
|
|
delete_energy_score,
|
|
list_energy_scores
|
|
)
|
|
from app.schemas.energy_score import EnergyScoreCalculationRequest, EnergyScoreUpdate
|
|
|
|
|
|
class TestCalculateAndStoreEnergyScore:
|
|
"""Test energy score calculation and storage."""
|
|
|
|
def test_calculate_and_store_with_all_sources(self, db_session):
|
|
"""Test calculation and storage with all sources available."""
|
|
# Arrange
|
|
request = EnergyScoreCalculationRequest(
|
|
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
|
|
energy_score = calculate_and_store_energy_score(db_session, request)
|
|
|
|
# Assert
|
|
assert energy_score is not None
|
|
assert energy_score.match_id == 1
|
|
assert energy_score.team_id == 1
|
|
assert 0 <= energy_score.score <= 100
|
|
assert energy_score.confidence > 0
|
|
assert len(energy_score.sources_used) == 3
|
|
assert energy_score.twitter_score is not None
|
|
assert energy_score.reddit_score is not None
|
|
assert energy_score.rss_score is not None
|
|
|
|
def test_calculate_and_store_with_twitter_only(self, db_session):
|
|
"""Test calculation and storage with only Twitter source."""
|
|
# Arrange
|
|
request = EnergyScoreCalculationRequest(
|
|
match_id=1,
|
|
team_id=1,
|
|
twitter_sentiments=[
|
|
{'compound': 0.5, 'positive': 0.6, 'negative': 0.2, 'neutral': 0.2, 'sentiment': 'positive'}
|
|
]
|
|
)
|
|
|
|
# Act
|
|
energy_score = calculate_and_store_energy_score(db_session, request)
|
|
|
|
# Assert
|
|
assert energy_score is not None
|
|
assert energy_score.match_id == 1
|
|
assert energy_score.team_id == 1
|
|
assert len(energy_score.sources_used) == 1
|
|
assert 'twitter' in energy_score.sources_used
|
|
assert energy_score.twitter_score is not None
|
|
assert energy_score.reddit_score is None
|
|
assert energy_score.rss_score is None
|
|
# Lower confidence in degraded mode
|
|
assert energy_score.confidence < 0.7
|
|
|
|
def test_calculate_and_store_no_sentiment_data(self, db_session):
|
|
"""Test calculation with no sentiment data."""
|
|
# Arrange
|
|
request = EnergyScoreCalculationRequest(
|
|
match_id=1,
|
|
team_id=1,
|
|
twitter_sentiments=[],
|
|
reddit_sentiments=[],
|
|
rss_sentiments=[]
|
|
)
|
|
|
|
# Act
|
|
energy_score = calculate_and_store_energy_score(db_session, request)
|
|
|
|
# Assert
|
|
assert energy_score is not None
|
|
assert energy_score.score == 0.0
|
|
assert energy_score.confidence == 0.0
|
|
assert len(energy_score.sources_used) == 0
|
|
|
|
|
|
class TestGetEnergyScore:
|
|
"""Test retrieval of energy scores."""
|
|
|
|
def test_get_energy_score_by_id(self, db_session):
|
|
"""Test retrieving energy score by ID."""
|
|
# Arrange
|
|
request = EnergyScoreCalculationRequest(
|
|
match_id=1,
|
|
team_id=1,
|
|
twitter_sentiments=[
|
|
{'compound': 0.5, 'positive': 0.6, 'negative': 0.2, 'neutral': 0.2, 'sentiment': 'positive'}
|
|
]
|
|
)
|
|
created = calculate_and_store_energy_score(db_session, request)
|
|
|
|
# Act
|
|
retrieved = get_energy_score(db_session, created.id)
|
|
|
|
# Assert
|
|
assert retrieved is not None
|
|
assert retrieved.id == created.id
|
|
assert retrieved.match_id == created.match_id
|
|
assert retrieved.team_id == created.team_id
|
|
assert retrieved.score == created.score
|
|
|
|
def test_get_energy_score_not_found(self, db_session):
|
|
"""Test retrieving non-existent energy score."""
|
|
# Act
|
|
retrieved = get_energy_score(db_session, 99999)
|
|
|
|
# Assert
|
|
assert retrieved is None
|
|
|
|
|
|
class TestGetEnergyScoresByMatch:
|
|
"""Test retrieval of energy scores by match."""
|
|
|
|
def test_get_energy_scores_by_match_id(self, db_session):
|
|
"""Test retrieving all energy scores for a specific match."""
|
|
# Arrange
|
|
# Create energy scores for different matches
|
|
for match_id in [1, 1, 2]:
|
|
request = EnergyScoreCalculationRequest(
|
|
match_id=match_id,
|
|
team_id=1,
|
|
twitter_sentiments=[
|
|
{'compound': 0.5, 'positive': 0.6, 'negative': 0.2, 'neutral': 0.2, 'sentiment': 'positive'}
|
|
]
|
|
)
|
|
calculate_and_store_energy_score(db_session, request)
|
|
|
|
# Act
|
|
scores_for_match_1 = get_energy_scores_by_match(db_session, 1)
|
|
scores_for_match_2 = get_energy_scores_by_match(db_session, 2)
|
|
|
|
# Assert
|
|
assert len(scores_for_match_1) == 2
|
|
assert all(score.match_id == 1 for score in scores_for_match_1)
|
|
assert len(scores_for_match_2) == 1
|
|
assert scores_for_match_2[0].match_id == 2
|
|
|
|
|
|
class TestGetEnergyScoresByTeam:
|
|
"""Test retrieval of energy scores by team."""
|
|
|
|
def test_get_energy_scores_by_team_id(self, db_session):
|
|
"""Test retrieving all energy scores for a specific team."""
|
|
# Arrange
|
|
# Create energy scores for different teams
|
|
for team_id in [1, 1, 2]:
|
|
request = EnergyScoreCalculationRequest(
|
|
match_id=1,
|
|
team_id=team_id,
|
|
twitter_sentiments=[
|
|
{'compound': 0.5, 'positive': 0.6, 'negative': 0.2, 'neutral': 0.2, 'sentiment': 'positive'}
|
|
]
|
|
)
|
|
calculate_and_store_energy_score(db_session, request)
|
|
|
|
# Act
|
|
scores_for_team_1 = get_energy_scores_by_team(db_session, 1)
|
|
scores_for_team_2 = get_energy_scores_by_team(db_session, 2)
|
|
|
|
# Assert
|
|
assert len(scores_for_team_1) == 2
|
|
assert all(score.team_id == 1 for score in scores_for_team_1)
|
|
assert len(scores_for_team_2) == 1
|
|
assert scores_for_team_2[0].team_id == 2
|
|
|
|
|
|
class TestGetEnergyScoreByMatchAndTeam:
|
|
"""Test retrieval of most recent energy score by match and team."""
|
|
|
|
def test_get_energy_score_by_match_and_team(self, db_session):
|
|
"""Test retrieving most recent energy score for match and team."""
|
|
# Arrange
|
|
request = EnergyScoreCalculationRequest(
|
|
match_id=1,
|
|
team_id=1,
|
|
twitter_sentiments=[
|
|
{'compound': 0.5, 'positive': 0.6, 'negative': 0.2, 'neutral': 0.2, 'sentiment': 'positive'}
|
|
]
|
|
)
|
|
calculate_and_store_energy_score(db_session, request)
|
|
|
|
# Act
|
|
retrieved = get_energy_score_by_match_and_team(db_session, 1, 1)
|
|
|
|
# Assert
|
|
assert retrieved is not None
|
|
assert retrieved.match_id == 1
|
|
assert retrieved.team_id == 1
|
|
|
|
def test_get_energy_score_by_match_and_team_not_found(self, db_session):
|
|
"""Test retrieving non-existent energy score."""
|
|
# Act
|
|
retrieved = get_energy_score_by_match_and_team(db_session, 999, 999)
|
|
|
|
# Assert
|
|
assert retrieved is None
|
|
|
|
|
|
class TestUpdateEnergyScore:
|
|
"""Test updating energy scores."""
|
|
|
|
def test_update_energy_score(self, db_session):
|
|
"""Test updating an existing energy score."""
|
|
# Arrange
|
|
request = EnergyScoreCalculationRequest(
|
|
match_id=1,
|
|
team_id=1,
|
|
twitter_sentiments=[
|
|
{'compound': 0.5, 'positive': 0.6, 'negative': 0.2, 'neutral': 0.2, 'sentiment': 'positive'}
|
|
]
|
|
)
|
|
created = calculate_and_store_energy_score(db_session, request)
|
|
|
|
# Act
|
|
update = EnergyScoreUpdate(
|
|
score=75.0,
|
|
confidence=0.9
|
|
)
|
|
updated = update_energy_score(db_session, created.id, update)
|
|
|
|
# Assert
|
|
assert updated is not None
|
|
assert updated.id == created.id
|
|
assert updated.score == 75.0
|
|
assert updated.confidence == 0.9
|
|
assert updated.updated_at > created.updated_at
|
|
|
|
def test_update_energy_score_not_found(self, db_session):
|
|
"""Test updating non-existent energy score."""
|
|
# Arrange
|
|
update = EnergyScoreUpdate(score=75.0, confidence=0.9)
|
|
|
|
# Act
|
|
updated = update_energy_score(db_session, 99999, update)
|
|
|
|
# Assert
|
|
assert updated is None
|
|
|
|
|
|
class TestDeleteEnergyScore:
|
|
"""Test deleting energy scores."""
|
|
|
|
def test_delete_energy_score(self, db_session):
|
|
"""Test deleting an existing energy score."""
|
|
# Arrange
|
|
request = EnergyScoreCalculationRequest(
|
|
match_id=1,
|
|
team_id=1,
|
|
twitter_sentiments=[
|
|
{'compound': 0.5, 'positive': 0.6, 'negative': 0.2, 'neutral': 0.2, 'sentiment': 'positive'}
|
|
]
|
|
)
|
|
created = calculate_and_store_energy_score(db_session, request)
|
|
|
|
# Act
|
|
deleted = delete_energy_score(db_session, created.id)
|
|
|
|
# Assert
|
|
assert deleted is True
|
|
retrieved = get_energy_score(db_session, created.id)
|
|
assert retrieved is None
|
|
|
|
def test_delete_energy_score_not_found(self, db_session):
|
|
"""Test deleting non-existent energy score."""
|
|
# Act
|
|
deleted = delete_energy_score(db_session, 99999)
|
|
|
|
# Assert
|
|
assert deleted is False
|
|
|
|
|
|
class TestListEnergyScores:
|
|
"""Test listing and filtering energy scores."""
|
|
|
|
def test_list_energy_scores_default(self, db_session):
|
|
"""Test listing energy scores with default parameters."""
|
|
# Arrange
|
|
for i in range(5):
|
|
request = EnergyScoreCalculationRequest(
|
|
match_id=i,
|
|
team_id=1,
|
|
twitter_sentiments=[
|
|
{'compound': 0.5, 'positive': 0.6, 'negative': 0.2, 'neutral': 0.2, 'sentiment': 'positive'}
|
|
]
|
|
)
|
|
calculate_and_store_energy_score(db_session, request)
|
|
|
|
# Act
|
|
scores = list_energy_scores(db_session)
|
|
|
|
# Assert
|
|
assert len(scores) == 5
|
|
|
|
def test_list_energy_scores_with_match_filter(self, db_session):
|
|
"""Test listing energy scores filtered by match ID."""
|
|
# Arrange
|
|
for match_id in [1, 1, 2]:
|
|
request = EnergyScoreCalculationRequest(
|
|
match_id=match_id,
|
|
team_id=1,
|
|
twitter_sentiments=[
|
|
{'compound': 0.5, 'positive': 0.6, 'negative': 0.2, 'neutral': 0.2, 'sentiment': 'positive'}
|
|
]
|
|
)
|
|
calculate_and_store_energy_score(db_session, request)
|
|
|
|
# Act
|
|
scores = list_energy_scores(db_session, match_id=1)
|
|
|
|
# Assert
|
|
assert len(scores) == 2
|
|
assert all(score.match_id == 1 for score in scores)
|
|
|
|
def test_list_energy_scores_with_score_filter(self, db_session):
|
|
"""Test listing energy scores filtered by score range."""
|
|
# Arrange
|
|
for score in [30.0, 50.0, 70.0, 90.0]:
|
|
request = EnergyScoreCalculationRequest(
|
|
match_id=1,
|
|
team_id=1,
|
|
twitter_sentiments=[
|
|
{'compound': 0.5, 'positive': 0.6, 'negative': 0.2, 'neutral': 0.2, 'sentiment': 'positive'}
|
|
]
|
|
)
|
|
created = calculate_and_store_energy_score(db_session, request)
|
|
# Manually update score for testing
|
|
created.score = score
|
|
db_session.commit()
|
|
|
|
# Act
|
|
scores = list_energy_scores(db_session, min_score=40.0, max_score=80.0)
|
|
|
|
# Assert
|
|
assert len(scores) == 2
|
|
assert all(40.0 <= score.score <= 80.0 for score in scores)
|
|
|
|
def test_list_energy_scores_with_pagination(self, db_session):
|
|
"""Test listing energy scores with pagination."""
|
|
# Arrange
|
|
for i in range(10):
|
|
request = EnergyScoreCalculationRequest(
|
|
match_id=i,
|
|
team_id=1,
|
|
twitter_sentiments=[
|
|
{'compound': 0.5, 'positive': 0.6, 'negative': 0.2, 'neutral': 0.2, 'sentiment': 'positive'}
|
|
]
|
|
)
|
|
calculate_and_store_energy_score(db_session, request)
|
|
|
|
# Act
|
|
page1 = list_energy_scores(db_session, limit=5, offset=0)
|
|
page2 = list_energy_scores(db_session, limit=5, offset=5)
|
|
|
|
# Assert
|
|
assert len(page1) == 5
|
|
assert len(page2) == 5
|
|
# Verify no overlap
|
|
page1_ids = [score.id for score in page1]
|
|
page2_ids = [score.id for score in page2]
|
|
assert set(page1_ids).isdisjoint(set(page2_ids))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
pytest.main([__file__, '-v'])
|