"""

Tests for the CREC procedural-event detector (crec.py).

Covers:

  - _VOICE_VOTE regex fires on canonical floor phrases

  - _UC regex fires on unanimous-consent phrases

  - _CLOTURE regex fires on cloture phrases

  - _SUSPENSION, _MTR, _DISCHARGE patterns

  - _detect_events() yields correct kind/chamber/date

  - members_jsonb is always None (no individual record for voice votes)

  - source_url is always a .gov URL

  - _extract_text_from_xml() handles valid and malformed XML

  - _infer_chamber() maps filenames correctly

  - _context_snippet() extracts reasonable surrounding text

  - Idempotency: re-processing the same text does not double-insert

"""

import io

import re

import sqlite3

import tempfile

import xml.etree.ElementTree as ET

import pytest

from ingest.pollers.crec import (

    _CLOTURE,

    _DISCHARGE,

    _MTR,

    _SUSPENSION,

    _UC,

    _VOICE_VOTE,

    _context_snippet,

    _detect_events,

    _ensure_tables,

    _extract_text_from_xml,

    _infer_chamber,

    poll_date,

)

from ingest.db import get_connection, init_schema

from ingest.source_validator import is_allowlisted

@pytest.fixture()

def temp_db():

    """In-memory (temp-file) SQLite DB with schema + procedural_events table."""

    tmp = tempfile.mktemp(suffix=".db")

    conn = get_connection(tmp)

    init_schema(conn)

    _ensure_tables(conn)

    yield conn

    conn.close()

    import os; os.unlink(tmp)

class TestVoiceVotePattern:

    """The PRESIDING OFFICER uses specific floor phrases to declare voice-vote results."""

    def test_ayes_have_it_lowercase(self):

        assert _VOICE_VOTE.search("the ayes have it")

    def test_ayes_have_it_titlecase(self):

        assert _VOICE_VOTE.search("The Ayes Have It")

    def test_yeas_have_it(self):

        assert _VOICE_VOTE.search("the yeas have it")

    def test_noes_have_it(self):

        assert _VOICE_VOTE.search("the noes have it")

    def test_nays_have_it(self):

        assert _VOICE_VOTE.search("the nays have it")

    def test_double_declaration(self):

        assert _VOICE_VOTE.search("the ayes have it, the ayes have it")

    def test_embedded_in_sentence(self):

        text = (

            "The PRESIDING OFFICER. The question is on the amendment. "

            "The ayes have it. The amendment is agreed to."

        )

        assert _VOICE_VOTE.search(text)

    def test_no_false_positive_on_roll_call(self):

        text = "The yeas are 67, nays 33. The motion is agreed to."

        assert not _VOICE_VOTE.search(text)

    def test_aye_singular(self):

        assert _VOICE_VOTE.search("the aye has it")

    def test_partial_words_no_match(self):

        assert not _VOICE_VOTE.search("ayesha")

class TestUCPattern:

    def test_without_objection(self):

        assert _UC.search("Without objection, so ordered.")

    def test_so_ordered(self):

        assert _UC.search("So ordered.")

    def test_unanimous_consent(self):

        assert _UC.search("Mr. Speaker, I ask unanimous consent to revise and extend")

    def test_case_insensitive(self):

        assert _UC.search("WITHOUT OBJECTION")

        assert _UC.search("unanimous CONSENT")

    def test_embedded(self):

        text = "The Senator from Ohio asks unanimous consent that the Senate proceed."

        assert _UC.search(text)

    def test_no_match_on_unrelated(self):

        assert not _UC.search("The bill passed by a recorded vote of 52 to 48.")

class TestCloturePattern:

    def test_cloture_motion(self):

        assert _CLOTURE.search("A cloture motion was filed on the nomination.")

    def test_invoke_cloture(self):

        assert _CLOTURE.search("The Senate voted to invoke cloture.")

    def test_vote_on_cloture(self):

        assert _CLOTURE.search("A vote on cloture is scheduled for Monday.")

    def test_cloture_was_invoked(self):

        assert _CLOTURE.search("Cloture was invoked by a vote of 60-38.")

    def test_motion_to_invoke_cloture(self):

        assert _CLOTURE.search("A motion to invoke cloture on the bill was filed.")

    def test_no_false_positive(self):

        assert not _CLOTURE.search("The committee voted on the closure of the hearing room.")

