tests/test_metrics_db.py

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"