vnctalk-prosody/tests/test_02_muc.py

"""MUC-specific tests for vnctalk requirements."""
import asyncio
import pytest
import slixmpp
from slixmpp.exceptions import IqError, IqTimeout


@pytest.mark.asyncio
class TestMuc:

    async def test_muc_component_disco(self, xmpp_client, xmpp_config):
        """MUC component must advertise MUC and MAM features."""
        muc_domain = xmpp_config.get("muc_domain")
        if not muc_domain:
            pytest.skip("No MUC domain configured")
        res = await xmpp_client.disco_info(to=muc_domain)
        features = [f.get("var") for f in res.xml.findall(".//{http://jabber.org/protocol/disco#info}feature")]
        assert "http://jabber.org/protocol/muc" in features, "MUC feature missing"
        assert "urn:xmpp:mam:2" in features, "MUC MAM feature missing"

    async def test_muc_room_creation(self, xmpp_client, xmpp_config):
        """Create an instant room and verify join."""
        muc_domain = xmpp_config.get("muc_domain")
        if not muc_domain:
            pytest.skip("No MUC domain configured")

        room_jid = f"testroom_{slixmpp.jid.nodeprep(xmpp_client.boundjid.local)}{muc_domain}"
        nick = "testuser"

        presence = xmpp_client.make_presence(pto=f"{room_jid}/{nick}")
        x = slixmpp.etree.SubElement(presence.xml, "{http://jabber.org/protocol/muc}x")
        await presence.send()

        # Wait a moment for room creation/join
        await asyncio.sleep(1)
        # If we didn't get an error, basic join succeeded.
        # In a full test we'd listen for presence from the room.

    async def test_muc_mam_query(self, xmpp_client, xmpp_config):
        """Query MUC MAM archive."""
        muc_domain = xmpp_config.get("muc_domain")
        if not muc_domain:
            pytest.skip("No MUC domain configured")

        room_jid = f"testroom_mam@{muc_domain}"
        # Join first
        nick = "testuser"
        presence = xmpp_client.make_presence(pto=f"{room_jid}/{nick}")
        x = slixmpp.etree.SubElement(presence.xml, "{http://jabber.org/protocol/muc}x")
        await presence.send()
        await asyncio.sleep(1)

        iq = xmpp_client.make_iq_set()
        query = slixmpp.etree.SubElement(iq.xml, "{urn:xmpp:mam:2}query")
        query.set("queryid", "test-muc-mam-1")
        # Send to room
        iq.attrib["to"] = room_jid
        try:
            res = await iq.send()
            fin = res.xml.find("{urn:xmpp:mam:2}fin")
            assert fin is not None, "MUC MAM query did not return <fin>"
        except IqError as e:
            # If room is not persistent or empty, may get item-not-found; that's ok for infra test.
            assert e.condition in ("item-not-found", "not-allowed")

    async def test_muc_automember(self, xmpp_client, xmpp_config, second_client):
        """Invite a user; they should automatically gain member affiliation."""
        muc_domain = xmpp_config.get("muc_domain")
        if not muc_domain:
            pytest.skip("No MUC domain configured")

        room_jid = f"testroom_auto_{slixmpp.jid.nodeprep(xmpp_client.boundjid.local)}@{muc_domain}"
        nick1 = "owner"
        nick2 = "invited"

        # Client 1 joins and becomes owner
        p1 = xmpp_client.make_presence(pto=f"{room_jid}/{nick1}")
        slixmpp.etree.SubElement(p1.xml, "{http://jabber.org/protocol/muc}x")
        await p1.send()
        await asyncio.sleep(1)

        # Client 1 invites client 2
        invite_msg = xmpp_client.make_message(mto=room_jid, mtype="normal")
        x = slixmpp.etree.SubElement(invite_msg.xml, "{http://jabber.org/protocol/muc#user}x")
        invite = slixmpp.etree.SubElement(x, "{http://jabber.org/protocol/muc#user}invite")
        invite.set("to", second_client.boundjid.bare)
        await invite_msg.send()
        await asyncio.sleep(1)

        # Client 2 joins
        p2 = second_client.make_presence(pto=f"{room_jid}/{nick2}")
        slixmpp.etree.SubElement(p2.xml, "{http://jabber.org/protocol/muc}x")
        await p2.send()
        await asyncio.sleep(1)

        # Query affiliations as owner
        iq = xmpp_client.make_iq_get(to=room_jid)
        query = slixmpp.etree.SubElement(iq.xml, "{http://jabber.org/protocol/muc#admin}query")
        item = slixmpp.etree.SubElement(query, "{http://jabber.org/protocol/muc#admin}item")
        item.set("affiliation", "member")
        try:
            res = await iq.send()
            members = res.xml.findall(".//{http://jabber.org/protocol/muc#admin}item")
            member_jids = [m.get("jid") for m in members]
            assert second_client.boundjid.bare in member_jids, "Automember did not grant membership"
        except IqError as e:
            pytest.skip(f"Could not query affiliations: {e.condition}")

    async def test_muc_vdata_disco(self, xmpp_client, xmpp_config):
        """MUC disco#info should contain vdata form field if mod_vnc_muc_data is loaded."""
        muc_domain = xmpp_config.get("muc_domain")
        if not muc_domain:
            pytest.skip("No MUC domain configured")
        room_jid = f"testroom_vd@{muc_domain}"
        nick = "test"
        p = xmpp_client.make_presence(pto=f"{room_jid}/{nick}")
        slixmpp.etree.SubElement(p.xml, "{http://jabber.org/protocol/muc}x")
        await p.send()
        await asyncio.sleep(1)

        res = await xmpp_client.disco_info(to=room_jid)
        # Look for extended info form
        x = res.xml.find("{jabber:x:data}x")
        if x is not None:
            fields = [f.get("var") for f in x.findall("{jabber:x:data}field")]
            # If mod_vnc_muc_data is loaded, muc#roominfo_vdata should appear
            if "muc#roominfo_vdata" not in fields:
                pytest.skip("mod_vnc_muc_data not loaded or field not present")
        else:
            pytest.skip("No extended disco info form returned")