echo-core/tests/test_voice_recv_dave.py

"""DAVE receive-side decrypt tests for the vendored voice-recv fork.

Exercises Lane A's patch on
`vendor/discord-ext-voice-recv/discord/ext/voice_recv/reader.py`:

* `_maybe_dave_decrypt(rtp_packet)` — DAVE E2E layer sandwiched between the
  transport-layer decrypt and the routing into the opus decoder. No-op when
  the room is non-DAVE, when davey isn't installed, or when the SSRC map
  hasn't caught up to a new speaker yet.
* `callback()` hook — feeds the DAVE-unwrapped plaintext into
  `packet_router.feed_rtp()` on success, drops the packet on failure WITHOUT
  killing the reader thread.

The test fixtures mirror `tests/test_voice_session_cleanup.py:33-54`:
  * Construct `AudioReader` via `AudioReader.__new__(AudioReader)` + manual
    attr set so the reader thread is never started.
  * `MagicMock` everything below the unit under test.

`_HAS_DAVE` / `_MEDIA_TYPE_AUDIO` on the reader module are monkey-patched per
test so the suite passes whether or not `davey` is importable in the venv.
The assertions only become meaningful once Lane A's patch has landed and the
package has been re-installed (`pip install -e vendor/discord-ext-voice-recv
--force-reinstall`); the FILE itself is valid Python regardless.

See plan: /home/moltbot/.claude/plans/wiggly-exploring-glade.md
"""
from __future__ import annotations

from unittest.mock import MagicMock

import pytest

from discord.ext.voice_recv.reader import AudioReader


# Sentinel for `_MEDIA_TYPE_AUDIO`. Using a plain object() keeps the tests
# independent of whether davey is importable — we just assert the value
# flows through to `dave_session.decrypt()` unchanged.
_FAKE_MEDIA_TYPE_AUDIO = object()


# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------


@pytest.fixture
def fake_dave_session():
    sess = MagicMock(name="dave_session")
    sess.ready = True
    # Default: this user is NOT in passthrough — DAVE decrypt must run.
    # Individual tests can override to True to exercise the passthrough path.
    sess.can_passthrough = MagicMock(return_value=False)
    sess.decrypt = MagicMock(return_value=b"plaintext_opus")
    return sess


@pytest.fixture
def fake_connection(fake_dave_session):
    conn = MagicMock(name="_connection")
    conn.dave_protocol_version = 1
    conn.dave_session = fake_dave_session
    return conn


@pytest.fixture
def fake_voice_client(fake_connection):
    vc = MagicMock(name="voice_client")
    vc._connection = fake_connection
    vc._ssrc_to_id = {12345: 999_000}
    return vc


@pytest.fixture
def fake_rtp_packet():
    pkt = MagicMock(name="rtp_packet")
    pkt.ssrc = 12345
    pkt.decrypted_data = b"ciphertext_after_transport_decrypt"
    pkt.is_silence = MagicMock(return_value=False)
    return pkt


@pytest.fixture
def reader(fake_voice_client):
    """`AudioReader` instance with no reader thread spawned.

    Same pattern used by `tests/test_voice_session_cleanup.py` for
    `VoiceSession` — bypass `__init__` so we can drive the public surface
    against pure mocks.
    """
    r = AudioReader.__new__(AudioReader)
    r.voice_client = fake_voice_client
    r.error = None
    return r


@pytest.fixture
def dave_enabled(monkeypatch):
    """Force the reader module's DAVE-availability flags ON.

    Pins `_MEDIA_TYPE_AUDIO` to a known sentinel so the happy-path test can
    assert exactly what gets passed to `dave_session.decrypt`. `raising=False`
    keeps the monkeypatch valid even if Lane A's patch hasn't landed yet —
    the tests will still fail (no `_maybe_dave_decrypt` attr), just for the
    right reason.
    """
    import discord.ext.voice_recv.reader as reader_mod

    monkeypatch.setattr(reader_mod, "_HAS_DAVE", True, raising=False)
    monkeypatch.setattr(
        reader_mod, "_MEDIA_TYPE_AUDIO", _FAKE_MEDIA_TYPE_AUDIO, raising=False
    )
    return reader_mod


# ---------------------------------------------------------------------------
# Unit tests: `_maybe_dave_decrypt`
# ---------------------------------------------------------------------------


