vnctalk-prosody/tests/test_03_vnctalk.py

"""VNCtalk-specific extension tests: vCard, timestamp, receipts, broadcast, avatar, open_host_store."""
import asyncio
import uuid
import pytest
import slixmpp
import xml.etree.ElementTree as ET


@pytest.mark.asyncio
class TestVnctalkExtensions:

    @pytest.mark.skip(reason="mod_vnc_vcard_fallback is not enabled in the config template; "
                        "enable the module first, then remove this skip")
    async def test_vcard_fallback(self, xmpp_client, xmpp_config):
        """Query vCard for a user that has none; server must auto-generate one."""
        try:
            res = await xmpp_client.get_vcard()
            vcard = res.xml.find("{vcard-temp}vCard")
            assert vcard is not None, "No vCard returned"
            fn = vcard.findtext("{vcard-temp}FN")
            assert fn is not None and fn.strip(), "mod_vnc_vcard_fallback did not generate FN"
        except slixmpp.exceptions.IqError as e:
            pytest.fail(f"vCard query failed: {e.condition}")

    async def test_vcard_avatar_upload_trigger(self, xmpp_client, xmpp_config):
        """Setting a vCard with PHOTO should trigger avatar upload (side-effect)."""
        vcard_xml = ET.Element("{vcard-temp}vCard")
        fn = ET.SubElement(vcard_xml, "{vcard-temp}FN")
        fn.text = "Test User"
        nick = ET.SubElement(vcard_xml, "{vcard-temp}NICKNAME")
        nick.text = "testuser"
        photo = ET.SubElement(vcard_xml, "{vcard-temp}PHOTO")
        ptype = ET.SubElement(photo, "{vcard-temp}TYPE")
        ptype.text = "image/png"
        pbin = ET.SubElement(photo, "{vcard-temp}BINVAL")
        pbin.text = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg=="
        try:
            await xmpp_client.set_vcard(vcard_xml)
        except slixmpp.exceptions.IqError as e:
            pytest.fail(f"vCard set failed: {e.condition}")

    async def test_timestamp_added(self, xmpp_client, xmpp_config, second_client):
        """mod_vnc_timestamp adds <stamp> to incoming chat messages."""
        received = asyncio.Event()
        stamps = []

        def on_message(msg):
            if msg["type"] == "chat":
                stamp = msg.xml.find("{xmpp:vnctalk:stamp}stamp")
                if stamp is not None:
                    stamps.append(stamp.get("stamp"))
                received.set()

        second_client.add_event_handler("message", on_message)
        try:
            msg = xmpp_client.make_message(mto=second_client.boundjid.bare, mtype="chat")
            msg["body"] = "test timestamp"  # slixmpp: dict-style, not msg.body
            msg.send()  # message.send() returns None, not awaitable
            await asyncio.wait_for(received.wait(), timeout=5)
            assert len(stamps) > 0, "No xmpp:vnctalk:stamp found on incoming message"
        finally:
            second_client.del_event_handler("message", on_message)

    async def test_receipts_archive(self, xmpp_client, xmpp_config, second_client):
        """Send a delivery receipt; server should accept it without error."""
        msg = xmpp_client.make_message(mto=second_client.boundjid.bare, mtype="chat")
        msg["body"] = "test receipt"  # slixmpp: dict-style, not msg.body
        msg_id = "receipt-test-1"
        msg["id"] = msg_id
        ET.SubElement(msg.xml, "{urn:xmpp:receipts}request")
        msg.send()  # message.send() returns None, not awaitable

        received = asyncio.Event()
        recv_id = None

        def on_message(msg):
            nonlocal recv_id
            if msg["type"] == "chat" and msg["body"]:
                recv_id = msg["id"]
                received.set()

        second_client.add_event_handler("message", on_message)
        try:
            await asyncio.wait_for(received.wait(), timeout=5)
        except asyncio.TimeoutError:
            pytest.fail("Message not received")

        receipt = second_client.make_message(mto=xmpp_client.boundjid.bare, mtype="chat")
        rec = ET.SubElement(receipt.xml, "{urn:xmpp:receipts}received")
        rec.set("id", recv_id or msg_id)
        receipt.send()  # message.send() returns None, not awaitable
        await asyncio.sleep(0.5)

    async def test_broadcast_component(self, xmpp_client, xmpp_config, second_client):
        """mod_vnc_broadcast fans a message out to its <to> recipients.

        The component is message-only (no disco#info / not in disco#items), so we
        exercise its actual function: post a <message> to broadcast@domain carrying
        a <vncTalkBroadcast> with a <to> recipient, and confirm that recipient
        receives the cast copy. Broadcasting is fire-and-forget (no IQ reply).
        """
        if xmpp_client.boundjid.bare == second_client.boundjid.bare:
            pytest.skip("Need a second distinct account to observe broadcast fanout")

        broadcast_jid = f"broadcast@{xmpp_config['domain']}"
        marker = f"bc-{uuid.uuid4().hex[:12]}"
        received = asyncio.Event()

        def on_message(msg):
            if (msg.xml.find("{xmpp:vnctalk}vncTalkBroadcast") is not None
                    and (msg["body"] or "") == marker):
                received.set()

        second_client.add_event_handler("message", on_message)
        try:
            msg = xmpp_client.make_message(mto=broadcast_jid, mtype="normal")
            msg["body"] = marker
            # no id on the child, else the module treats it as already-cast
            bc = ET.SubElement(msg.xml, "{xmpp:vnctalk}vncTalkBroadcast")
            bc.set("title", "broadcast test")
            to_el = ET.SubElement(bc, "{xmpp:vnctalk}to")
            to_el.text = second_client.boundjid.bare
            msg.send()  # message.send() returns None, not awaitable
            await asyncio.wait_for(received.wait(), timeout=8)
        except asyncio.TimeoutError:
            pytest.fail("Broadcast recipient did not receive the cast message")
        finally:
            second_client.del_event_handler("message", on_message)

    async def test_http_upload_external_disco(self, xmpp_client, xmpp_config):
        """Server should advertise HTTP upload slot service if mod_http_upload_external is loaded."""
        res = await xmpp_client.disco_items(to=xmpp_config["domain"])
        items = [i.get("jid") for i in res.xml.findall(".//{http://jabber.org/protocol/disco#items}item")]
        assert items is not None

    async def test_open_host_store_smoke(self, xmpp_client, xmpp_config):
        """Priority 3 from AUDIT: mod_vnc_fcm uses open_host_store via moduleapi patch.
        We cannot directly call open_host_store from XMPP, but we can verify the FCM
        token IQ endpoint works (it depends on the store being open).
        """
        # Add a token via the vnctalk:fcm IQ endpoint
        iq = xmpp_client.make_iq_set()
        add = ET.SubElement(iq.xml, "{xmpp:vnctalk:fcm}add")
        fcm = ET.SubElement(add, "{xmpp:vnctalk:fcm}fcm")
        fcm.set("device", "test-device-1")
        fcm.set("token", "test-token-abc123")
        fcm.set("os", "android")
        try:
            res = await iq.send()
            # Should get an IQ result (200-ish at XMPP level)
            assert res["type"] == "result", f"FCM token IQ failed: {res['type']}"
        except slixmpp.exceptions.IqError as e:
            # If the endpoint doesn't exist, that's a module load failure
            pytest.fail(f"FCM token store endpoint unreachable — open_host_store patch or mod_vnc_fcm may be broken: {e.condition}")


