"""Tests for database models and schema."""
import pytest
from datetime import datetime, timezone
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from unittest.mock import patch, MagicMock
from src.models.database import Base, News, NewsAnalysis
from src.models.database import NewsRepository, AnalysisRepository
class TestDatabaseModels:
"""Test database models and schema."""
@pytest.fixture
def engine(self):
"""Create test database engine."""
# Use in-memory SQLite for testing
engine = create_engine("sqlite:///:memory:", echo=False)
Base.metadata.create_all(engine)
return engine
@pytest.fixture
def session(self, engine):
"""Create test database session."""
Session = sessionmaker(bind=engine)
session = Session()
yield session
session.close()
def test_news_model_creation(self):
"""Test News model can be created with required fields."""
news = News(
source="naver",
title="Test News Title",
content="Test news content",
url="https://news.naver.com/test",
published_at=datetime.now(timezone.utc),
hash="test_hash_123"
)
assert news.source == "naver"
assert news.title == "Test News Title"
assert news.content == "Test news content"
assert news.url == "https://news.naver.com/test"
assert news.hash == "test_hash_123"
assert isinstance(news.published_at, datetime)
assert news.collected_at is None # Should be set by database
def test_news_model_validation(self):
"""Test News model validates required fields."""
# Test that required fields must be provided for database operations
# SQLAlchemy models don't raise TypeError at instantiation, but at commit
news = News() # This is allowed
assert news.source is None
# Test with valid data
news = News(
source="naver",
title="Test",
content="Content",
url="invalid-url",
published_at=datetime.now(timezone.utc),
hash="hash123"
)
# For now, just check it doesn't crash
assert news.url == "invalid-url"
def test_news_analysis_model(self):
"""Test NewsAnalysis model creation."""
analysis = NewsAnalysis(
news_id=1,
sentiment_score=0.75,
sentiment_label="positive",
impact_score=0.60,
entities={"companies": ["AAPL", "GOOGL"], "topics": ["tech", "earnings"]},
keywords=["technology", "earnings", "profit"]
)
assert analysis.news_id == 1
assert analysis.sentiment_score == 0.75
assert analysis.sentiment_label == "positive"
assert analysis.impact_score == 0.60
assert analysis.entities == {"companies": ["AAPL", "GOOGL"], "topics": ["tech", "earnings"]}
assert analysis.keywords == ["technology", "earnings", "profit"]
def test_news_database_operations(self, session):
"""Test basic database operations for News model."""
# Create
news = News(
source="naver",
title="Database Test News",
content="Test content for database",
url="https://example.com/news/1",
published_at=datetime.now(timezone.utc),
hash="db_test_hash"
)
session.add(news)
session.commit()
assert news.id is not None
assert news.collected_at is not None # Should be auto-set
# Read
retrieved_news = session.query(News).filter_by(hash="db_test_hash").first()
assert retrieved_news is not None
assert retrieved_news.title == "Database Test News"
# Update
retrieved_news.title = "Updated News Title"
session.commit()
updated_news = session.query(News).filter_by(hash="db_test_hash").first()
assert updated_news.title == "Updated News Title"
# Delete
session.delete(updated_news)
session.commit()
deleted_news = session.query(News).filter_by(hash="db_test_hash").first()
assert deleted_news is None
def test_news_analysis_relationship(self, session):
"""Test relationship between News and NewsAnalysis."""
# Create news
news = News(
source="test",
title="Relationship Test",
content="Content",
url="https://example.com/rel",
published_at=datetime.now(timezone.utc),
hash="rel_test_hash"
)
session.add(news)
session.commit()
# Create analysis
analysis = NewsAnalysis(
news_id=news.id,
sentiment_score=0.5,
sentiment_label="neutral",
impact_score=0.3
)
session.add(analysis)
session.commit()
# Test relationship
assert analysis.news == news
assert news.analysis == analysis
def test_hash_uniqueness_constraint(self, session):
"""Test that news hash field has uniqueness constraint."""
# Create first news
news1 = News(
source="test",
title="First News",
content="Content 1",
url="https://example.com/1",
published_at=datetime.now(timezone.utc),
hash="duplicate_hash"
)
session.add(news1)
session.commit()
# Try to create second news with same hash
news2 = News(
source="test",
title="Second News",
content="Content 2",
url="https://example.com/2",
published_at=datetime.now(timezone.utc),
hash="duplicate_hash"
)
session.add(news2)
# This should raise an integrity error
with pytest.raises(Exception): # IntegrityError in real DB
session.commit()
class TestNewsRepository:
"""Test NewsRepository class."""
@pytest.fixture
def mock_session(self):
"""Create mock database session."""
return MagicMock()
@pytest.fixture
def repository(self, mock_session):
"""Create NewsRepository instance."""
return NewsRepository(mock_session)
def test_create_news(self, repository, mock_session):
"""Test creating news through repository."""
news_data = {
"source": "naver",
"title": "Repo Test News",
"content": "Content",
"url": "https://example.com/repo",
"published_at": datetime.now(timezone.utc),
"hash": "repo_hash"
}
result = repository.create_news(**news_data)
mock_session.add.assert_called_once()
mock_session.commit.assert_called_once()
assert isinstance(result, News)
assert result.title == "Repo Test News"
def test_get_news_by_hash(self, repository, mock_session):
"""Test retrieving news by hash."""
mock_news = News(
source="test", title="Test", content="Content",
url="https://test.com", hash="test_hash",
published_at=datetime.now(timezone.utc)
)
mock_session.query.return_value.filter_by.return_value.first.return_value = mock_news
result = repository.get_by_hash("test_hash")
mock_session.query.assert_called_once_with(News)
mock_session.query.return_value.filter_by.assert_called_with(hash="test_hash")
assert result == mock_news
def test_get_recent_news(self, repository, mock_session):
"""Test getting recent news with filters."""
mock_query = mock_session.query.return_value
mock_query.filter.return_value = mock_query
mock_query.order_by.return_value = mock_query
mock_query.limit.return_value.all.return_value = []
result = repository.get_recent_news(limit=10, hours=24, keyword="test")
mock_session.query.assert_called_once_with(News)
# Should have applied filters and ordering
assert mock_query.filter.called
assert mock_query.order_by.called
assert mock_query.limit.called
def test_bulk_create_news(self, repository, mock_session):
"""Test bulk creation of news items."""
news_items = [
{"source": "naver", "title": "News 1", "content": "Content 1",
"url": "https://example.com/1", "hash": "hash1",
"published_at": datetime.now(timezone.utc)},
{"source": "daum", "title": "News 2", "content": "Content 2",
"url": "https://example.com/2", "hash": "hash2",
"published_at": datetime.now(timezone.utc)},
]
result = repository.bulk_create_news(news_items)
# Should have added multiple items
assert mock_session.add.call_count == 2
mock_session.commit.assert_called_once()
assert len(result) == 2
class TestAnalysisRepository:
"""Test AnalysisRepository class."""
@pytest.fixture
def mock_session(self):
"""Create mock database session."""
return MagicMock()
@pytest.fixture
def repository(self, mock_session):
"""Create AnalysisRepository instance."""
return AnalysisRepository(mock_session)
def test_create_analysis(self, repository, mock_session):
"""Test creating analysis through repository."""
analysis_data = {
"news_id": 1,
"sentiment_score": 0.8,
"sentiment_label": "positive",
"impact_score": 0.6,
"entities": {"companies": ["AAPL"]},
"keywords": ["tech", "earnings"]
}
result = repository.create_analysis(**analysis_data)
mock_session.add.assert_called_once()
mock_session.commit.assert_called_once()
assert isinstance(result, NewsAnalysis)
def test_get_analysis_by_news_id(self, repository, mock_session):
"""Test retrieving analysis by news ID."""
mock_analysis = NewsAnalysis(news_id=1, sentiment_score=0.5, sentiment_label="neutral")
mock_session.query.return_value.filter_by.return_value.first.return_value = mock_analysis
result = repository.get_by_news_id(1)
mock_session.query.assert_called_once_with(NewsAnalysis)
mock_session.query.return_value.filter_by.assert_called_with(news_id=1)
assert result == mock_analysis
