vnctalk-prosody/tests/test_10_smacks.py

"""Upstream mod_smacks parity tests (test-improvement.md Phase 4).

After the custom vnctalk/mod_smacks fork is dropped in favour of upstream
mod_smacks + mod_smacks_offline, these verify the core SM contracts VNCtalk
relies on still hold: <enable>, <r>/<a> ack exchange, hibernation within
smacks_hibernation_time, and session resume replaying queued stanzas.

This is a *parity/regression* test, not a fork-feature test.

NOTE: the resume/expiry tests drive slixmpp's xep_0198 plugin, which is
fiddly. The expiry/offline tests need a low smacks_hibernation_time (set via
SMACKS_HIBERNATION_TIME in docker-compose.yml, e.g. 10s); if the configured
value is large they are skipped to avoid multi-minute waits.
"""
import asyncio
import os
import uuid

import pytest

import slixmpp

from conftest import VNCXmppClient


def _hibernation_time():
    """The server's smacks_hibernation_time, as known to the test harness."""
    try:
        return int(os.environ.get("SMACKS_HIBERNATION_TIME", "300"))
    except ValueError:
        return 300


class SmacksClient(VNCXmppClient):
    """VNCXmppClient with XEP-0198 stream management enabled for resume.

    Tracks sm_enabled / session_resumed / sm_failed so tests can await the
    right event on a fresh connect vs. a resuming reconnect.
    """

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.register_plugin("xep_0198")
        self.sm = self.plugin["xep_0198"]
        self.sm.allow_resume = True
        self.sm_enabled_event = asyncio.Event()
        self.resumed_event = asyncio.Event()
        self.sm_failed_event = asyncio.Event()
        self.acked_count = 0
        self.add_event_handler("sm_enabled", self._on_sm_enabled)
        self.add_event_handler("session_resumed", self._on_resumed)
        self.add_event_handler("sm_failed", self._on_failed)
        self.add_event_handler("stanza_acked", self._on_acked)

    def _on_sm_enabled(self, _e):
        self.sm_enabled_event.set()

    def _on_resumed(self, _e):
        self.resumed_event.set()
        # On resume, session_start is not re-fired; announce presence manually.
        try:
            self.send_presence()
        except Exception:
            pass

    def _on_failed(self, _e):
        self.sm_failed_event.set()

    def _on_acked(self, _e):
        self.acked_count = self.sm.last_ack

    def abrupt_disconnect(self):
        """Close the TCP abruptly without </stream:stream>.

        The fork mod_smacks hibernates on abrupt disconnect; a clean stream
        close tears the session down, so resume must be exercised via an
        aborted transport. SM state (sm_id) is preserved.
        """
        self.end_session_on_disconnect = False
        try:
            if self.transport is not None:
                self.transport.abort()
        except Exception:
            pass

    async def connect_fresh(self, timeout=30):
        """Connect and negotiate a new SM session (wait for sm_enabled)."""
        self.sm_enabled_event.clear()
        self.resumed_event.clear()
        self.sm_failed_event.clear()
        await self.connect(self._connect_host, self._connect_port)
        done, pending = await asyncio.wait(
            {
                asyncio.create_task(self.sm_enabled_event.wait()),
                asyncio.create_task(self.session_started_event.wait()),
            },
            timeout=timeout,
            return_when=asyncio.FIRST_COMPLETED,
        )
        if not done:
            raise asyncio.TimeoutError("SM enable/session_start timed out")
        # Announce presence so the server routes bare-JID messages to this
        # session instead of offline storage (mirrors VNCXmppClient.async_connect).
        self.send_presence()
        await asyncio.sleep(0.3)

    async def reconnect_and_wait(self, timeout=30):
        """Reconnect; wait for either session_resumed or sm_failed."""
        self.resumed_event.clear()
        self.sm_failed_event.clear()
        # session_started persists from the first connect; clear it so the
        # wait below doesn't return immediately on a stale event. On a
        # successful resume session_start is NOT re-fired (only session_resumed).
        self.session_started_event.clear()
        await self.connect(self._connect_host, self._connect_port)
        done, pending = await asyncio.wait(
            {
                asyncio.create_task(self.resumed_event.wait()),
                asyncio.create_task(self.sm_failed_event.wait()),
            },
            timeout=timeout,
            return_when=asyncio.FIRST_COMPLETED,
        )
        for t in pending:
            t.cancel()

    @property
    def sm_id(self):
        return self.sm.sm_id


