File size: 5,144 Bytes
559c2ff 8dbbf97 559c2ff e6496c0 | 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 | """tests/test_sections_db.py — unit tests for storage/sections_db.py."""
from __future__ import annotations
import sqlite3
import tempfile
from pathlib import Path
from unittest.mock import patch
import pytest
def _tmp_db(tmp_path: Path):
return tmp_path / "sections.db"
def test_upsert_and_get_section(tmp_path):
db_path = _tmp_db(tmp_path)
with patch("storage.sections_db.SECTIONS_DB_PATH", db_path):
from storage.sections_db import init_sections_db, upsert_section, get_section
init_sections_db()
upsert_section("AAPL", "Q12026", "10-Q", "mda", "Revenue grew 8% YoY...")
text = get_section("AAPL", "Q12026", "mda")
assert text == "Revenue grew 8% YoY..."
def test_upsert_overwrites_existing(tmp_path):
db_path = _tmp_db(tmp_path)
with patch("storage.sections_db.SECTIONS_DB_PATH", db_path):
from storage.sections_db import init_sections_db, upsert_section, get_section
init_sections_db()
upsert_section("AAPL", "Q12026", "10-Q", "mda", "v1")
upsert_section("AAPL", "Q12026", "10-Q", "mda", "v2")
assert get_section("AAPL", "Q12026", "mda") == "v2"
def test_get_section_returns_none_when_missing(tmp_path):
db_path = _tmp_db(tmp_path)
with patch("storage.sections_db.SECTIONS_DB_PATH", db_path):
from storage.sections_db import init_sections_db, get_section
init_sections_db()
assert get_section("NVDA", "Q12026", "mda") is None
def test_get_periods_for_ticker(tmp_path):
db_path = _tmp_db(tmp_path)
with patch("storage.sections_db.SECTIONS_DB_PATH", db_path):
from storage.sections_db import init_sections_db, upsert_section, get_periods_for_ticker
init_sections_db()
for period in ["Q12026", "Q42025", "Q32025"]:
upsert_section("NVDA", period, "10-Q", "mda", f"text for {period}")
periods = get_periods_for_ticker("NVDA", form_type="10-Q")
assert "Q12026" in periods
assert "Q42025" in periods
assert "Q32025" in periods
def test_get_periods_returns_empty_when_no_db(tmp_path):
nonexistent = tmp_path / "nosuchfile.db"
with patch("storage.sections_db.SECTIONS_DB_PATH", nonexistent):
from storage.sections_db import get_periods_for_ticker
assert get_periods_for_ticker("AAPL") == []
def test_get_periods_chronological_sort(tmp_path):
"""Q12027 must come before Q32026 — chronological not alphabetical."""
db_path = _tmp_db(tmp_path)
with patch("storage.sections_db.SECTIONS_DB_PATH", db_path):
from storage.sections_db import init_sections_db, upsert_section, get_periods_for_ticker
init_sections_db()
for period in ["Q32026", "Q22026", "Q12027", "Q12026"]:
upsert_section("NVDA", period, "10-Q", "mda", f"text {period}")
periods = get_periods_for_ticker("NVDA", form_type="10-Q")
assert periods[0] == "Q12027", f"Expected Q12027 first, got {periods}"
assert periods[1] == "Q32026", f"Expected Q32026 second, got {periods}"
def test_ticker_is_case_insensitive(tmp_path):
db_path = _tmp_db(tmp_path)
with patch("storage.sections_db.SECTIONS_DB_PATH", db_path):
from storage.sections_db import init_sections_db, upsert_section, get_section
init_sections_db()
upsert_section("aapl", "Q12026", "10-Q", "mda", "lowercase insert")
text = get_section("AAPL", "Q12026", "mda")
assert text == "lowercase insert"
def test_get_recent_transcripts_chronological_with_fy(tmp_path):
"""FY2025 sorts as Q4 2025: between Q32025 and Q12026, oldest first."""
db_path = _tmp_db(tmp_path)
with patch("storage.sections_db.SECTIONS_DB_PATH", db_path):
from storage.sections_db import init_sections_db, upsert_section, get_recent_transcripts
init_sections_db()
upsert_section("AMD", "Q12026", "10-Q", "transcript", "call q1 2026")
upsert_section("AMD", "FY2025", "10-K", "transcript", "call fy 2025")
upsert_section("AMD", "Q32025", "10-Q", "transcript", "call q3 2025")
upsert_section("AMD", "Q22025", "10-Q", "transcript", "call q2 2025")
result = get_recent_transcripts("AMD", n=3)
assert [p for p, _ in result] == ["Q32025", "FY2025", "Q12026"]
assert result[-1][1] == "call q1 2026"
def test_get_recent_transcripts_excludes_empty(tmp_path):
db_path = _tmp_db(tmp_path)
with patch("storage.sections_db.SECTIONS_DB_PATH", db_path):
from storage.sections_db import init_sections_db, upsert_section, get_recent_transcripts
init_sections_db()
upsert_section("AMD", "Q12026", "10-Q", "transcript", "real call")
upsert_section("AMD", "Q42025", "10-Q", "transcript", "") # AV gap
result = get_recent_transcripts("AMD", n=4)
assert [p for p, _ in result] == ["Q12026"]
def test_get_recent_transcripts_no_db(tmp_path):
nonexistent = tmp_path / "nosuchfile.db"
with patch("storage.sections_db.SECTIONS_DB_PATH", nonexistent):
from storage.sections_db import get_recent_transcripts
assert get_recent_transcripts("AAPL") == []
|