services/ingest/tests/test_calendar.py

689 lines · python

"""
Tests for ingest/calendar.py
 
Coverage:
  • parse_effective_dates()  -  all regex patterns, mixed text, no false positives
  • _normalize_date()        -  ISO passthrough, month-name parsing, bad input
  • ensure_schema()          -  table + index creation, effective_on column
  • _aggregate_effective()   -  pulls FR rules with effective_on from DB
  • _aggregate_enacted()     -  pulls BecameLaw actions from DB
  • _aggregate_votes()       -  pulls from votes table
  • _aggregate_bill_actions()  -  introduced / committee / floor
  • _aggregate_apportionments()  -  pulls from apportionments table
  • _aggregate_effective_parsed()  -  regex over action text
  • run() idempotency         -  double-run does not duplicate rows
"""
 
import os
import sqlite3
import tempfile
 
import pytest
 
from ingest.db import get_connection, init_schema
from ingest.calendar import (
    EFFECTIVE_DATE_PATTERNS,
    _normalize_date,
    _aggregate_effective,
    _aggregate_enacted,
    _aggregate_votes,
    _aggregate_bill_actions,
    _aggregate_apportionments,
    _aggregate_effective_parsed,
    ensure_schema,
    parse_effective_dates,
    run,
)
 
 
 
def _make_db() -> tuple[sqlite3.Connection, str]:
    """Return (conn, path) for a fresh temp DB with full schema."""
    path = tempfile.mktemp(suffix=".db")
    conn = get_connection(path)
    init_schema(conn)
    ensure_schema(conn)
    return conn, path
 
 
def _seed_executive_action(
    conn: sqlite3.Connection,
    external_id: str = "2026-99999",
    doc_type: str = "final_rule",
    title: str = "Test Rule",
    effective_on: str | None = "2026-08-01",
    source_url: str = "https://www.federalregister.gov/documents/2026/06/26/2026-99999/test-rule",
) -> int:
    """Insert one executive_actions row; return its id."""
    conn.execute(
        """
        INSERT INTO executive_actions
            (source, external_id, type, title, fr_doc_number,
             date, effective_on, source_url)
        VALUES ('federal_register', ?, ?, ?, ?, '2026-06-26', ?, ?)
        ON CONFLICT(source, external_id) DO NOTHING
        """,
        (external_id, doc_type, title, external_id, effective_on, source_url),
    )
    conn.commit()
    row = conn.execute(
        "SELECT id FROM executive_actions WHERE external_id=?", (external_id,)
    ).fetchone()
    return row["id"] if row else -1
 
 
def _seed_bill(
    conn: sqlite3.Connection,
    external_id: str = "119-hr-1",
    title: str = "Test Act",
    source_url: str = "https://www.congress.gov/bill/119th-congress/house-bill/1",
) -> int:
    conn.execute(
        """
        INSERT INTO bills
            (source, external_id, congress, number, title,
             status, updated_at, source_url)
        VALUES ('govinfo_billstatus', ?, 119, '1', ?, 'enacted', '2025-01-01', ?)
        ON CONFLICT(source, external_id) DO NOTHING
        """,
        (external_id, title, source_url),
    )
    conn.commit()
    row = conn.execute(
        "SELECT id FROM bills WHERE external_id=?", (external_id,)
    ).fetchone()
    return row["id"] if row else -1
 
 
def _seed_action(
    conn: sqlite3.Connection,
    bill_id: int,
    action_type: str,
    action_date: str,
    text: str = "",
    source_url: str = "https://www.congress.gov/bill/119th-congress/house-bill/1",
) -> int:
    conn.execute(
        """
        INSERT INTO actions
            (bill_id, source, action_type, action_date, text, source_url)
        VALUES (?, 'govinfo_billstatus', ?, ?, ?, ?)
        """,
        (bill_id, action_type, action_date, text, source_url),
    )
    conn.commit()
    return conn.execute("SELECT last_insert_rowid()").fetchone()[0]
 
 
def _seed_vote(
    conn: sqlite3.Connection,
    date: str = "2025-03-15",
    result: str = "Passed",
    source_url: str = "https://www.senate.gov/legislative/LIS/roll_call_votes/vote1191/vote_119_1_00001.xml",
) -> int:
    conn.execute(
        """
        INSERT INTO votes (chamber, date, yea, nay, present, result, source_url)
        VALUES ('senate', ?, 60, 40, 0, ?, ?)
        """,
        (date, result, source_url),
    )
    conn.commit()
    return conn.execute("SELECT last_insert_rowid()").fetchone()[0]
 
 
