HERMES-GIELIE/tests/test_issue4812_session_sse_contract_rfc.py

"""Docs/source contract tests for issue #4812 session SSE contract RFC.

These tests assert the invariants of docs/rfcs/session-sse-contract-v1.md
without requiring a running server or endpoint implementation.
"""

import pathlib

REPO = pathlib.Path(__file__).parent.parent
RFC = REPO / "docs" / "rfcs" / "session-sse-contract-v1.md"
RFC_README = REPO / "docs" / "rfcs" / "README.md"
CONTRACTS = REPO / "docs" / "CONTRACTS.md"


def _rfc() -> str:
    return RFC.read_text(encoding="utf-8")


def _readme() -> str:
    return RFC_README.read_text(encoding="utf-8")


def _contracts() -> str:
    return CONTRACTS.read_text(encoding="utf-8")


class TestRFCExists:
    def test_rfc_file_exists(self):
        assert RFC.exists(), f"RFC file not found: {RFC}"

    def test_rfc_has_status_proposed(self):
        assert "Status:** Proposed" in _rfc(), "RFC must have 'Status: Proposed' header"

    def test_rfc_has_author(self):
        assert "Author:** @" in _rfc(), "RFC must have 'Author: @...' header"

    def test_rfc_has_created_date(self):
        assert "Created:** " in _rfc(), "RFC must have 'Created: YYYY-MM-DD' header"

    def test_rfc_has_tracking_issue(self):
        assert "Tracking:** #4812" in _rfc(), "RFC must have 'Tracking: #4812' header"

    def test_rfc_refs_not_closes(self):
        text = _rfc()
        assert "Refs #4812" in text, "RFC must use 'Refs #4812', not 'Closes #4812'"
        assert "Closes #4812" not in text, "RFC must not use 'Closes #4812'"


class TestRFCIndexed:
    def test_readme_indexes_rfc(self):
        assert "session-sse-contract-v1.md" in _readme(), (
            "docs/rfcs/README.md must list session-sse-contract-v1.md"
        )

    def test_readme_indexes_rfc_with_issue(self):
        assert "#4812" in _readme(), (
            "docs/rfcs/README.md index entry must reference #4812"
        )

    def test_contracts_indexes_rfc(self):
        assert "session-sse-contract-v1.md" in _contracts(), (
            "docs/CONTRACTS.md must reference session-sse-contract-v1.md"
        )


class TestEndpointDistinction:
    def test_rfc_names_proposed_per_session_endpoint(self):
        assert "/api/sessions/{session_id}/events" in _rfc(), (
            "RFC must propose GET /api/sessions/{session_id}/events"
        )

    def test_rfc_distinguishes_global_sessions_events(self):
        text = _rfc()
        assert "/api/sessions/events" in text, (
            "RFC must mention existing GET /api/sessions/events"
        )
        assert "/api/sessions/{session_id}/events" in text, (
            "RFC must distinguish per-session endpoint from global endpoint"
        )

    def test_rfc_states_global_endpoint_is_different(self):
        text = _rfc()
        assert "path-distinct" in text or "different endpoint" in text, (
            "RFC must explicitly state the two endpoints are distinct"
        )

    def test_rfc_cites_current_global_endpoint_source(self):
        text = _rfc()
        assert "api/routes.py:12320-12321" in text, (
            "RFC must cite the current global sessions/events route match"
        )
        assert "api/routes.py:16142" in text, (
            "RFC must cite the current _handle_session_events_stream definition"
        )
        assert "api/routes.py:12296-12297" not in text
        assert "api/routes.py:16118" not in text

    def test_contracts_distinguishes_both_endpoints(self):
        text = _contracts()
        assert "/api/sessions/{session_id}/events" in text, (
            "CONTRACTS.md must mention proposed per-session endpoint"
        )
        assert "/api/sessions/events" in text, (
            "CONTRACTS.md must mention existing global sessions/events endpoint"
        )


class TestSequenceAndReplaySemantics:
    def test_rfc_includes_last_event_id(self):
        assert "Last-Event-ID" in _rfc(), "RFC must include Last-Event-ID"

    def test_rfc_includes_event_id(self):
        assert "event_id" in _rfc(), "RFC must include event_id"

    def test_rfc_includes_stream_id(self):
        assert "stream_id" in _rfc(), "RFC must include stream_id"

    def test_rfc_includes_seq(self):
        assert '"seq"' in _rfc() or "`seq`" in _rfc(), "RFC must include seq field"

    def test_rfc_includes_session_snapshot(self):
        assert "session_snapshot" in _rfc(), "RFC must include session_snapshot event"

    def test_rfc_states_seq_is_stream_scoped(self):
        text = _rfc()
        assert "monotonic within a stream" in text or "stream/run-scoped" in text, (
            "RFC must state seq is monotonic within a stream/run, not session-global"
        )

    def test_rfc_does_not_promise_session_global_counter(self):
        text = _rfc()
        assert "does not claim a pre-existing session-global sequence" in text or (
            "not a session-global counter" in text
        ), (
            "RFC must explicitly state Phase 1 does not claim a session-global counter"
        )

    def test_rfc_states_event_id_is_opaque(self):
        assert "opaque" in _rfc(), "RFC must state event_id is opaque to clients"

    def test_rfc_names_run_journal_as_replay_source(self):
        text = _rfc()
        assert "run journal" in text.lower(), (
            "RFC must name the run journal as the replay source"
        )

    def test_rfc_defines_session_snapshot_as_fallback(self):
        text = _rfc()
        assert "session_snapshot" in text
        assert "fallback" in text.lower() or "snapshot fallback" in text.lower(), (
            "RFC must define session_snapshot as the stale-cursor fallback"
        )

    def test_rfc_snapshot_is_not_exact_replay(self):
        text = _rfc()
        assert "not proof of exact missed-event replay" in text or (
            "recovery boundary" in text
        ), (
            "RFC must state snapshot is a recovery boundary, not exact missed-event replay"
        )


class TestHeartbeat:
    def test_rfc_references_heartbeat_constant(self):
        assert "_SSE_HEARTBEAT_INTERVAL_SECONDS" in _rfc(), (
            "RFC must reference _SSE_HEARTBEAT_INTERVAL_SECONDS rather than inventing a new constant"
        )

    def test_rfc_does_not_add_new_heartbeat_knob(self):
        text = _rfc()
        assert "new per-session configurable heartbeat" not in text or (
            "not added" in text.replace("*", "").lower()
        ), (
            "RFC must not add a new per-session heartbeat knob in Phase 1"
        )


class TestDocsOnlyScope:
    def test_rfc_states_no_endpoint_implementation(self):
        text = _rfc()
        assert "does **not** implement" in text or "does not implement" in text, (
            "RFC must state it does not implement the endpoint"
        )

    def test_rfc_has_non_goals_section(self):
        assert "Non-goals" in _rfc(), "RFC must have a Non-goals section"

    def test_rfc_lists_implementation_gates(self):
        text = _rfc()
        assert "implementation gate" in text.lower() or "open question" in text.lower(), (
            "RFC must list open implementation gates"
        )

    def test_contracts_preserves_no_implementation_warning(self):
        text = _contracts()
        assert "not authorize implementation" in text or (
            "Proposed RFCs are review guardrails, not implementation authorization" in text
        ), (
            "CONTRACTS.md must preserve warning that proposed RFCs do not authorize implementation"
        )