@pytest.fixture
def smacks_config(xmpp_config):
    if xmpp_config.get("skip_live") or not xmpp_config.get("jid2"):
        pytest.skip("SMACKS tests need a live server and a second account")
    return xmpp_config


async def _make_sm_client(cfg):
    client = SmacksClient(
        cfg["jid2"], cfg["password2"],
        host=cfg["host"], port=cfg["port"], use_ssl=True, verify_ssl=cfg["verify_ssl"],
    )
    return client


@pytest.mark.asyncio
class TestSmacksEnableAck:
    async def test_smacks_enable_and_ack(self, smacks_config, xmpp_client):
        """Negotiate SM, send 3 messages, request ack, get <a h=3/>.

        <r/> asks the server to ack the stanzas the CLIENT has sent, so the SM
        client is the sender here.
        """
        cfg = smacks_config
        sm = await _make_sm_client(cfg)
        try:
            await sm.connect_fresh(timeout=30)
            assert sm.sm.enabled_in, "SM was not enabled (<enabled> not received)"

            # SM client sends three messages → these are counted by the server.
            before = sm.sm.last_ack
            for i in range(3):
                msg = sm.make_message(mto=xmpp_client.boundjid.bare, mtype="chat")
                msg["body"] = f"sm-ack-{i}-{uuid.uuid4().hex[:6]}"
                msg["id"] = f"smack-{uuid.uuid4().hex[:8]}"
                msg.send()

            # Request an ack and poll last_ack (slixmpp may also auto-request
            # once its window fills, so poll rather than wait on a single event).
            await asyncio.sleep(0.5)
            sm.sm.request_ack()  # send <r/>
            for _ in range(40):
                if sm.sm.last_ack >= before + 3:
                    break
                await asyncio.sleep(0.25)
            assert sm.sm.last_ack >= before + 3, (
                f"server <a/> reported h={sm.sm.last_ack}, expected >={before + 3}"
            )
        finally:
            if sm.session_started_event.is_set() or sm.resumed_event.is_set():
                await sm.async_disconnect()


@pytest.mark.asyncio
class TestSmacksResume:
    async def test_smacks_resume_replays_queue(self, smacks_config, xmpp_client):
        """A hibernated session resumes and replays messages queued while offline."""
        cfg = smacks_config
        sm = await _make_sm_client(cfg)
        try:
            await sm.connect_fresh(timeout=30)
            assert sm.sm_id, "no SM id granted — server did not allow resumption"
            previd = sm.sm_id

            # Collect chat messages addressed to the SM client.
            received = []

            def on_message(msg):
                if msg["type"] == "chat" and msg["body"]:
                    received.append(str(msg["body"]))

            sm.add_event_handler("message", on_message)

            # Disconnect cleanly; with SM+resume the server hibernates the session.
            sm.abrupt_disconnect()
            await asyncio.wait_for(sm.disconnected_event.wait(), timeout=10)
            await asyncio.sleep(1.0)

            # Send two messages while the SM client is hibernated → server queues them.
            queued = [f"queued-{i}-{uuid.uuid4().hex[:6]}" for i in range(2)]
            for b in queued:
                msg = xmpp_client.make_message(mto=cfg["jid2"], mtype="chat")
                msg["body"] = b
                msg["id"] = f"q-{uuid.uuid4().hex[:8]}"
                msg.send()
            await asyncio.sleep(1.0)

            # Reconnect within the hibernation window → resume replays the queue.
            await sm.reconnect_and_wait(timeout=30)
            assert sm.resumed_event.is_set(), (
                "resume failed (<failed> or no <resumed>) — session did not resume"
            )
            assert sm.sm_id == previd, "resumed session has a different SM id"

            for _ in range(40):
                if len(received) >= 2:
                    break
                await asyncio.sleep(0.25)
            for b in queued:
                assert b in received, (
                    f"queued message {b!r} was not replayed after resume; got {received}"
                )
        finally:
            if sm.session_started_event.is_set() or sm.resumed_event.is_set():
                await sm.async_disconnect()


