signal-platform/tests/unit/test_dolt_earnings_importer.py

"""Integration tests for the DoltHub earnings importer, driven through the real
import framework with a fake dolt client (no subprocess, no clone).

Covers the load-bearing behaviors: symbol-normalized join to the tracked
universe, calendar<->history pairing (matched → EPS, unmatched → null), the
destructive-but-safe reschedule/cancel promotion, past rows never deleted, and
the fail-closed forward-calendar gates that guard the destructive promote.
"""

from __future__ import annotations

import os
import shutil
import tempfile
from datetime import date
from pathlib import Path

import pytest
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

from app.database import Base
import app.models  # noqa: F401
from app.models.earnings_event import EarningsEvent
from app.models.ticker import Ticker
from app.services.data_import import STATUS_FAILED, STATUS_PROMOTED, run_import
from app.services.dolt_earnings_importer import DoltEarningsImporter

TODAY = date(2026, 7, 22)


@pytest.fixture
async def engine():
    fd, path = tempfile.mkstemp(suffix=".db")
    os.close(fd)
    eng = create_async_engine(f"sqlite+aiosqlite:///{path}")
    async with eng.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    try:
        yield eng
    finally:
        await eng.dispose()
        try:
            os.unlink(path)
        except OSError:
            pass


def _factory(engine):
    return async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)


class FakeDolt:
    def __init__(self, calendar, history, commit="c1"):
        self.calendar = calendar
        self.history = history
        self.commit = commit
        self.pulled = False

    async def pull(self, repo_dir, *, binary):
        self.pulled = True

    async def current_commit(self, repo_dir, *, binary):
        return self.commit

    async def query_csv(self, repo_dir, sql, *, binary):
        if "earnings_calendar" in sql:
            return self.calendar
        if "eps_history" in sql:
            return self.history
        return []


def _cal(symbol, d, when="After market close"):
    return {"act_symbol": symbol, "date": d, "when": when}


def _hist(symbol, pe, reported, estimate):
    return {"act_symbol": symbol, "period_end_date": pe, "reported": str(reported), "estimate": str(estimate)}


def _importer(fake, commit=None):
    if commit:
        fake.commit = commit
    return DoltEarningsImporter(
        repo_dir="unused", binary="unused", today=TODAY, do_pull=False, dolt=fake
    )


async def _seed_tickers(factory, symbols):
    async with factory() as s:
        for sym in symbols:
            s.add(Ticker(symbol=sym))
        await s.commit()
    async with factory() as s:
        return {sym: tid for tid, sym in (await s.execute(select(Ticker.id, Ticker.symbol))).all()}


async def _events(factory):
    async with factory() as s:
        rows = (
            await s.execute(select(EarningsEvent).order_by(EarningsEvent.announce_date))
        ).scalars().all()
        return list(rows)


# ---------------------------------------------------------------------------


async def test_stage_and_promote_basic(engine):
    factory = _factory(engine)
    await _seed_tickers(factory, ["AAPL"])
    fake = FakeDolt(
        calendar=[_cal("AAPL", "2026-05-01"), _cal("AAPL", "2026-08-01")],
        history=[_hist("AAPL", "2026-03-31", 1.5, 1.4)],  # only the reported quarter
    )
    run = await run_import(_importer(fake), engine=engine)

    assert run.status == STATUS_PROMOTED
    assert run.source_max_date == date(2026, 8, 1)

    events = await _events(factory)
    assert len(events) == 2
    past = next(e for e in events if e.announce_date == date(2026, 5, 1))
    future = next(e for e in events if e.announce_date == date(2026, 8, 1))
    # past announcement paired to the reported quarter
    assert past.eps_actual == 1.5 and past.eps_estimate == 1.4
    assert past.period_end == date(2026, 3, 31) and past.session == "amc"
    assert past.import_run_id == run.id
    # future announcement has no results yet → null EPS/period, session kept
    assert future.eps_actual is None and future.period_end is None
    assert future.session == "amc"


async def test_symbol_normalisation_join(engine):
    factory = _factory(engine)
    ids = await _seed_tickers(factory, ["AAPL", "BRK.B"])
    fake = FakeDolt(
        calendar=[_cal("AAPL", "2026-08-01"), _cal("BRK.B", "2026-08-05")],  # dotted source symbol
        history=[],
    )
    run = await run_import(_importer(fake), engine=engine)
    assert run.status == STATUS_PROMOTED

    events = await _events(factory)
    mapped = {e.ticker_id for e in events}
    assert mapped == {ids["AAPL"], ids["BRK.B"]}  # dotted BRK.B joined via normalization