@pytest.mark.asyncio
class TestVnctalkMUCExtensions:
    """Tests that require a MUC component."""

    async def test_muc_e2e_config_field(self, xmpp_client, xmpp_config):
        """mod_vnc_e2ehints should expose muc#roominfo_e2e in disco#info."""
        muc_domain = xmpp_config.get("muc_domain")
        if not muc_domain:
            pytest.skip("No MUC domain configured")
        room_jid = f"e2e_{uuid.uuid4().hex[:8]}@{muc_domain}"
        nick = "test"
        await xmpp_client.join_muc(room_jid, nick)
        await asyncio.sleep(1)
        try:
            res = await xmpp_client.disco_info(to=room_jid)
            # the data form is nested inside <query>, not a direct child of <iq>
            x = res.xml.find(".//{jabber:x:data}x")
            if x is None:
                pytest.fail("No extended disco info form returned")
            fields = [f.get("var") for f in x.findall("{jabber:x:data}field")]
            assert "muc#roominfo_e2e" in fields, "mod_vnc_e2ehints field missing"
        finally:
            await xmpp_client.leave_muc(room_jid, nick)
            await xmpp_client.destroy_muc(room_jid)

    async def test_muc_data_broadcast_on_config_change(self, xmpp_client, xmpp_config, second_client):
        """mod_vnc_muc_data broadcasts room config on change."""
        muc_domain = xmpp_config.get("muc_domain")
        if not muc_domain:
            pytest.skip("No MUC domain configured")

        if xmpp_client.boundjid.bare == second_client.boundjid.bare:
            pytest.xfail("Need a second distinct account to observe broadcast")

        room_jid = f"configbc_{uuid.uuid4().hex[:8]}@{muc_domain}"
        nick1 = "owner"
        nick2 = "observer"

        await xmpp_client.join_muc(room_jid, nick1)
        # unlock the freshly-created (locked) room so the observer can join
        await xmpp_client.configure_muc(room_jid)
        await second_client.join_muc(room_jid, nick2)
        await asyncio.sleep(1)

        from slixmpp.xmlstream.handler import Callback
        from slixmpp.xmlstream.matcher import MatchXPath

        received = asyncio.Event()
        update_ns = "xmpp:vnctalk:update"

        # The update broadcast is a bodyless <message><x xmlns='xmpp:vnctalk:update'>,
        # so slixmpp's high-level "message" event (which requires a <body>) never
        # fires for it — catch it with a low-level stream handler instead.
        def on_update(stanza):
            if str(stanza["from"]).startswith(room_jid):
                received.set()

        handler_name = f"vnc_update_{uuid.uuid4().hex[:8]}"
        second_client.register_handler(Callback(
            handler_name,
            MatchXPath(f"{{{second_client.default_ns}}}message/{{{update_ns}}}x"),
            on_update,
        ))
        try:
            # mod_vnc_muc_data broadcasts <x xmlns='xmpp:vnctalk:update'> on
            # muc-config-submitted (not on subject change). Submit a config
            # change to trigger it now that the observer is in the room.
            await xmpp_client.configure_muc(
                room_jid, fields={"muc#roomconfig_vdata": "broadcast-test"}
            )
            await asyncio.wait_for(received.wait(), timeout=5)
        finally:
            second_client.remove_handler(handler_name)
            await xmpp_client.leave_muc(room_jid, nick1)
            await second_client.leave_muc(room_jid, nick2)
            await xmpp_client.destroy_muc(room_jid)


