204cd0e75b
- PR1: Project scaffold, DB models, price loader - PR2: Congressional trade ingestion (House Stock Watcher) - PR3: Security enrichment + deployment infrastructure - 37 passing tests, 87%+ coverage - Docker + Proxmox deployment ready - Complete documentation - Works 100% offline with fixtures
130 lines
4.0 KiB
Python
130 lines
4.0 KiB
Python
"""
|
|
Tests for database models.
|
|
"""
|
|
|
|
from datetime import date
|
|
from decimal import Decimal
|
|
|
|
from sqlalchemy import select
|
|
|
|
from pote.db.models import Price, Security, Trade
|
|
|
|
|
|
def test_create_official(test_db_session, sample_official):
|
|
"""Test creating an official."""
|
|
assert sample_official.id is not None
|
|
assert sample_official.name == "Jane Doe"
|
|
assert sample_official.chamber == "Senate"
|
|
assert sample_official.party == "Independent"
|
|
assert sample_official.state == "CA"
|
|
|
|
|
|
def test_create_security(test_db_session, sample_security):
|
|
"""Test creating a security."""
|
|
assert sample_security.id is not None
|
|
assert sample_security.ticker == "AAPL"
|
|
assert sample_security.name == "Apple Inc."
|
|
assert sample_security.sector == "Technology"
|
|
|
|
|
|
def test_create_trade(test_db_session, sample_trade):
|
|
"""Test creating a trade with relationships."""
|
|
assert sample_trade.id is not None
|
|
assert sample_trade.official_id is not None
|
|
assert sample_trade.security_id is not None
|
|
assert sample_trade.side == "buy"
|
|
assert sample_trade.value_min == Decimal("15000.00")
|
|
|
|
# Test relationships
|
|
assert sample_trade.official.name == "Jane Doe"
|
|
assert sample_trade.security.ticker == "AAPL"
|
|
|
|
|
|
def test_create_price(test_db_session, sample_price):
|
|
"""Test creating a price record."""
|
|
assert sample_price.id is not None
|
|
assert sample_price.close == Decimal("181.25")
|
|
assert sample_price.volume == 50000000
|
|
assert sample_price.security.ticker == "AAPL"
|
|
|
|
|
|
def test_unique_constraints(test_db_session, sample_security):
|
|
"""Test that unique constraints work."""
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
# Try to create duplicate security with same ticker
|
|
dup_security = Security(ticker="AAPL", name="Apple Duplicate")
|
|
test_db_session.add(dup_security)
|
|
|
|
try:
|
|
test_db_session.commit()
|
|
assert False, "Should have raised IntegrityError"
|
|
except IntegrityError:
|
|
test_db_session.rollback()
|
|
# Expected behavior
|
|
|
|
|
|
def test_price_unique_per_security_date(test_db_session, sample_security):
|
|
"""Test that we can't have duplicate prices for same security/date."""
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
price1 = Price(
|
|
security_id=sample_security.id,
|
|
date=date(2024, 1, 1),
|
|
close=Decimal("100.00"),
|
|
)
|
|
test_db_session.add(price1)
|
|
test_db_session.commit()
|
|
|
|
price2 = Price(
|
|
security_id=sample_security.id,
|
|
date=date(2024, 1, 1),
|
|
close=Decimal("101.00"),
|
|
)
|
|
test_db_session.add(price2)
|
|
|
|
try:
|
|
test_db_session.commit()
|
|
assert False, "Should have raised IntegrityError"
|
|
except IntegrityError:
|
|
test_db_session.rollback()
|
|
# Expected behavior
|
|
|
|
|
|
def test_trade_queries(test_db_session, sample_official, sample_security):
|
|
"""Test querying trades by official and date range."""
|
|
# Create multiple trades
|
|
trades_data = [
|
|
{"date": date(2024, 1, 10), "side": "buy"},
|
|
{"date": date(2024, 1, 15), "side": "sell"},
|
|
{"date": date(2024, 2, 1), "side": "buy"},
|
|
]
|
|
|
|
for i, td in enumerate(trades_data):
|
|
trade = Trade(
|
|
official_id=sample_official.id,
|
|
security_id=sample_security.id,
|
|
source="test",
|
|
external_id=f"test-{i}",
|
|
transaction_date=td["date"],
|
|
side=td["side"],
|
|
value_min=Decimal("10000.00"),
|
|
value_max=Decimal("50000.00"),
|
|
)
|
|
test_db_session.add(trade)
|
|
test_db_session.commit()
|
|
|
|
# Query trades in January
|
|
stmt = (
|
|
select(Trade)
|
|
.where(Trade.official_id == sample_official.id)
|
|
.where(Trade.transaction_date >= date(2024, 1, 1))
|
|
.where(Trade.transaction_date < date(2024, 2, 1))
|
|
.order_by(Trade.transaction_date)
|
|
)
|
|
jan_trades = test_db_session.scalars(stmt).all()
|
|
|
|
assert len(jan_trades) == 2
|
|
assert jan_trades[0].transaction_date == date(2024, 1, 10)
|
|
assert jan_trades[1].transaction_date == date(2024, 1, 15)
|