class TestSuspensionPattern:

    def test_suspend_the_rules(self):

        assert _SUSPENSION.search("On motion to suspend the rules and pass the bill")

    def test_suspension_of_the_rules(self):

        assert _SUSPENSION.search("Suspension of the rules - H.R. 1234")

    def test_case_insensitive(self):

        assert _SUSPENSION.search("SUSPEND THE RULES")

    def test_no_false_positive_on_other_suspend(self):

        text = "The officer was suspended for misconduct."

        assert not _SUSPENSION.search(text)

class TestMTRPattern:

    def test_basic(self):

        assert _MTR.search("The gentleman offers a motion to recommit the bill.")

    def test_case_insensitive(self):

        assert _MTR.search("MOTION TO RECOMMIT")

    def test_no_false_positive(self):

        assert not _MTR.search("The motion to table failed.")

class TestDischargePattern:

    def test_discharge_petition(self):

        assert _DISCHARGE.search("A discharge petition was filed for H.R. 10.")

    def test_discharged_from_committee(self):

        assert _DISCHARGE.search("The bill was discharged from committee.")

    def test_discharge_calendar(self):

        assert _DISCHARGE.search("The measure was placed on the discharge calendar.")

    def test_no_false_positive(self):

        assert not _DISCHARGE.search("The soldier was honorably discharged.")

class TestExtractTextFromXml:

    def test_basic_xml(self):

        xml = b"<root><p>The ayes have it.</p><p>So ordered.</p></root>"

        text = _extract_text_from_xml(xml)

        assert "ayes have it" in text

        assert "So ordered" in text

    def test_nested_elements(self):

        xml = b"<a><b><c>deep text</c></b></a>"

        assert "deep text" in _extract_text_from_xml(xml)

    def test_html_tags_stripped(self):

        htm = b"<p><b>Mr. JONES.</b> Without objection, so ordered.</p>"

        text = _extract_text_from_xml(htm)

        assert "without objection" in text.lower()

        assert "<p>" not in text

        assert "<b>" not in text

    def test_html_entities_unescaped(self):

        htm = b"<p>The ayes have it.&nbsp; The amendment is agreed&nbsp;to.</p>"

        text = _extract_text_from_xml(htm)

        assert "ayes have it" in text

    def test_empty_bytes(self):

        assert _extract_text_from_xml(b"") == ""

    def test_malformed_html_decoded(self):

        raw = b"<broken>The noes have it.<br>"

        text = _extract_text_from_xml(raw)

        assert "noes have it" in text

class TestInferChamber:

    def test_house_filename(self):

        assert _infer_chamber("CREC-2026-06-24-pt1-PgH1.xml") == "House"

    def test_senate_filename(self):

        assert _infer_chamber("CREC-2026-06-24-pt1-PgS100.xml") == "Senate"

    def test_digest_filename(self):

        assert _infer_chamber("CREC-2026-06-24-pt1-PgD1.xml") == "Both"

    def test_unknown_filename(self):

        assert _infer_chamber("CREC-2026-06-24-extensions.xml") == "Unknown"

class TestContextSnippet:

    def test_returns_text_around_match(self):

        text = "The Presiding Officer called for order. The ayes have it. The amendment is agreed to."

        m = _VOICE_VOTE.search(text)

        snippet = _context_snippet(text, m, window=40)

        assert "ayes have it" in snippet

    def test_does_not_exceed_text_bounds(self):

        text = "The ayes have it."

        m = _VOICE_VOTE.search(text)

        snippet = _context_snippet(text, m, window=200)

        assert snippet in text or set(snippet) <= set(text + " ")

    def test_strips_newlines(self):

        text = "Motion to recommit\nwas offered."

        m = _MTR.search(text)

        snippet = _context_snippet(text, m, window=60)

        assert "\n" not in snippet