async def test_reschedule_moves_future_row(engine):
    factory = _factory(engine)
    await _seed_tickers(factory, ["AAPL"])

    fake1 = FakeDolt(calendar=[_cal("AAPL", "2026-08-01")], history=[], commit="c1")
    await run_import(_importer(fake1), engine=engine)

    fake2 = FakeDolt(calendar=[_cal("AAPL", "2026-08-08")], history=[], commit="c2")  # moved
    run2 = await run_import(_importer(fake2), engine=engine)
    assert run2.status == STATUS_PROMOTED

    dates = {e.announce_date for e in await _events(factory)}
    assert dates == {date(2026, 8, 8)}  # old future date gone, new one present


async def test_cancellation_removes_future_row(engine):
    factory = _factory(engine)
    await _seed_tickers(factory, ["AAPL"])

    fake1 = FakeDolt(
        calendar=[_cal("AAPL", "2026-08-01"), _cal("AAPL", "2026-08-15")], history=[], commit="c1"
    )
    await run_import(_importer(fake1), engine=engine)

    fake2 = FakeDolt(calendar=[_cal("AAPL", "2026-08-15")], history=[], commit="c2")  # 08-01 cancelled
    await run_import(_importer(fake2), engine=engine)

    dates = {e.announce_date for e in await _events(factory)}
    assert dates == {date(2026, 8, 15)}


async def test_past_row_never_deleted(engine):
    factory = _factory(engine)
    await _seed_tickers(factory, ["AAPL"])

    fake1 = FakeDolt(
        calendar=[_cal("AAPL", "2026-05-01"), _cal("AAPL", "2026-08-01")], history=[], commit="c1"
    )
    await run_import(_importer(fake1), engine=engine)

    # Second import's calendar omits the past date but keeps a future one.
    fake2 = FakeDolt(calendar=[_cal("AAPL", "2026-08-01")], history=[], commit="c2")
    await run_import(_importer(fake2), engine=engine)

    dates = {e.announce_date for e in await _events(factory)}
    assert date(2026, 5, 1) in dates  # past result survived
    assert date(2026, 8, 1) in dates


async def test_validate_fails_when_no_future(engine):
    factory = _factory(engine)
    await _seed_tickers(factory, ["AAPL"])
    fake = FakeDolt(calendar=[_cal("AAPL", "2026-05-01")], history=[])  # only past

    run = await run_import(_importer(fake), engine=engine)

    assert run.status == STATUS_FAILED
    assert "future" in (run.error_details or "")
    assert len(await _events(factory)) == 0  # nothing promoted


async def test_validate_fails_on_forward_collapse(engine):
    factory = _factory(engine)
    await _seed_tickers(factory, ["AAPL", "MSFT", "NVDA", "AMZN"])

    fake1 = FakeDolt(
        calendar=[
            _cal("AAPL", "2026-08-01"),
            _cal("MSFT", "2026-08-02"),
            _cal("NVDA", "2026-08-03"),
            _cal("AMZN", "2026-08-04"),
        ],
        history=[],
        commit="c1",
    )
    await run_import(_importer(fake1), engine=engine)
    assert len([e for e in await _events(factory) if e.announce_date > TODAY]) == 4

    # Only one future row now → 1 < 50% of 4 → fail-closed, no destructive wipe.
    fake2 = FakeDolt(calendar=[_cal("AAPL", "2026-08-01")], history=[], commit="c2")
    run2 = await run_import(_importer(fake2), engine=engine)

    assert run2.status == STATUS_FAILED
    assert "collapsed" in (run2.error_details or "")
    assert len([e for e in await _events(factory) if e.announce_date > TODAY]) == 4  # preserved


# --- Real-clone smoke test: exercises the actual dolt subprocess + parse + align
# against the local clone. Skips in CI / anywhere the binary or clone is absent.

_DOLT_BIN = os.environ.get("DOLT_BINARY") or shutil.which("dolt") or r"C:\Program Files\Dolt\bin\dolt.exe"
_CLONE_DIR = Path("dolt-data/earnings")


@pytest.mark.skipif(
    not (Path(_DOLT_BIN).exists() and _CLONE_DIR.exists()),
    reason="real dolt binary / earnings clone not available",
)
async def test_real_clone_smoke(engine):
    from app.services import dolt_client

    factory = _factory(engine)
    await _seed_tickers(factory, ["AAPL", "MSFT", "JPM"])
    imp = DoltEarningsImporter(
        repo_dir=_CLONE_DIR, binary=_DOLT_BIN, today=date.today(), do_pull=False, dolt=dolt_client
    )
    run = await run_import(imp, engine=engine)

    assert run.status == STATUS_PROMOTED
    events = await _events(factory)
    assert events, "no earnings parsed from the real clone"
    assert any(e.announce_date > date.today() for e in events), "no forward calendar"
    assert any(e.eps_actual is not None for e in events), "no calendar<->history pairing"