"""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") == []