def _seed_apportionment(
    conn: sqlite3.Connection,
    external_id: str = "test-apportionment-1",
    agency: str = "Department of Defense",
    date: str = "2025-10-01",
    source_url: str = "https://apportionment-public.max.gov/test.json",
) -> int:
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS apportionments (
            id          INTEGER PRIMARY KEY AUTOINCREMENT,
            source      TEXT NOT NULL DEFAULT 'openomb',
            external_id TEXT NOT NULL,
            agency      TEXT,
            account     TEXT,
            appropriated REAL,
            apportioned  REAL,
            status      TEXT,
            date        TEXT,
            source_url  TEXT NOT NULL,
            UNIQUE(source, external_id)
        )
        """
    )
    conn.execute(
        """
        INSERT INTO apportionments
            (source, external_id, agency, account, status, date, source_url)
        VALUES ('openomb', ?, ?, 'Operations', 'released', ?, ?)
        ON CONFLICT(source, external_id) DO NOTHING
        """,
        (external_id, agency, date, source_url),
    )
    conn.commit()
    row = conn.execute(
        "SELECT id FROM apportionments WHERE external_id=?", (external_id,)
    ).fetchone()
    return row["id"] if row else -1
 
 
 
class TestNormalizeDate:
    def test_iso_passthrough(self):
        assert _normalize_date("2026-08-01") == "2026-08-01"
 
    def test_full_month_name(self):
        assert _normalize_date("August 5, 2026") == "2026-08-05"
 
    def test_abbreviated_month_name(self):
        assert _normalize_date("Jan 1, 2025") == "2025-01-01"
 
    def test_leading_trailing_space(self):
        assert _normalize_date("  2026-01-15  ") == "2026-01-15"
 
    def test_unparseable_returns_none(self):
        assert _normalize_date("sometime next year") is None
        assert _normalize_date("") is None
        assert _normalize_date("180 days after enactment") is None
 
 
 
class TestParseEffectiveDates:
    def test_shall_take_effect_on(self):
        text = "This section shall take effect on January 1, 2026."
        dates = parse_effective_dates(text)
        assert "2026-01-01" in dates
 
    def test_takes_effect_on(self):
        text = "The rule takes effect on August 5, 2026."
        dates = parse_effective_dates(text)
        assert "2026-08-05" in dates
 
    def test_effective_on(self):
        text = "Effective on September 30, 2025, the following applies."
        dates = parse_effective_dates(text)
        assert "2025-09-30" in dates
 
    def test_effective_date_is(self):
        text = "The effective date is March 15, 2026."
        dates = parse_effective_dates(text)
        assert "2026-03-15" in dates
 
    def test_effective_date_shall_be(self):
        text = "The effective date shall be July 4, 2026."
        dates = parse_effective_dates(text)
        assert "2026-07-04" in dates
 
    def test_this_act_takes_effect(self):
        text = "This Act shall take effect on December 31, 2025."
        dates = parse_effective_dates(text)
        assert "2025-12-31" in dates
 
    def test_iso_date_format(self):
        text = "Effective on 2026-06-01, new requirements apply."
        dates = parse_effective_dates(text)
        assert "2026-06-01" in dates
 
    def test_no_dates_returns_empty(self):
        text = "The rule is effective 180 days after enactment."
        dates = parse_effective_dates(text)
        assert dates == []
 
    def test_deduplication(self):
        text = (
            "Effective on January 1, 2026. "
            "This Act takes effect on January 1, 2026."
        )
        dates = parse_effective_dates(text)
        assert dates.count("2026-01-01") == 1
 
    def test_multiple_distinct_dates(self):
        text = (
            "Section 1 takes effect on March 1, 2026. "
            "Section 2 shall take effect on June 1, 2026."
        )
        dates = parse_effective_dates(text)
        assert "2026-03-01" in dates
        assert "2026-06-01" in dates
        assert len(dates) == 2
 
    def test_relative_clause_not_extracted(self):
        text = "This provision takes effect 90 days after enactment."
        dates = parse_effective_dates(text)
        assert dates == []
 
    def test_empty_string(self):
        assert parse_effective_dates("") == []
 
    def test_patterns_compile(self):
        """Ensure all regex patterns are valid and pre-compiled."""
        for p in EFFECTIVE_DATE_PATTERNS:
            assert hasattr(p, "finditer"), f"Pattern {p!r} is not compiled"
 
 
 
class TestEnsureSchema:
    def test_table_created(self):
        conn, path = _make_db()
        try:
            tables = {
                r[0]
                for r in conn.execute(
                    "SELECT name FROM sqlite_master WHERE type='table'"
                )
            }
            assert "calendar_events" in tables
        finally:
            conn.close()
            os.unlink(path)
 
    def test_date_index_created(self):
        conn, path = _make_db()
        try:
            indexes = {
                r[0]
                for r in conn.execute(
                    "SELECT name FROM sqlite_master WHERE type='index'"
                )
            }
            assert "idx_calendar_events_date" in indexes
        finally:
            conn.close()
            os.unlink(path)
 
    def test_kind_index_created(self):
        conn, path = _make_db()
        try:
            indexes = {
                r[0]
                for r in conn.execute(
                    "SELECT name FROM sqlite_master WHERE type='index'"
                )
            }
            assert "idx_calendar_events_kind" in indexes
        finally:
            conn.close()
            os.unlink(path)
 
    def test_effective_on_column_added_to_executive_actions(self):
        conn, path = _make_db()
        conn.execute(
            """
            CREATE TABLE IF NOT EXISTS executive_actions (
                id          INTEGER PRIMARY KEY AUTOINCREMENT,
                source      TEXT NOT NULL DEFAULT 'federal_register',
                external_id TEXT NOT NULL,
                type        TEXT NOT NULL,
                title       TEXT,
                agency      TEXT,
                fr_doc_number TEXT,
                date        TEXT,
                source_url  TEXT NOT NULL,
                UNIQUE(source, external_id)
            )
            """
        )
        conn.commit()
        ensure_schema(conn)
        cols = {
            r[1]
            for r in conn.execute("PRAGMA table_info(executive_actions)")
        }
        assert "effective_on" in cols
        conn.close()
        os.unlink(path)
 
    def test_ensure_schema_idempotent(self):
        """Calling ensure_schema twice must not raise."""
        conn, path = _make_db()
        try:
            ensure_schema(conn)
        finally:
            conn.close()
            os.unlink(path)
 
    def test_kind_check_constraint(self):
        """Invalid kind values must be rejected by the CHECK constraint."""
        conn, path = _make_db()
        try:
            with pytest.raises(sqlite3.IntegrityError):
                conn.execute(
                    """
                    INSERT INTO calendar_events
                        (date, kind, ref_type, source_url)
                    VALUES ('2026-01-01', 'INVALID_KIND', 'bill',
                            'https://www.congress.gov/')
                    """
                )
                conn.commit()
        finally:
            conn.close()
            os.unlink(path)
 
 
 
class TestAggregateEffective:
    def test_basic(self):
        conn, path = _make_db()
        try:
            _seed_executive_action(conn, effective_on="2026-08-05")
            count = _aggregate_effective(conn)
            conn.commit()
            assert count == 1
            row = conn.execute(
                "SELECT * FROM calendar_events WHERE kind='effective'"
            ).fetchone()
            assert row is not None
            assert row["date"] == "2026-08-05"
            assert row["ref_type"] == "executive_action"
            assert row["confidence"] == 1.0
        finally:
            conn.close()
            os.unlink(path)
 
    def test_null_effective_on_excluded(self):
        conn, path = _make_db()
        try:
            _seed_executive_action(conn, effective_on=None)
            count = _aggregate_effective(conn)
            conn.commit()
            assert count == 0
        finally:
            conn.close()
            os.unlink(path)
 
    def test_notice_type_excluded(self):
        conn, path = _make_db()
        try:
            _seed_executive_action(
                conn, external_id="2026-99998", doc_type="notice", effective_on="2026-09-01"
            )
            count = _aggregate_effective(conn)
            conn.commit()
            assert count == 0
        finally:
            conn.close()
            os.unlink(path)
 
    def test_source_url_present(self):
        conn, path = _make_db()
        try:
            _seed_executive_action(conn, effective_on="2026-10-01")
            _aggregate_effective(conn)
            conn.commit()
            row = conn.execute(
                "SELECT source_url FROM calendar_events WHERE kind='effective'"
            ).fetchone()
            assert row["source_url"].startswith("https://www.federalregister.gov")
        finally:
            conn.close()
            os.unlink(path)
 
 
 
class TestAggregateEnacted:
    def test_basic(self):
        conn, path = _make_db()
        try:
            bill_id = _seed_bill(conn)
            _seed_action(
                conn, bill_id, "BecameLaw", "2025-07-04",
                text="Became Public Law No: 119-10.",
            )
            count = _aggregate_enacted(conn)
            conn.commit()
            assert count == 1
            row = conn.execute(
                "SELECT * FROM calendar_events WHERE kind='enacted'"
            ).fetchone()
            assert row["date"] == "2025-07-04"
            assert row["ref_type"] == "bill"
        finally:
            conn.close()
            os.unlink(path)
 
    def test_non_became_law_action_excluded(self):
        conn, path = _make_db()
        try:
            bill_id = _seed_bill(conn)
            _seed_action(conn, bill_id, "IntroReferral", "2025-01-10")
            count = _aggregate_enacted(conn)
            conn.commit()
            assert count == 0
        finally:
            conn.close()
            os.unlink(path)
 
 
 
class TestAggregateVotes:
    def test_basic(self):
        conn, path = _make_db()
        try:
            _seed_vote(conn, date="2025-05-20", result="Passed")
            count = _aggregate_votes(conn)
            conn.commit()
            assert count == 1
            row = conn.execute(
                "SELECT * FROM calendar_events WHERE kind='vote'"
            ).fetchone()
            assert row["date"] == "2025-05-20"
            assert row["ref_type"] == "vote"
        finally:
            conn.close()
            os.unlink(path)
 
    def test_multiple_votes(self):
        conn, path = _make_db()
        try:
            for day in ["2025-01-09", "2025-01-13", "2025-01-15"]:
                _seed_vote(conn, date=day)
            count = _aggregate_votes(conn)
            conn.commit()
            assert count == 3
        finally:
            conn.close()
            os.unlink(path)
 
 
 
class TestAggregateBillActions:
    def test_introduced(self):
        conn, path = _make_db()
        try:
            bill_id = _seed_bill(conn)
            _seed_action(conn, bill_id, "IntroReferral", "2025-01-03")
            _aggregate_bill_actions(conn)
            conn.commit()
            row = conn.execute(
                "SELECT * FROM calendar_events WHERE kind='introduced'"
            ).fetchone()
            assert row is not None
            assert row["date"] == "2025-01-03"
        finally:
            conn.close()
            os.unlink(path)
 
    def test_committee(self):
        conn, path = _make_db()
        try:
            bill_id = _seed_bill(conn)
            _seed_action(conn, bill_id, "Committee", "2025-02-14")
            _aggregate_bill_actions(conn)
            conn.commit()
            row = conn.execute(
                "SELECT * FROM calendar_events WHERE kind='committee'"
            ).fetchone()
            assert row is not None
            assert row["date"] == "2025-02-14"
        finally:
            conn.close()
            os.unlink(path)
 
    def test_floor(self):
        conn, path = _make_db()
        try:
            bill_id = _seed_bill(conn)
            _seed_action(conn, bill_id, "Floor", "2025-03-01")
            _aggregate_bill_actions(conn)
            conn.commit()
            row = conn.execute(
                "SELECT * FROM calendar_events WHERE kind='floor'"
            ).fetchone()
            assert row is not None
            assert row["date"] == "2025-03-01"
        finally:
            conn.close()
            os.unlink(path)
 
    def test_all_three_kinds_populated(self):
        conn, path = _make_db()
        try:
            bill_id = _seed_bill(conn)
            _seed_action(conn, bill_id, "IntroReferral", "2025-01-03")
            _seed_action(conn, bill_id, "Committee",     "2025-01-20")
            _seed_action(conn, bill_id, "Floor",         "2025-02-10")
            _aggregate_bill_actions(conn)
            conn.commit()
            kinds = {
                r[0]
                for r in conn.execute(
                    "SELECT DISTINCT kind FROM calendar_events"
                )
            }
            assert {"introduced", "committee", "floor"} <= kinds
        finally:
            conn.close()
            os.unlink(path)
 
 
 
class TestAggregateApportionments:
    def test_basic(self):
        conn, path = _make_db()
        try:
            _seed_apportionment(conn, date="2025-10-01")
            count = _aggregate_apportionments(conn)
            conn.commit()
            assert count == 1
            row = conn.execute(
                "SELECT * FROM calendar_events WHERE kind='apportionment'"
            ).fetchone()
            assert row is not None
            assert row["date"] == "2025-10-01"
            assert row["ref_type"] == "apportionment"
        finally:
            conn.close()
            os.unlink(path)
 
    def test_no_table_returns_zero(self):
        """If apportionments table does not exist, return 0 gracefully."""
        conn, path = _make_db()
        try:
            count = _aggregate_apportionments(conn)
            assert count == 0
        finally:
            conn.close()
            os.unlink(path)
 
 
 
class TestAggregateEffectiveParsed:
    def test_extracts_date_from_action_text(self):
        conn, path = _make_db()
        try:
            bill_id = _seed_bill(conn)
            _seed_action(
                conn, bill_id, "BecameLaw", "2025-07-04",
                text=(
                    "Became Public Law No: 119-10. "
                    "This Act shall take effect on January 1, 2026."
                ),
            )
            count = _aggregate_effective_parsed(conn)
            conn.commit()
            assert count >= 1
            row = conn.execute(
                "SELECT * FROM calendar_events WHERE kind='effective_parsed'"
            ).fetchone()
            assert row is not None
            assert row["date"] == "2026-01-01"
            assert row["confidence"] == 0.5
        finally:
            conn.close()
            os.unlink(path)
 
    def test_no_dates_in_text_returns_zero(self):
        conn, path = _make_db()
        try:
            bill_id = _seed_bill(conn)
            _seed_action(
                conn, bill_id, "BecameLaw", "2025-07-04",
                text="Became Public Law No: 119-10.",
            )
            count = _aggregate_effective_parsed(conn)
            conn.commit()
            assert count == 0
        finally:
            conn.close()
            os.unlink(path)
 
 
 
class TestRunIdempotency:
    def test_double_run_no_duplicates(self):
        """Running run() twice must not duplicate rows."""
        conn, path = _make_db()
        conn.close()
        try:
            bill_id_conn = get_connection(path)
            bill_id_conn.row_factory = sqlite3.Row
            init_schema(bill_id_conn)
            ensure_schema(bill_id_conn)
 
            bill_id = _seed_bill(bill_id_conn)
            _seed_action(bill_id_conn, bill_id, "BecameLaw", "2025-07-04",
                         text="Became Public Law No: 119-10.")
            _seed_action(bill_id_conn, bill_id, "IntroReferral", "2025-01-03")
            _seed_vote(bill_id_conn, date="2025-05-20")
 
            _seed_executive_action(
                bill_id_conn, effective_on="2026-08-01"
            )
            bill_id_conn.commit()
            bill_id_conn.close()
 
            counts1 = run(db_path=path)
 
            counts2 = run(db_path=path)
 
            for kind in counts1:
                assert counts1[kind] == counts2[kind], (
                    f"kind={kind!r}: first run={counts1[kind]}, "
                    f"second run={counts2[kind]}"
                )
        finally:
            if os.path.exists(path):
                os.unlink(path)
 
    def test_run_returns_counts_dict(self):
        """run() must return a dict with at least the core kinds."""
        conn, path = _make_db()
        conn.close()
        try:
            counts = run(db_path=path)
            assert isinstance(counts, dict)
            for kind in (
                "effective", "enacted", "vote",
                "introduced", "committee", "floor",
                "apportionment", "effective_parsed",
            ):
                assert kind in counts, f"Missing kind {kind!r} in counts"
        finally:
            if os.path.exists(path):
                os.unlink(path)

1	"""
2	Tests for ingest/calendar.py
3
4	Coverage:
5	• parse_effective_dates() - all regex patterns, mixed text, no false positives
6	• _normalize_date() - ISO passthrough, month-name parsing, bad input
7	• ensure_schema() - table + index creation, effective_on column
8	• _aggregate_effective() - pulls FR rules with effective_on from DB
9	• _aggregate_enacted() - pulls BecameLaw actions from DB
10	• _aggregate_votes() - pulls from votes table
11	• _aggregate_bill_actions() - introduced / committee / floor
12	• _aggregate_apportionments() - pulls from apportionments table
13	• _aggregate_effective_parsed() - regex over action text
14	• run() idempotency - double-run does not duplicate rows
15	"""
16
17	import os
18	import sqlite3
19	import tempfile
20
21	import pytest
22
23	from ingest.db import get_connection, init_schema
24	from ingest.calendar import (
25	EFFECTIVE_DATE_PATTERNS,
26	_normalize_date,
27	_aggregate_effective,
28	_aggregate_enacted,
29	_aggregate_votes,
30	_aggregate_bill_actions,
31	_aggregate_apportionments,
32	_aggregate_effective_parsed,
33	ensure_schema,
34	parse_effective_dates,
35	run,
36	)
37
38
39
40	def _make_db() -> tuple[sqlite3.Connection, str]:
41	"""Return (conn, path) for a fresh temp DB with full schema."""
42	path = tempfile.mktemp(suffix=".db")
43	conn = get_connection(path)
44	init_schema(conn)
45	ensure_schema(conn)
46	return conn, path
47
48
49	def _seed_executive_action(
50	conn: sqlite3.Connection,
51	external_id: str = "2026-99999",
52	doc_type: str = "final_rule",
53	title: str = "Test Rule",
54	effective_on: str \| None = "2026-08-01",
55	source_url: str = "https://www.federalregister.gov/documents/2026/06/26/2026-99999/test-rule",
56	) -> int:
57	"""Insert one executive_actions row; return its id."""
58	conn.execute(
59	"""
60	INSERT INTO executive_actions
61	(source, external_id, type, title, fr_doc_number,
62	date, effective_on, source_url)
63	VALUES ('federal_register', ?, ?, ?, ?, '2026-06-26', ?, ?)
64	ON CONFLICT(source, external_id) DO NOTHING
65	""",
66	(external_id, doc_type, title, external_id, effective_on, source_url),
67	)
68	conn.commit()
69	row = conn.execute(
70	"SELECT id FROM executive_actions WHERE external_id=?", (external_id,)
71	).fetchone()
72	return row["id"] if row else -1
73
74
75	def _seed_bill(
76	conn: sqlite3.Connection,
77	external_id: str = "119-hr-1",
78	title: str = "Test Act",
79	source_url: str = "https://www.congress.gov/bill/119th-congress/house-bill/1",
80	) -> int:
81	conn.execute(
82	"""
83	INSERT INTO bills
84	(source, external_id, congress, number, title,
85	status, updated_at, source_url)
86	VALUES ('govinfo_billstatus', ?, 119, '1', ?, 'enacted', '2025-01-01', ?)
87	ON CONFLICT(source, external_id) DO NOTHING
88	""",
89	(external_id, title, source_url),
90	)
91	conn.commit()
92	row = conn.execute(
93	"SELECT id FROM bills WHERE external_id=?", (external_id,)
94	).fetchone()
95	return row["id"] if row else -1
96
97
98	def _seed_action(
99	conn: sqlite3.Connection,
100	bill_id: int,
101	action_type: str,
102	action_date: str,
103	text: str = "",
104	source_url: str = "https://www.congress.gov/bill/119th-congress/house-bill/1",
105	) -> int:
106	conn.execute(
107	"""
108	INSERT INTO actions
109	(bill_id, source, action_type, action_date, text, source_url)
110	VALUES (?, 'govinfo_billstatus', ?, ?, ?, ?)
111	""",
112	(bill_id, action_type, action_date, text, source_url),
113	)
114	conn.commit()
115	return conn.execute("SELECT last_insert_rowid()").fetchone()[0]
116
117
118	def _seed_vote(
119	conn: sqlite3.Connection,
120	date: str = "2025-03-15",
121	result: str = "Passed",
122	source_url: str = "https://www.senate.gov/legislative/LIS/roll_call_votes/vote1191/vote_119_1_00001.xml",
123	) -> int:
124	conn.execute(
125	"""
126	INSERT INTO votes (chamber, date, yea, nay, present, result, source_url)
127	VALUES ('senate', ?, 60, 40, 0, ?, ?)
128	""",
129	(date, result, source_url),
130	)
131	conn.commit()
132	return conn.execute("SELECT last_insert_rowid()").fetchone()[0]
133
134
135	def _seed_apportionment(
136	conn: sqlite3.Connection,
137	external_id: str = "test-apportionment-1",
138	agency: str = "Department of Defense",
139	date: str = "2025-10-01",
140	source_url: str = "https://apportionment-public.max.gov/test.json",
141	) -> int:
142	conn.execute(
143	"""
144	CREATE TABLE IF NOT EXISTS apportionments (
145	id INTEGER PRIMARY KEY AUTOINCREMENT,
146	source TEXT NOT NULL DEFAULT 'openomb',
147	external_id TEXT NOT NULL,
148	agency TEXT,
149	account TEXT,
150	appropriated REAL,
151	apportioned REAL,
152	status TEXT,
153	date TEXT,
154	source_url TEXT NOT NULL,
155	UNIQUE(source, external_id)
156	)
157	"""
158	)
159	conn.execute(
160	"""
161	INSERT INTO apportionments
162	(source, external_id, agency, account, status, date, source_url)
163	VALUES ('openomb', ?, ?, 'Operations', 'released', ?, ?)
164	ON CONFLICT(source, external_id) DO NOTHING
165	""",
166	(external_id, agency, date, source_url),
167	)
168	conn.commit()
169	row = conn.execute(
170	"SELECT id FROM apportionments WHERE external_id=?", (external_id,)
171	).fetchone()
172	return row["id"] if row else -1
173
174
175
176	class TestNormalizeDate:
177	def test_iso_passthrough(self):
178	assert _normalize_date("2026-08-01") == "2026-08-01"
179
180	def test_full_month_name(self):
181	assert _normalize_date("August 5, 2026") == "2026-08-05"
182
183	def test_abbreviated_month_name(self):
184	assert _normalize_date("Jan 1, 2025") == "2025-01-01"
185
186	def test_leading_trailing_space(self):
187	assert _normalize_date(" 2026-01-15 ") == "2026-01-15"
188
189	def test_unparseable_returns_none(self):
190	assert _normalize_date("sometime next year") is None
191	assert _normalize_date("") is None
192	assert _normalize_date("180 days after enactment") is None
193
194
195
196	class TestParseEffectiveDates:
197	def test_shall_take_effect_on(self):
198	text = "This section shall take effect on January 1, 2026."
199	dates = parse_effective_dates(text)
200	assert "2026-01-01" in dates
201
202	def test_takes_effect_on(self):
203	text = "The rule takes effect on August 5, 2026."
204	dates = parse_effective_dates(text)
205	assert "2026-08-05" in dates
206
207	def test_effective_on(self):
208	text = "Effective on September 30, 2025, the following applies."
209	dates = parse_effective_dates(text)
210	assert "2025-09-30" in dates
211
212	def test_effective_date_is(self):
213	text = "The effective date is March 15, 2026."
214	dates = parse_effective_dates(text)
215	assert "2026-03-15" in dates
216
217	def test_effective_date_shall_be(self):
218	text = "The effective date shall be July 4, 2026."
219	dates = parse_effective_dates(text)
220	assert "2026-07-04" in dates
221
222	def test_this_act_takes_effect(self):
223	text = "This Act shall take effect on December 31, 2025."
224	dates = parse_effective_dates(text)
225	assert "2025-12-31" in dates
226
227	def test_iso_date_format(self):
228	text = "Effective on 2026-06-01, new requirements apply."
229	dates = parse_effective_dates(text)
230	assert "2026-06-01" in dates
231
232	def test_no_dates_returns_empty(self):
233	text = "The rule is effective 180 days after enactment."
234	dates = parse_effective_dates(text)
235	assert dates == []
236
237	def test_deduplication(self):
238	text = (
239	"Effective on January 1, 2026. "
240	"This Act takes effect on January 1, 2026."
241	)
242	dates = parse_effective_dates(text)
243	assert dates.count("2026-01-01") == 1
244
245	def test_multiple_distinct_dates(self):
246	text = (
247	"Section 1 takes effect on March 1, 2026. "
248	"Section 2 shall take effect on June 1, 2026."
249	)
250	dates = parse_effective_dates(text)
251	assert "2026-03-01" in dates
252	assert "2026-06-01" in dates
253	assert len(dates) == 2
254
255	def test_relative_clause_not_extracted(self):
256	text = "This provision takes effect 90 days after enactment."
257	dates = parse_effective_dates(text)
258	assert dates == []
259
260	def test_empty_string(self):
261	assert parse_effective_dates("") == []
262
263	def test_patterns_compile(self):
264	"""Ensure all regex patterns are valid and pre-compiled."""
265	for p in EFFECTIVE_DATE_PATTERNS:
266	assert hasattr(p, "finditer"), f"Pattern {p!r} is not compiled"
267
268
269
270	class TestEnsureSchema:
271	def test_table_created(self):
272	conn, path = _make_db()
273	try:
274	tables = {
275	r[0]
276	for r in conn.execute(
277	"SELECT name FROM sqlite_master WHERE type='table'"
278	)
279	}
280	assert "calendar_events" in tables
281	finally:
282	conn.close()
283	os.unlink(path)
284
285	def test_date_index_created(self):
286	conn, path = _make_db()
287	try:
288	indexes = {
289	r[0]
290	for r in conn.execute(
291	"SELECT name FROM sqlite_master WHERE type='index'"
292	)
293	}
294	assert "idx_calendar_events_date" in indexes
295	finally:
296	conn.close()
297	os.unlink(path)
298
299	def test_kind_index_created(self):
300	conn, path = _make_db()
301	try:
302	indexes = {
303	r[0]
304	for r in conn.execute(
305	"SELECT name FROM sqlite_master WHERE type='index'"
306	)
307	}
308	assert "idx_calendar_events_kind" in indexes
309	finally:
310	conn.close()
311	os.unlink(path)
312
313	def test_effective_on_column_added_to_executive_actions(self):
314	conn, path = _make_db()
315	conn.execute(
316	"""
317	CREATE TABLE IF NOT EXISTS executive_actions (
318	id INTEGER PRIMARY KEY AUTOINCREMENT,
319	source TEXT NOT NULL DEFAULT 'federal_register',
320	external_id TEXT NOT NULL,
321	type TEXT NOT NULL,
322	title TEXT,
323	agency TEXT,
324	fr_doc_number TEXT,
325	date TEXT,
326	source_url TEXT NOT NULL,
327	UNIQUE(source, external_id)
328	)
329	"""
330	)
331	conn.commit()
332	ensure_schema(conn)
333	cols = {
334	r[1]
335	for r in conn.execute("PRAGMA table_info(executive_actions)")
336	}
337	assert "effective_on" in cols
338	conn.close()
339	os.unlink(path)
340
341	def test_ensure_schema_idempotent(self):
342	"""Calling ensure_schema twice must not raise."""
343	conn, path = _make_db()
344	try:
345	ensure_schema(conn)
346	finally:
347	conn.close()
348	os.unlink(path)
349
350	def test_kind_check_constraint(self):
351	"""Invalid kind values must be rejected by the CHECK constraint."""
352	conn, path = _make_db()
353	try:
354	with pytest.raises(sqlite3.IntegrityError):
355	conn.execute(
356	"""
357	INSERT INTO calendar_events
358	(date, kind, ref_type, source_url)
359	VALUES ('2026-01-01', 'INVALID_KIND', 'bill',
360	'https://www.congress.gov/')
361	"""
362	)
363	conn.commit()
364	finally:
365	conn.close()
366	os.unlink(path)
367
368
369
370	class TestAggregateEffective:
371	def test_basic(self):
372	conn, path = _make_db()
373	try:
374	_seed_executive_action(conn, effective_on="2026-08-05")
375	count = _aggregate_effective(conn)
376	conn.commit()
377	assert count == 1
378	row = conn.execute(
379	"SELECT * FROM calendar_events WHERE kind='effective'"
380	).fetchone()
381	assert row is not None
382	assert row["date"] == "2026-08-05"
383	assert row["ref_type"] == "executive_action"
384	assert row["confidence"] == 1.0
385	finally:
386	conn.close()
387	os.unlink(path)
388
389	def test_null_effective_on_excluded(self):
390	conn, path = _make_db()
391	try:
392	_seed_executive_action(conn, effective_on=None)
393	count = _aggregate_effective(conn)
394	conn.commit()
395	assert count == 0
396	finally:
397	conn.close()
398	os.unlink(path)
399
400	def test_notice_type_excluded(self):
401	conn, path = _make_db()
402	try:
403	_seed_executive_action(
404	conn, external_id="2026-99998", doc_type="notice", effective_on="2026-09-01"
405	)
406	count = _aggregate_effective(conn)
407	conn.commit()
408	assert count == 0
409	finally:
410	conn.close()
411	os.unlink(path)
412
413	def test_source_url_present(self):
414	conn, path = _make_db()
415	try:
416	_seed_executive_action(conn, effective_on="2026-10-01")
417	_aggregate_effective(conn)
418	conn.commit()
419	row = conn.execute(
420	"SELECT source_url FROM calendar_events WHERE kind='effective'"
421	).fetchone()
422	assert row["source_url"].startswith("https://www.federalregister.gov")
423	finally:
424	conn.close()
425	os.unlink(path)
426
427
428
429	class TestAggregateEnacted:
430	def test_basic(self):
431	conn, path = _make_db()
432	try:
433	bill_id = _seed_bill(conn)
434	_seed_action(
435	conn, bill_id, "BecameLaw", "2025-07-04",
436	text="Became Public Law No: 119-10.",
437	)
438	count = _aggregate_enacted(conn)
439	conn.commit()
440	assert count == 1
441	row = conn.execute(
442	"SELECT * FROM calendar_events WHERE kind='enacted'"
443	).fetchone()
444	assert row["date"] == "2025-07-04"
445	assert row["ref_type"] == "bill"
446	finally:
447	conn.close()
448	os.unlink(path)
449
450	def test_non_became_law_action_excluded(self):
451	conn, path = _make_db()
452	try:
453	bill_id = _seed_bill(conn)
454	_seed_action(conn, bill_id, "IntroReferral", "2025-01-10")
455	count = _aggregate_enacted(conn)
456	conn.commit()
457	assert count == 0
458	finally:
459	conn.close()
460	os.unlink(path)
461
462
463
464	class TestAggregateVotes:
465	def test_basic(self):
466	conn, path = _make_db()
467	try:
468	_seed_vote(conn, date="2025-05-20", result="Passed")
469	count = _aggregate_votes(conn)
470	conn.commit()
471	assert count == 1
472	row = conn.execute(
473	"SELECT * FROM calendar_events WHERE kind='vote'"
474	).fetchone()
475	assert row["date"] == "2025-05-20"
476	assert row["ref_type"] == "vote"
477	finally:
478	conn.close()
479	os.unlink(path)
480
481	def test_multiple_votes(self):
482	conn, path = _make_db()
483	try:
484	for day in ["2025-01-09", "2025-01-13", "2025-01-15"]:
485	_seed_vote(conn, date=day)
486	count = _aggregate_votes(conn)
487	conn.commit()
488	assert count == 3
489	finally:
490	conn.close()
491	os.unlink(path)
492
493
494
495	class TestAggregateBillActions:
496	def test_introduced(self):
497	conn, path = _make_db()
498	try:
499	bill_id = _seed_bill(conn)
500	_seed_action(conn, bill_id, "IntroReferral", "2025-01-03")
501	_aggregate_bill_actions(conn)
502	conn.commit()
503	row = conn.execute(
504	"SELECT * FROM calendar_events WHERE kind='introduced'"
505	).fetchone()
506	assert row is not None
507	assert row["date"] == "2025-01-03"
508	finally:
509	conn.close()
510	os.unlink(path)
511
512	def test_committee(self):
513	conn, path = _make_db()
514	try:
515	bill_id = _seed_bill(conn)
516	_seed_action(conn, bill_id, "Committee", "2025-02-14")
517	_aggregate_bill_actions(conn)
518	conn.commit()
519	row = conn.execute(
520	"SELECT * FROM calendar_events WHERE kind='committee'"
521	).fetchone()
522	assert row is not None
523	assert row["date"] == "2025-02-14"
524	finally:
525	conn.close()
526	os.unlink(path)
527
528	def test_floor(self):
529	conn, path = _make_db()
530	try:
531	bill_id = _seed_bill(conn)
532	_seed_action(conn, bill_id, "Floor", "2025-03-01")
533	_aggregate_bill_actions(conn)
534	conn.commit()
535	row = conn.execute(
536	"SELECT * FROM calendar_events WHERE kind='floor'"
537	).fetchone()
538	assert row is not None
539	assert row["date"] == "2025-03-01"
540	finally:
541	conn.close()
542	os.unlink(path)
543
544	def test_all_three_kinds_populated(self):
545	conn, path = _make_db()
546	try:
547	bill_id = _seed_bill(conn)
548	_seed_action(conn, bill_id, "IntroReferral", "2025-01-03")
549	_seed_action(conn, bill_id, "Committee", "2025-01-20")
550	_seed_action(conn, bill_id, "Floor", "2025-02-10")
551	_aggregate_bill_actions(conn)
552	conn.commit()
553	kinds = {
554	r[0]
555	for r in conn.execute(
556	"SELECT DISTINCT kind FROM calendar_events"
557	)
558	}
559	assert {"introduced", "committee", "floor"} <= kinds
560	finally:
561	conn.close()
562	os.unlink(path)
563
564
565
566	class TestAggregateApportionments:
567	def test_basic(self):
568	conn, path = _make_db()
569	try:
570	_seed_apportionment(conn, date="2025-10-01")
571	count = _aggregate_apportionments(conn)
572	conn.commit()
573	assert count == 1
574	row = conn.execute(
575	"SELECT * FROM calendar_events WHERE kind='apportionment'"
576	).fetchone()
577	assert row is not None
578	assert row["date"] == "2025-10-01"
579	assert row["ref_type"] == "apportionment"
580	finally:
581	conn.close()
582	os.unlink(path)
583
584	def test_no_table_returns_zero(self):
585	"""If apportionments table does not exist, return 0 gracefully."""
586	conn, path = _make_db()
587	try:
588	count = _aggregate_apportionments(conn)
589	assert count == 0
590	finally:
591	conn.close()
592	os.unlink(path)
593
594
595
596	class TestAggregateEffectiveParsed:
597	def test_extracts_date_from_action_text(self):
598	conn, path = _make_db()
599	try:
600	bill_id = _seed_bill(conn)
601	_seed_action(
602	conn, bill_id, "BecameLaw", "2025-07-04",
603	text=(
604	"Became Public Law No: 119-10. "
605	"This Act shall take effect on January 1, 2026."
606	),
607	)
608	count = _aggregate_effective_parsed(conn)
609	conn.commit()
610	assert count >= 1
611	row = conn.execute(
612	"SELECT * FROM calendar_events WHERE kind='effective_parsed'"
613	).fetchone()
614	assert row is not None
615	assert row["date"] == "2026-01-01"
616	assert row["confidence"] == 0.5
617	finally:
618	conn.close()
619	os.unlink(path)
620
621	def test_no_dates_in_text_returns_zero(self):
622	conn, path = _make_db()
623	try:
624	bill_id = _seed_bill(conn)
625	_seed_action(
626	conn, bill_id, "BecameLaw", "2025-07-04",
627	text="Became Public Law No: 119-10.",
628	)
629	count = _aggregate_effective_parsed(conn)
630	conn.commit()
631	assert count == 0
632	finally:
633	conn.close()
634	os.unlink(path)
635
636
637
638	class TestRunIdempotency:
639	def test_double_run_no_duplicates(self):
640	"""Running run() twice must not duplicate rows."""
641	conn, path = _make_db()
642	conn.close()
643	try:
644	bill_id_conn = get_connection(path)
645	bill_id_conn.row_factory = sqlite3.Row
646	init_schema(bill_id_conn)
647	ensure_schema(bill_id_conn)
648
649	bill_id = _seed_bill(bill_id_conn)
650	_seed_action(bill_id_conn, bill_id, "BecameLaw", "2025-07-04",
651	text="Became Public Law No: 119-10.")
652	_seed_action(bill_id_conn, bill_id, "IntroReferral", "2025-01-03")
653	_seed_vote(bill_id_conn, date="2025-05-20")
654
655	_seed_executive_action(
656	bill_id_conn, effective_on="2026-08-01"
657	)
658	bill_id_conn.commit()
659	bill_id_conn.close()
660
661	counts1 = run(db_path=path)
662
663	counts2 = run(db_path=path)
664
665	for kind in counts1:
666	assert counts1[kind] == counts2[kind], (
667	f"kind={kind!r}: first run={counts1[kind]}, "
668	f"second run={counts2[kind]}"
669	)
670	finally:
671	if os.path.exists(path):
672	os.unlink(path)
673
674	def test_run_returns_counts_dict(self):
675	"""run() must return a dict with at least the core kinds."""
676	conn, path = _make_db()
677	conn.close()
678	try:
679	counts = run(db_path=path)
680	assert isinstance(counts, dict)
681	for kind in (
682	"effective", "enacted", "vote",
683	"introduced", "committee", "floor",
684	"apportionment", "effective_parsed",
685	):
686	assert kind in counts, f"Missing kind {kind!r} in counts"
687	finally:
688	if os.path.exists(path):
689	os.unlink(path)