The test built the importer with today=date.today() while running against a fixed local clone with do_pull=False, so its forward horizon shrank by a day per real day. It has now decayed past the initial-load gate -- 19d against the 21d MIN_FORWARD_HORIZON_DAYS floor -- and would have kept failing, worse each day. Anchors today to the clone's own calendar (max reporting date across the seeded dot-free symbols, minus 35 days, mirroring the ~35d horizon the importer's own comment cites) and uses that date in the forward-calendar assertion. Also surfaces run.error_details on failure, which is how the cause was found. Test-only. MIN_FORWARD_HORIZON_DAYS and the importer are untouched: production pulls fresh data on every run and was never affected by this. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
281 lines
10 KiB
Python
281 lines
10 KiB
Python
"""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, timedelta
|
|
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, timeout=None):
|
|
self.pulled = True
|
|
|
|
async def current_commit(self, repo_dir, *, binary, timeout=None):
|
|
return self.commit
|
|
|
|
async def query_csv(self, repo_dir, sql, *, binary, timeout=None):
|
|
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-20")],
|
|
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, 20)
|
|
|
|
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, 20))
|
|
# 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-20"), _cal("BRK.B", "2026-08-25")], # 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-20")], history=[], commit="c1")
|
|
await run_import(_importer(fake1), engine=engine)
|
|
|
|
fake2 = FakeDolt(calendar=[_cal("AAPL", "2026-08-27")], 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, 27)} # 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-20")], 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-20")], 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, 20) 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-20"),
|
|
_cal("MSFT", "2026-08-21"),
|
|
_cal("NVDA", "2026-08-22"),
|
|
_cal("AMZN", "2026-08-23"),
|
|
],
|
|
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-20")], 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)
|
|
# A few tickers spanning near + further-out reporters so the initial-load
|
|
# forward-horizon gate (>= 21d) is satisfied on the fixed clone.
|
|
await _seed_tickers(factory, ["AAPL", "MSFT", "NVDA", "JPM", "BRK.B"])
|
|
|
|
# "today" is anchored to the clone, NOT the wall clock. The clone is fixed
|
|
# and do_pull=False, so a wall-clock today makes this test decay: the
|
|
# forward horizon shrinks a day per real day and eventually trips the
|
|
# >= 21d gate (it did, at 19d). Anchoring keeps it time-stable. Production
|
|
# pulls fresh data and is unaffected. Dot-free symbols only, so the query
|
|
# needs no symbol normalisation.
|
|
rows = await dolt_client.query_csv(
|
|
_CLONE_DIR,
|
|
"SELECT MAX(`date`) AS max_date FROM earnings_calendar "
|
|
"WHERE act_symbol IN ('AAPL', 'MSFT', 'NVDA', 'JPM')",
|
|
binary=_DOLT_BIN,
|
|
)
|
|
max_date = date.fromisoformat(rows[0]["max_date"])
|
|
today = max_date - timedelta(days=35) # ~35d horizon, per the importer's note
|
|
|
|
imp = DoltEarningsImporter(
|
|
repo_dir=_CLONE_DIR, binary=_DOLT_BIN, today=today, do_pull=False, dolt=dolt_client
|
|
)
|
|
run = await run_import(imp, engine=engine)
|
|
|
|
assert run.status == STATUS_PROMOTED, run.error_details
|
|
events = await _events(factory)
|
|
assert events, "no earnings parsed from the real clone"
|
|
assert any(e.announce_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"
|