File size: 5,313 Bytes
35676b4 7880373 35676b4 7880373 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 | import sqlite3
import pytest
from storage import metrics_db
@pytest.fixture(autouse=True)
def tmp_db(tmp_path, monkeypatch):
monkeypatch.setattr(metrics_db, "DB_PATH", tmp_path / "metrics.db")
def _sample_row(**overrides) -> dict:
base = {
"ticker": "AAPL", "company_name": "Apple Inc.", "filing_date": "2024-11-01",
"period": "FY2024", "form_type": "10-K", "revenue": 391035000000.0,
"revenue_yoy_pct": 2.0, "eps": 6.11, "gross_margin": 0.461,
"operating_margin": 0.314, "free_cash_flow": 108807000000.0,
"guidance_disclosed": 0, "guidance_text": None,
"ingested_at": "2026-05-04T12:00:00",
"shares_diluted": None, "effective_tax_rate": None, "interest_expense": None,
"total_debt": None, "dividends_paid": None, "buybacks": None,
"capex": None, "stockholders_equity": None,
}
return {**base, **overrides}
def test_init_creates_table():
metrics_db.init_db()
with sqlite3.connect(metrics_db.DB_PATH) as conn:
tables = [r[0] for r in conn.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall()]
assert "metrics" in tables
def test_init_migrates_legacy_schema_without_losing_rows():
with sqlite3.connect(metrics_db.DB_PATH) as conn:
conn.execute("""
CREATE TABLE metrics (
ticker TEXT PRIMARY KEY,
period TEXT,
company_name TEXT,
filing_date TEXT,
revenue REAL
)
""")
conn.executemany(
"INSERT INTO metrics VALUES (?, ?, ?, ?, ?)",
[
("AAPL", "FY2023", "Apple Inc.", "2023-11-03", 383285000000.0),
("MSFT", "FY2024", "Microsoft Corp.", "2024-07-30", 245122000000.0),
],
)
metrics_db.init_db()
with sqlite3.connect(metrics_db.DB_PATH) as conn:
rows = conn.execute(
"SELECT ticker, period, company_name, filing_date, revenue "
"FROM metrics ORDER BY ticker"
).fetchall()
pk_columns = [
row[1]
for row in sorted(
conn.execute("PRAGMA table_info(metrics)").fetchall(),
key=lambda row: row[5] or 99,
)
if row[5]
]
archive_rows = conn.execute(
"SELECT ticker, period, revenue FROM metrics_legacy_v1 ORDER BY ticker"
).fetchall()
assert rows == [
("AAPL", "FY2023", "Apple Inc.", "2023-11-03", 383285000000.0),
("MSFT", "FY2024", "Microsoft Corp.", "2024-07-30", 245122000000.0),
]
assert pk_columns == ["ticker", "period"]
assert archive_rows == [
("AAPL", "FY2023", 383285000000.0),
("MSFT", "FY2024", 245122000000.0),
]
def test_legacy_migration_allows_new_periods_and_is_idempotent():
with sqlite3.connect(metrics_db.DB_PATH) as conn:
conn.execute("""
CREATE TABLE metrics (
ticker TEXT PRIMARY KEY,
company_name TEXT,
filing_date TEXT,
revenue REAL
)
""")
conn.execute(
"INSERT INTO metrics VALUES (?, ?, ?, ?)",
("AAPL", "Apple Inc.", "2023-11-03", 383285000000.0),
)
metrics_db.init_db()
metrics_db.init_db()
metrics_db.upsert_metrics(_sample_row(period="FY2024"))
with sqlite3.connect(metrics_db.DB_PATH) as conn:
rows = conn.execute(
"SELECT ticker, period, revenue FROM metrics ORDER BY period"
).fetchall()
archives = conn.execute(
"SELECT name FROM sqlite_master "
"WHERE type = 'table' AND name LIKE 'metrics_legacy_v%'"
).fetchall()
assert rows == [
("AAPL", "FY2024", 391035000000.0),
("AAPL", "LEGACY", 383285000000.0),
]
assert archives == [("metrics_legacy_v1",)]
def test_upsert_and_get():
metrics_db.init_db()
metrics_db.upsert_metrics(_sample_row())
result = metrics_db.get_metrics("AAPL")
assert result["ticker"] == "AAPL"
assert result["revenue"] == 391035000000.0
assert result["form_type"] == "10-K"
def test_upsert_replaces_on_conflict():
metrics_db.init_db()
metrics_db.upsert_metrics(_sample_row(filing_date="2023-11-01", revenue=380000000000.0))
metrics_db.upsert_metrics(_sample_row(filing_date="2024-11-01", revenue=391035000000.0))
result = metrics_db.get_metrics("AAPL")
assert result["filing_date"] == "2024-11-01"
assert result["revenue"] == 391035000000.0
def test_get_missing_returns_none():
metrics_db.init_db()
assert metrics_db.get_metrics("ZZZZ") is None
def test_metric_lineage_round_trip():
metrics_db.init_db()
metrics_db.upsert_metrics(_sample_row(
period_basis="annual",
accession="0000320193-24-000123",
report_date="2024-09-28",
source_url="https://www.sec.gov/example",
metric_contexts={"revenue": {"selection": "exact_annual"}},
quality_warnings=[],
data_quality_status="VERIFIED",
))
result = metrics_db.get_metrics("AAPL")
assert result["metric_contexts"]["revenue"]["selection"] == "exact_annual"
assert result["quality_warnings"] == []
assert result["data_quality_status"] == "VERIFIED"
|