@pytest.mark.asyncio
class TestSmacksHibernationExpiry:
    async def test_smacks_hibernation_expiry(self, smacks_config, xmpp_client):
        """After smacks_hibernation_time elapses, resume returns <failed>."""
        htime = _hibernation_time()
        if htime > 30:
            pytest.skip(
                f"smacks_hibernation_time={htime}s is too large for an automated expiry test"
            )

        cfg = smacks_config
        sm = await _make_sm_client(cfg)
        try:
            await sm.connect_fresh(timeout=30)
            assert sm.sm_id, "server did not allow resumption"

            sm.abrupt_disconnect()
            await asyncio.wait_for(sm.disconnected_event.wait(), timeout=10)
            # Wait past the hibernation window so the server drops the session.
            await asyncio.sleep(htime + 3)

            await sm.reconnect_and_wait(timeout=30)
            assert sm.sm_failed_event.is_set(), (
                "expected <failed> after hibernation expiry, but resume succeeded"
            )
        finally:
            if sm.session_started_event.is_set() or sm.resumed_event.is_set():
                await sm.async_disconnect()


@pytest.mark.asyncio
class TestSmacksOfflineReplay:
    @pytest.mark.xfail(
        reason=(
            "Premise not satisfied by the vnctalk mod_smacks fork in this stack: "
            "mod_smacks_offline logs 'no longer required' and, with MAM enabled, "
            "stanzas queued during hibernation are already in MAM and are not "
            "flushed to mod_offline storage on expiry. Expected to xpass once the "
            "fork is removed in favour of upstream mod_smacks + mod_smacks_offline "
            "(test-improvement.md §0, Phase 4)."
        ),
        strict=False,
    )
    async def test_smacks_offline_queue_after_expiry(self, smacks_config, xmpp_client):
        """mod_smacks_offline: messages queued during hibernation are delivered
        from offline storage on the next fresh connect after expiry."""
        htime = _hibernation_time()
        if htime > 30:
            pytest.skip(
                f"smacks_hibernation_time={htime}s is too large for an automated offline test"
            )

        cfg = smacks_config
        sm = await _make_sm_client(cfg)
        try:
            await sm.connect_fresh(timeout=30)
            assert sm.sm_id, "server did not allow resumption"

            queued_body = f"offline-after-expiry-{uuid.uuid4().hex[:6]}"

            received = []

            def on_message(msg):
                if msg["body"]:
                    received.append(str(msg["body"]))

            sm.add_event_handler("message", on_message)

            sm.abrupt_disconnect()
            await asyncio.wait_for(sm.disconnected_event.wait(), timeout=10)
            await asyncio.sleep(1.0)

            msg = xmpp_client.make_message(mto=cfg["jid2"], mtype="chat")
            msg["body"] = queued_body
            msg["id"] = f"off-{uuid.uuid4().hex[:8]}"
            msg.send()

            # Wait past expiry so the hibernated session is gone.
            await asyncio.sleep(htime + 3)

            # Fresh connect (new SM session) — mod_smacks_offline should replay
            # the queued message from offline storage.
            sm.sm_enabled_event.clear()
            sm.session_started_event.clear()
            await sm.connect_fresh(timeout=30)

            for _ in range(40):
                if queued_body in received:
                    break
                await asyncio.sleep(0.25)
            assert queued_body in received, (
                f"queued message {queued_body!r} not delivered from offline storage; "
                f"got {received}"
            )
        finally:
            if sm.session_started_event.is_set() or sm.resumed_event.is_set():
                await sm.async_disconnect()