@pytest.mark.asyncio
class TestVncMucHook:
    """mod_vnc_muc_hook: notify an online, non-joined affiliated member."""

    async def test_muc_hook_notifies_nonjoined_affiliate(
        self, xmpp_client, second_client, xmpp_config
    ):
        muc_domain = xmpp_config.get("muc_domain")
        if not muc_domain:
            pytest.skip("No MUC domain configured")
        if xmpp_client.boundjid.bare == second_client.boundjid.bare:
            pytest.xfail("Need --xmpp-jid2 with a distinct account")

        room_jid = f"muchook_{uuid.uuid4().hex[:8]}@{muc_domain}"
        nick1 = "owner"
        await xmpp_client.join_muc(room_jid, nick1)
        await xmpp_client.configure_muc(room_jid)
        await asyncio.sleep(0.5)
        # Grant the second client membership WITHOUT them joining the room.
        await xmpp_client.set_muc_affiliation(room_jid, second_client.boundjid.bare, "member")
        await asyncio.sleep(0.5)

        notified = asyncio.Event()

        def on_message(msg):
            # mod_vnc_muc_hook sends either a muc#hook notification or a
            # mediated invite (when muc_notification_invite=true).
            if str(msg["from"]).startswith(room_jid):
                if msg.xml.find("{http://vnc.biz/xmpp/muc#hook}notification") is not None:
                    notified.set()
                elif msg.xml.find("{http://jabber.org/protocol/muc#user}x") is not None:
                    notified.set()

        second_client.add_event_handler("message", on_message)
        try:
            msg = xmpp_client.make_message(mto=room_jid, mtype="groupchat")
            msg["body"] = f"hook probe {uuid.uuid4().hex[:6]}"
            msg["id"] = f"hook-{uuid.uuid4().hex[:8]}"
            msg.send()
            await asyncio.wait_for(notified.wait(), timeout=8)
        finally:
            second_client.del_event_handler("message", on_message)
            await xmpp_client.leave_muc(room_jid, nick1)
            await xmpp_client.destroy_muc(room_jid)