class TestMaybeDaveDecrypt:
    """Seven unit tests on the DAVE-decrypt gate.

    The gate mirrors `voice_client.can_encrypt` in discord.py 2.7.1 exactly
    (`voice_state.py:272-273`). Bypass semantics on every "DAVE inactive"
    branch let non-DAVE rooms and davey-less environments keep working.
    """

    def test_protocol_version_zero_bypasses_decrypt(
        self, dave_enabled, reader, fake_connection, fake_dave_session, fake_rtp_packet,
    ):
        """`dave_protocol_version == 0` → return the transport-decrypted
        payload unchanged; never touch `dave_session.decrypt`."""
        fake_connection.dave_protocol_version = 0
        result = reader._maybe_dave_decrypt(fake_rtp_packet)
        assert result is fake_rtp_packet.decrypted_data
        fake_dave_session.decrypt.assert_not_called()

    def test_dave_session_none_bypasses_decrypt(
        self, dave_enabled, reader, fake_connection, fake_rtp_packet,
    ):
        """`dave_session is None` → bypass. Pre-MLS-handshake state."""
        fake_connection.dave_session = None
        result = reader._maybe_dave_decrypt(fake_rtp_packet)
        assert result is fake_rtp_packet.decrypted_data

    def test_dave_session_not_ready_bypasses_decrypt(
        self, dave_enabled, reader, fake_dave_session, fake_rtp_packet,
    ):
        """`dave_session.ready is False` → bypass. Pre-MLS-epoch-1 packets
        are transport-only on the wire."""
        fake_dave_session.ready = False
        result = reader._maybe_dave_decrypt(fake_rtp_packet)
        assert result is fake_rtp_packet.decrypted_data
        fake_dave_session.decrypt.assert_not_called()

    def test_unknown_ssrc_returns_none(
        self, dave_enabled, reader, fake_voice_client, fake_dave_session, fake_rtp_packet,
    ):
        """SSRC not in `_ssrc_to_id` → drop (return None).

        Accepted regression: davey requires per-user keys; when SPEAKING
        events race behind the first audio packet, 1-5 packets per new
        speaker per session are dropped. See plan §Edge cases.
        """
        fake_voice_client._ssrc_to_id.clear()
        result = reader._maybe_dave_decrypt(fake_rtp_packet)
        assert result is None
        fake_dave_session.decrypt.assert_not_called()

    def test_happy_path_invokes_decrypt_and_returns_plaintext(
        self, dave_enabled, reader, fake_dave_session, fake_rtp_packet,
    ):
        """Full DAVE-active path: `decrypt(user_id, MediaType.audio, ciphertext)`
        called exactly once with the expected args; method returns the
        davey plaintext bytes verbatim."""
        ciphertext = fake_rtp_packet.decrypted_data
        result = reader._maybe_dave_decrypt(fake_rtp_packet)
        assert result == b"plaintext_opus"
        fake_dave_session.decrypt.assert_called_once_with(
            999_000, _FAKE_MEDIA_TYPE_AUDIO, ciphertext,
        )

    def test_decrypt_raises_returns_none_no_crash(
        self, dave_enabled, reader, fake_dave_session, fake_rtp_packet,
    ):
        """davey.decrypt raising → drop the packet, don't propagate, and
        leave `reader.error` untouched so the reader thread stays alive.

        MLS epoch transitions can produce transient decrypt failures —
        bumping `reader.error` would call `self.stop()` and kill the whole
        receive pipeline."""
        fake_dave_session.decrypt.side_effect = RuntimeError(
            "simulated MLS epoch transition fail"
        )
        result = reader._maybe_dave_decrypt(fake_rtp_packet)
        assert result is None
        assert reader.error is None

    def test_has_dave_false_bypasses_even_with_session_present(
        self, monkeypatch, reader, fake_dave_session, fake_rtp_packet,
    ):
        """`_HAS_DAVE = False` → bypass everything, even if a real session
        somehow showed up on the connection. Defensive shim that keeps the
        tests (and any davey-less deploys) green."""
        import discord.ext.voice_recv.reader as reader_mod

        monkeypatch.setattr(reader_mod, "_HAS_DAVE", False, raising=False)
        result = reader._maybe_dave_decrypt(fake_rtp_packet)
        assert result is fake_rtp_packet.decrypted_data
        fake_dave_session.decrypt.assert_not_called()

    def test_can_passthrough_true_returns_payload_without_decrypt(
        self, dave_enabled, reader, fake_dave_session, fake_rtp_packet,
    ):
        """`can_passthrough(user_id) == True` → return the transport-decrypted
        payload as-is; never call `decrypt`. Mirrors Discord's protocol where
        a passthrough-mode peer sends non-DAVE-wrapped packets that the
        receiver must accept verbatim."""
        fake_dave_session.can_passthrough = MagicMock(return_value=True)
        result = reader._maybe_dave_decrypt(fake_rtp_packet)
        assert result is fake_rtp_packet.decrypted_data
        fake_dave_session.can_passthrough.assert_called_once_with(999_000)
        fake_dave_session.decrypt.assert_not_called()

    def test_can_passthrough_raises_falls_through_to_decrypt(
        self, dave_enabled, reader, fake_dave_session, fake_rtp_packet,
    ):
        """`can_passthrough` raising → swallow the error and try `decrypt`.
        Defensive: an older davey build or transient internal state shouldn't
        break the receive pipeline."""
        fake_dave_session.can_passthrough = MagicMock(
            side_effect=RuntimeError("simulated davey internal error")
        )
        result = reader._maybe_dave_decrypt(fake_rtp_packet)
        assert result == b"plaintext_opus"
        fake_dave_session.decrypt.assert_called_once()