class TestDetectEvents:

    BASE_URL = "https://www.govinfo.gov/app/details/CREC-2026-06-24"

    def _events(self, text, chamber="Senate"):

        return list(

            _detect_events(

                text,

                chamber=chamber,

                date_iso="2026-06-24",

                source_url=self.BASE_URL,

                filename="test-section.xml",

            )

        )

    def test_voice_vote_detected(self):

        events = self._events("The PRESIDING OFFICER. The ayes have it.")

        assert any(e["kind"] == "voice_vote" for e in events)

    def test_uc_detected(self):

        events = self._events("Without objection, so ordered.")

        assert any(e["kind"] == "unanimous_consent" for e in events)

    def test_cloture_detected(self):

        events = self._events("Cloture was invoked on the nomination by 62-38.")

        assert any(e["kind"] == "cloture" for e in events)

    def test_suspension_detected(self):

        events = self._events("On motion to suspend the rules and pass H.R. 99.")

        assert any(e["kind"] == "suspension" for e in events)

    def test_mtr_detected(self):

        events = self._events("Mr. SMITH offered a motion to recommit with instructions.")

        assert any(e["kind"] == "mtr" for e in events)

    def test_discharge_detected(self):

        events = self._events("A discharge petition was filed for H.R. 10.")

        assert any(e["kind"] == "discharge" for e in events)

    def test_members_jsonb_is_none(self):

        """Voice votes have no individual member record  -  members_jsonb must be null."""

        events = self._events("The ayes have it. The resolution is agreed to.")

        vv = [e for e in events if e["kind"] == "voice_vote"]

        assert len(vv) >= 1

        for e in vv:

            assert e["members_jsonb"] is None, (

                "Voice-vote events must have members_jsonb=None "

                "(no individual record exists - that absence is itself information)"

            )

    def test_source_url_is_gov(self):

        events = self._events("The ayes have it.")

        for e in events:

            assert is_allowlisted(e["source_url"]), (

                f"source_url {e['source_url']!r} is not a .gov URL"

            )

    def test_chamber_propagated(self):

        events = self._events("The ayes have it.", chamber="House")

        vv = [e for e in events if e["kind"] == "voice_vote"]

        assert all(e["chamber"] == "House" for e in vv)

    def test_date_propagated(self):

        events = self._events("The noes have it.")

        for e in events:

            assert e["date"] == "2026-06-24"

    def test_external_id_unique_per_match(self):

        text = "The ayes have it. The ayes have it. The noes have it."

        events = self._events(text)

        vv_ids = [e["external_id"] for e in events if e["kind"] == "voice_vote"]

        assert len(vv_ids) == len(set(vv_ids)), "external_ids must be unique"

    def test_empty_text_yields_no_events(self):

        assert self._events("") == []

    def test_unrelated_text_yields_no_events(self):

        text = "Mr. SMITH. I yield back the balance of my time."

        assert self._events(text) == []

class TestSourceUrlIntegrity:

    """All source_urls produced by the CREC poller must be .gov allowlisted."""

    def test_govinfo_page_url_is_gov(self):

        from ingest.pollers.crec import _govinfo_page_url

        from datetime import datetime

        url = _govinfo_page_url(datetime(2026, 6, 24))

        assert "govinfo.gov" in url

        assert is_allowlisted(url)

    def test_govinfo_zip_url_is_gov(self):

        from ingest.pollers.crec import _govinfo_url

        from datetime import datetime

        url = _govinfo_url(datetime(2026, 6, 24))

        assert "govinfo.gov" in url

        assert is_allowlisted(url)

class TestIdempotency:

    """Re-processing the same events must not create duplicate rows."""

    SOURCE_URL = "https://www.govinfo.gov/app/details/CREC-2026-06-24"

    def test_no_duplicate_rows_on_second_pass(self, temp_db):

        """

        Call _detect_events twice with the same text and simulate two DB

        insert passes; the second pass must yield 0 new stored rows.

        """

        text = "The ayes have it. Without objection, so ordered."

        events = list(

            _detect_events(text, "Senate", "2026-06-24", self.SOURCE_URL, "test.xml")

        )

        assert len(events) > 0

        from ingest.change_detection import is_new_event, mark_seen

        for evt in events:

            mark_seen(temp_db, "crec", evt["external_id"], evt["update_date"])

        new = [

            e for e in events

            if is_new_event(temp_db, "crec", e["external_id"], e["update_date"])

        ]

        assert new == [], f"Second pass should be empty; got {new}"