# ---------------------------------------------------------------------------
# Integration tests: `callback()` exercises the DAVE hook
# ---------------------------------------------------------------------------


class TestCallbackIntegration:
    """Two integration tests for the hook Lane A inserts between transport
    decrypt (reader.py:141) and the post-decrypt routing (reader.py:159).

    Strategy: stub the transport-decrypt and RTP parsing path so `callback()`
    reaches the hook, then mock `_maybe_dave_decrypt` directly on the reader
    instance. The assertion focuses on `feed_rtp` being called (test 8) vs.
    not called (test 9). The transport path correctness is covered by
    voice-recv's own upstream tests.
    """

    @staticmethod
    def _wire_callback(reader, monkeypatch, fake_rtp_packet):
        import discord.ext.voice_recv.reader as reader_mod

        # Redirect rtp parsing — we want an RTP path (not RTCP) so the hook fires.
        monkeypatch.setattr(reader_mod.rtp, "is_rtcp", lambda data: False)
        monkeypatch.setattr(reader_mod.rtp, "decode_rtp", lambda data: fake_rtp_packet)

        # Stub the instance attrs `callback()` touches besides the hook.
        reader.decryptor = MagicMock(name="decryptor")
        reader.decryptor.decrypt_rtp = MagicMock(return_value=b"ciphertext")
        reader.packet_router = MagicMock(name="packet_router")
        reader.packet_router.feed_rtp = MagicMock()
        reader.speaking_timer = MagicMock(name="speaking_timer")
        reader.sink = MagicMock(name="sink")

    def test_callback_feeds_when_dave_returns_bytes(
        self, monkeypatch, reader, fake_rtp_packet,
    ):
        """Hook returns plaintext → `feed_rtp` called once with the
        rtp_packet whose `decrypted_data` is now the post-DAVE plaintext."""
        self._wire_callback(reader, monkeypatch, fake_rtp_packet)
        plaintext = b"dave_unwrapped_opus_payload"
        reader._maybe_dave_decrypt = MagicMock(return_value=plaintext)

        reader.callback(b"raw_packet_bytes")

        reader._maybe_dave_decrypt.assert_called_once_with(fake_rtp_packet)
        assert reader.packet_router.feed_rtp.call_count == 1
        called_with = reader.packet_router.feed_rtp.call_args[0][0]
        assert called_with is fake_rtp_packet
        assert fake_rtp_packet.decrypted_data == plaintext
        assert reader.error is None

    def test_callback_drops_when_dave_returns_none(
        self, monkeypatch, reader, fake_rtp_packet,
    ):
        """Hook returns None → `feed_rtp` NOT called, no exception propagated,
        `reader.error` stays None (reader thread survives the drop)."""
        self._wire_callback(reader, monkeypatch, fake_rtp_packet)
        reader._maybe_dave_decrypt = MagicMock(return_value=None)

        reader.callback(b"raw_packet_bytes")

        reader._maybe_dave_decrypt.assert_called_once_with(fake_rtp_packet)
        reader.packet_router.feed_rtp.assert_not_called()
        assert reader.error is None