libervia-backend/tests/unit/test_plugin_xep_0356.py

#!/usr/bin/env python3

# Libervia: an XMPP client
# Copyright (C) 2009-2026 Jérôme Poisson (goffi@goffi.org)

# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.

# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.

# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

"""
Tests for XEP-0356 Privileged Entity implementation.
"""

from unittest.mock import MagicMock

from pytest import fixture, raises
from pytest_twisted import ensureDeferred as ed
from twisted.internet import reactor
from twisted.words.protocols.jabber import jid
from twisted.words.xish import domish

from libervia.backend import G
from libervia.backend.plugins.plugin_xep_0356 import (
    XEP_0356,
    NS_PRIVILEGE,
    ClientPrivilegeData,
    RosterCacheEntry,
    IMPORT_NAME,
    PrivilegeError,
)
from tx_xmpp.xmppim import RosterItem, NS_ROSTER


@fixture
def xep_0356(host):
    """Create XEP-0356 plugin instance."""
    return XEP_0356(host)


@fixture(autouse=True)
def setup_component_client(client):
    """Set up client as a component for XEP-0356 tests."""
    client.is_component = True
    _priv_data = ClientPrivilegeData()
    client.get_plugin_data = lambda import_name, data_class: _priv_data


def make_privilege_advertisement(privileges):
    """Helper to create privilege advertisement message.

    @param privileges: Dict of privilege type -> access level
        e.g., {"roster": "both", "message": "outgoing"}
    """
    msg = domish.Element(
        (None, "message"),
        attribs={"from": "test.example", "to": "pubsub.test.example"},
    )
    privilege = domish.Element((NS_PRIVILEGE, "privilege"))

    for access, perm_type in privileges.items():
        perm_attrs = {"access": access}
        if perm_type:
            perm_attrs["type"] = perm_type
        if access == "roster" and perm_type in ("get", "both"):
            perm_attrs["push"] = "true"
        perm = domish.Element((NS_PRIVILEGE, "perm"), attribs=perm_attrs)
        privilege.addChild(perm)

    msg.addChild(privilege)
    return msg


def make_iq_privilege_advertisement(namespaces):
    """Helper to create IQ privilege advertisement.

    @param namespaces: Dict of namespace -> type
        e.g., {"http://jabber.org/protocol/pubsub": "set"}
    """
    msg = domish.Element(
        (None, "message"),
        attribs={"from": "test.example", "to": "pubsub.test.example"},
    )
    privilege = domish.Element((NS_PRIVILEGE, "privilege"))
    perm = domish.Element((NS_PRIVILEGE, "perm"), attribs={"access": "iq"})

    for ns, perm_type in namespaces.items():
        ns_elt = domish.Element(
            (NS_PRIVILEGE, "namespace"),
            attribs={"ns": ns, "type": perm_type},
        )
        perm.addChild(ns_elt)

    privilege.addChild(perm)
    msg.addChild(privilege)
    return msg


class TestXEP0356Advertisement:
    """Tests for privilege advertisement handling."""

    @ed
    async def test_parses_roster_privilege_both(self, xep_0356, client):
        """Roster privilege with type='both' sets access correctly."""
        msg = make_privilege_advertisement({"roster": "both"})
        xep_0356.on_advertise(msg, client)

        data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData)
        assert data.privileges.roster_access == "both"
        assert xep_0356.can_access_roster(client)
        assert xep_0356.can_modify_roster(client)
        assert xep_0356.can_receive_roster_pushes(client)

    @ed
    async def test_parses_roster_privilege_get(self, xep_0356, client):
        """Roster privilege with type='get' allows read-only access."""
        msg = make_privilege_advertisement({"roster": "get"})
        xep_0356.on_advertise(msg, client)

        data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData)
        assert data.privileges.roster_access == "get"
        assert xep_0356.can_access_roster(client)
        assert not xep_0356.can_modify_roster(client)
        assert xep_0356.can_receive_roster_pushes(client)

    @ed
    async def test_parses_roster_privilege_set(self, xep_0356, client):
        """Roster privilege with type='set' allows write-only access."""
        msg = make_privilege_advertisement({"roster": "set"})
        xep_0356.on_advertise(msg, client)

        data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData)
        assert data.privileges.roster_access == "set"
        assert not xep_0356.can_access_roster(client)
        assert xep_0356.can_modify_roster(client)
        assert not xep_0356.can_receive_roster_pushes(client)

    @ed
    async def test_parses_message_privilege(self, xep_0356, client):
        """Message privilege is correctly set."""
        msg = make_privilege_advertisement({"message": "outgoing"})
        xep_0356.on_advertise(msg, client)

        data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData)
        assert data.privileges.message_privilege == "outgoing"
        assert xep_0356.can_send_messages(client)

    @ed
    async def test_parses_presence_privilege_managed_entity(self, xep_0356, client):
        """Presence privilege with managed_entity type."""
        msg = make_privilege_advertisement({"presence": "managed_entity"})
        xep_0356.on_advertise(msg, client)

        data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData)
        assert data.privileges.presence_privilege == "managed_entity"
        assert xep_0356.can_receive_managed_presence(client)
        assert not xep_0356.can_receive_roster_presence(client)

    @ed
    async def test_parses_presence_privilege_roster(self, xep_0356, client):
        """Presence privilege with roster type."""
        msg = make_privilege_advertisement({"presence": "roster"})
        xep_0356.on_advertise(msg, client)

        data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData)
        assert data.privileges.presence_privilege == "roster"
        assert xep_0356.can_receive_managed_presence(client)
        assert xep_0356.can_receive_roster_presence(client)

    @ed
    async def test_parses_iq_privilege(self, xep_0356, client):
        """IQ privilege with namespace is correctly set."""
        msg = make_iq_privilege_advertisement(
            {"http://jabber.org/protocol/pubsub": "set"}
        )
        xep_0356.on_advertise(msg, client)

        data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData)
        assert "http://jabber.org/protocol/pubsub" in data.privileges.iq_permissions
        assert xep_0356.can_send_iq(client, "http://jabber.org/protocol/pubsub", "set")
        assert not xep_0356.can_send_iq(
            client, "http://jabber.org/protocol/pubsub", "get"
        )

    @ed
    async def test_parses_iq_privilege_both(self, xep_0356, client):
        """IQ privilege with type='both' allows get and set."""
        msg = make_iq_privilege_advertisement(
            {"http://jabber.org/protocol/pubsub": "both"}
        )
        xep_0356.on_advertise(msg, client)

        assert xep_0356.can_send_iq(client, "http://jabber.org/protocol/pubsub", "get")
        assert xep_0356.can_send_iq(client, "http://jabber.org/protocol/pubsub", "set")

    @ed
    async def test_rejects_advertisement_from_wrong_server(self, xep_0356, client):
        """Advertisement from wrong server raises StanzaError."""
        from twisted.words.protocols.jabber.error import StanzaError

        msg = make_privilege_advertisement({"roster": "both"})
        msg["from"] = "other.example"

        with raises(StanzaError, match="not-allowed"):
            xep_0356.on_advertise(msg, client)


class TestXEP0356RosterAccess:
    """Tests for roster access operations."""

    @ed
    async def test_get_roster_without_privilege_raises(self, xep_0356, client):
        """get_roster raises PrivilegeError without roster privilege."""
        user_jid = jid.JID("user@test.example")

        with raises(PrivilegeError, match="Roster access not permitted"):
            await xep_0356.get_roster(client, user_jid)

    @ed
    async def test_get_roster_sends_request(self, xep_0356, client, client_IQ):
        """get_roster sends proper roster get request."""
        # Grant privilege
        msg = make_privilege_advertisement({"roster": "both"})
        xep_0356.on_advertise(msg, client)

        user_jid = jid.JID("user@test.example")
        contact_jid = jid.JID("contact@test.example")

        # Create mock IQ with roster response
        response_elt = domish.Element((None, "iq"))
        response_elt["type"] = "result"
        item = domish.Element((NS_ROSTER, "item"))
        item["jid"] = contact_jid.full()
        item["subscription"] = "both"
        response_elt.addChild(item)

        client.IQ = client_IQ(response_elt=response_elt)

        # Call get_roster
        roster = await xep_0356.get_roster(client, user_jid)

        # We check that the request was sent correctly
        assert len(client.IQ.sent_iqs) == 1
        sent_iq = client.IQ.sent_iqs[0]
        assert sent_iq["type"] == "get"
        assert sent_iq["to"] == user_jid.full()
        assert len(list(sent_iq.elements("jabber:iq:roster", "query"))) == 1

        # We check that roster was parsed correctly
        assert len(roster) == 1
        assert contact_jid in roster
        assert isinstance(roster[contact_jid], RosterItem)
        assert roster[contact_jid].subscriptionTo
        assert roster[contact_jid].subscriptionFrom

    @ed
    async def test_get_roster_uses_cache(self, xep_0356, client, client_IQ):
        """get_roster uses cached roster if not expired."""
        # Grant privilege
        msg = make_privilege_advertisement({"roster": "both"})
        xep_0356.on_advertise(msg, client)

        user_jid = jid.JID("user@test.example")

        # Pre-populate cache
        data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData)
        cached_item = RosterItem(user_jid)
        cached_item.subscriptionTo = True
        data.roster_cache[user_jid] = RosterCacheEntry(
            roster={user_jid: cached_item},
            version="v1",
            # Fresh timestamp
            timestamp=reactor.seconds(),
        )

        client.IQ = client_IQ()

        roster = await xep_0356.get_roster(client, user_jid)

        # We check that cache was used (no IQ sent)
        assert len(client.IQ.sent_iqs) == 0
        assert len(roster) == 1

    @ed
    async def test_on_roster_push_updates_cache(self, xep_0356, client):
        """on_roster_push updates the roster cache."""
        data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData)
        # We want to be sure that roster is empty
        assert len(data.roster_cache) == 0

        # Grant roster push privilege
        msg = make_privilege_advertisement({"roster": "both"})
        xep_0356.on_advertise(msg, client)

        owner_jid = jid.JID("user@test.example")

        # Create roster push stanza
        iq = domish.Element((None, "iq"))
        iq["from"] = owner_jid.full()
        iq["type"] = "set"
        query = domish.Element(("jabber:iq:roster", "query"))
        item = domish.Element(("jabber:iq:roster", "item"))
        item["jid"] = "contact@test.example"
        item["subscription"] = "both"
        query.addChild(item)
        iq.addChild(query)

        xep_0356.on_roster_push(client, iq)

        # We check that cache was updated
        assert owner_jid in data.roster_cache
        cached_entry = data.roster_cache[owner_jid]
        assert isinstance(cached_entry, RosterCacheEntry)
        cached_roster = cached_entry.roster
        assert len(cached_roster) == 1
        contact_jid = jid.JID("contact@test.example")
        assert contact_jid in cached_roster
        assert isinstance(cached_roster[contact_jid], RosterItem)

    @ed
    async def test_on_roster_push_remove_item(self, xep_0356, client):
        """on_roster_push removes item when subscription='remove'."""
        # Grant roster push privilege
        msg = make_privilege_advertisement({"roster": "both"})
        xep_0356.on_advertise(msg, client)

        owner_jid = jid.JID("user@test.example")

        # Pre-populate cache with an item
        data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData)
        existing_item = RosterItem(jid.JID("contact@test.example"))
        data.roster_cache[owner_jid] = RosterCacheEntry(
            roster={jid.JID("contact@test.example"): existing_item},
            version="v1",
            timestamp=reactor.seconds(),
        )

        # Create roster push with remove
        iq = domish.Element((None, "iq"))
        iq["from"] = owner_jid.full()
        iq["type"] = "set"
        query = domish.Element(("jabber:iq:roster", "query"))
        item = domish.Element(("jabber:iq:roster", "item"))
        item["jid"] = "contact@test.example"
        item["subscription"] = "remove"
        query.addChild(item)
        iq.addChild(query)

        xep_0356.on_roster_push(client, iq)

        # We check that item was removed
        cached_entry = data.roster_cache[owner_jid]
        cached_roster = cached_entry.roster
        assert len(cached_roster) == 0

    @ed
    async def test_on_roster_push_without_privilege(self, xep_0356, client):
        """on_roster_push is ignored without roster push privilege."""
        # Don't grant privilege
        owner_jid = jid.JID("user@test.example")

        iq = domish.Element((None, "iq"))
        iq["from"] = owner_jid.full()
        iq["type"] = "set"
        query = domish.Element(("jabber:iq:roster", "query"))
        iq.addChild(query)

        xep_0356.on_roster_push(client, iq)

        # We check that cache was not updated
        data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData)
        assert owner_jid not in data.roster_cache


class TestXEP0356PrivilegedMessages:
    """Tests for privileged message sending."""

    @ed
    async def test_send_privileged_message_without_privilege_raises(
        self, xep_0356, client
    ):
        """send_privileged_message raises PrivilegeError without message privilege."""
        message = domish.Element((None, "message"))
        message["from"] = "user@test.example"

        with raises(PrivilegeError, match="Message privilege not permitted"):
            await xep_0356.send_privileged_message(client, message)

    @ed
    async def test_send_privileged_message_without_from_raises(self, xep_0356, client):
        """send_privileged_message raises PrivilegeError if message has no 'from'."""
        # Grant privilege
        msg = make_privilege_advertisement({"message": "outgoing"})
        xep_0356.on_advertise(msg, client)

        message = domish.Element((None, "message"))

        with raises(PrivilegeError, match='Message must have a "from" attribute'):
            await xep_0356.send_privileged_message(client, message)

    @ed
    async def test_send_privileged_message_wrong_server_raises(self, xep_0356, client):
        """send_privileged_message raises if from is not from our server."""
        # Grant privilege
        msg = make_privilege_advertisement({"message": "outgoing"})
        xep_0356.on_advertise(msg, client)

        message = domish.Element((None, "message"))
        message["from"] = "user@other.example"

        with raises(PrivilegeError, match="from_jid must be a bare JID from our server"):
            await xep_0356.send_privileged_message(client, message)

    @ed
    async def test_send_privileged_message_with_resource_raises(self, xep_0356, client):
        """send_privileged_message raises if from has a resource."""
        # Grant privilege
        msg = make_privilege_advertisement({"message": "outgoing"})
        xep_0356.on_advertise(msg, client)

        message = domish.Element((None, "message"))
        message["from"] = "user@test.example/resource"

        with raises(PrivilegeError, match="from_jid must be a bare JID from our server"):
            await xep_0356.send_privileged_message(client, message)

    @ed
    async def test_send_privileged_message_sends_wrapped_message(self, xep_0356, client):
        """send_privileged_message sends properly wrapped message."""
        # Grant privilege
        msg = make_privilege_advertisement({"message": "outgoing"})
        xep_0356.on_advertise(msg, client)

        from_jid = jid.JID("user@test.example")
        to_jid = jid.JID("other@test.example")
        inner_msg = domish.Element(
            (None, "message"),
            attribs={"from": from_jid.full(), "to": to_jid.full()},
        )
        body = domish.Element((None, "body"))
        body.textContent = "Hello"
        inner_msg.addChild(body)

        # Track sent messages
        sent_messages = []

        def mock_send(elt):
            sent_messages.append(elt)

        client.xmlstream.send = mock_send

        # Call send_privileged_message
        await xep_0356.send_privileged_message(client, inner_msg)

        # We check that message was sent
        assert len(sent_messages) == 1
        wrapper = sent_messages[0]

        assert wrapper.name == "message"
        assert wrapper["from"] == client.jid.full()
        assert wrapper["to"] == client.server_jid.full()
        assert "id" in wrapper.attributes

        # We check that privilege wrapper
        privilege = next(wrapper.elements(NS_PRIVILEGE, "privilege"))
        assert privilege is not None


class TestXEP0356PrivilegedIQ:
    """Tests for privileged IQ sending."""

    @ed
    async def test_send_privileged_iq_without_privilege_raises(self, xep_0356, client):
        """send_privileged_iq raises PrivilegeError without IQ privilege."""
        iq = domish.Element(("jabber:client", "iq"))
        iq["type"] = "set"
        iq["to"] = "user@test.example"
        payload = domish.Element(("http://jabber.org/protocol/pubsub", "publish"))
        iq.addChild(payload)

        with raises(PrivilegeError, match="IQ privilege not permitted"):
            await xep_0356.send_privileged_iq(client, iq)

    @ed
    async def test_send_privileged_iq_no_to_raises(self, xep_0356, client):
        """send_privileged_iq raises if IQ has no 'to' attribute."""
        # Grant privilege
        msg = make_iq_privilege_advertisement(
            {"http://jabber.org/protocol/pubsub": "set"}
        )
        xep_0356.on_advertise(msg, client)

        iq = domish.Element(("jabber:client", "iq"))
        iq["type"] = "set"
        payload = domish.Element(("http://jabber.org/protocol/pubsub", "publish"))
        iq.addChild(payload)

        with raises(PrivilegeError, match="IQ must have a 'to' attribute"):
            await xep_0356.send_privileged_iq(client, iq)

    @ed
    async def test_send_privileged_iq_wrong_server_raises(self, xep_0356, client):
        """send_privileged_iq raises if to is not from our server."""
        # Grant privilege
        msg = make_iq_privilege_advertisement(
            {"http://jabber.org/protocol/pubsub": "set"}
        )
        xep_0356.on_advertise(msg, client)

        iq = domish.Element(("jabber:client", "iq"))
        iq["type"] = "set"
        iq["to"] = "user@other.example"
        payload = domish.Element(("http://jabber.org/protocol/pubsub", "publish"))
        iq.addChild(payload)

        with raises(PrivilegeError, match="user_jid must be a bare JID from our server"):
            await xep_0356.send_privileged_iq(client, iq)

    @ed
    async def test_send_privileged_iq_no_payload_raises(self, xep_0356, client):
        """send_privileged_iq raises if IQ has no payload."""
        # Grant privilege
        msg = make_iq_privilege_advertisement(
            {"http://jabber.org/protocol/pubsub": "set"}
        )
        xep_0356.on_advertise(msg, client)

        iq = domish.Element(("jabber:client", "iq"))
        iq["type"] = "set"
        iq["to"] = "user@test.example"

        with raises(PrivilegeError, match="IQ has no payload"):
            await xep_0356.send_privileged_iq(client, iq)

    @ed
    async def test_send_privileged_iq_unauthorized_namespace_raises(
        self, xep_0356, client
    ):
        """send_privileged_iq raises for unauthorized namespace."""
        # Grant privilege for pubsub only
        msg = make_iq_privilege_advertisement(
            {"http://jabber.org/protocol/pubsub": "set"}
        )
        xep_0356.on_advertise(msg, client)

        iq = domish.Element(("jabber:client", "iq"))
        iq["type"] = "set"
        iq["to"] = "user@test.example"
        payload = domish.Element(("http://jabber.org/protocol/bookmarks", "bookmarks"))
        iq.addChild(payload)

        with raises(PrivilegeError, match="IQ privilege not permitted"):
            await xep_0356.send_privileged_iq(client, iq)

    @ed
    async def test_send_privileged_iq_sends_wrapped_iq(self, xep_0356, client, client_IQ):
        """send_privileged_iq sends properly wrapped IQ."""
        # Grant privilege
        msg = make_iq_privilege_advertisement(
            {"http://jabber.org/protocol/pubsub": "set"}
        )
        xep_0356.on_advertise(msg, client)

        user_jid = jid.JID("user@test.example")
        inner_iq = domish.Element(
            ("jabber:client", "iq"), attribs={"type": "set", "id": "inner1"}
        )
        pubsub = domish.Element(("http://jabber.org/protocol/pubsub", "publish"))
        pubsub["node"] = "test_node"
        inner_iq.addChild(pubsub)
        inner_iq["to"] = user_jid.full()

        # Create mock response
        response_elt = domish.Element((None, "iq"))
        response_elt["type"] = "result"
        privilege = domish.Element((NS_PRIVILEGE, "privilege"))
        forwarded = domish.Element(("urn:xmpp:forward:0", "forwarded"))
        inner_response = domish.Element(("jabber:client", "iq"))
        inner_response["type"] = "result"
        inner_response["id"] = "inner1"
        forwarded.addChild(inner_response)
        privilege.addChild(forwarded)
        response_elt.addChild(privilege)

        client.IQ = client_IQ(response_elt=response_elt)

        result = await xep_0356.send_privileged_iq(client, inner_iq)

        # We check that the request was sent correctly
        assert len(client.IQ.sent_iqs) == 1
        sent_iq = client.IQ.sent_iqs[0]
        assert sent_iq["type"] == "set"
        assert sent_iq["to"] == user_jid.full()

        # We check that privilege wrapper
        priv_iq = next(sent_iq.elements(NS_PRIVILEGE, "privileged_iq"))
        assert priv_iq is not None

        # We check that result was unwrapped
        assert result["type"] == "result"
        assert result["id"] == "inner1"

    @ed
    async def test_unwrap_privileged_iq_result_invalid_raises(self, xep_0356):
        """_unwrap_privileged_iq_result raises for invalid response format."""
        # Invalid response (no privilege wrapper)
        invalid_response = domish.Element((None, "iq"))
        invalid_response["type"] = "result"

        with raises(PrivilegeError, match="Invalid response format"):
            xep_0356._unwrap_privileged_iq_result(invalid_response)


class TestXEP0356PresenceHandling:
    """Tests for presence handling."""

    @ed
    async def test_profile_connected_does_not_register_observers(self, xep_0356, client):
        """profile_connected no longer registers observers (done by handler instead)."""
        assert client.is_component

        # Mock addObserver to track calls
        observer_calls = []

        def mock_add_observer(xpath, handler, client=None):
            observer_calls.append((xpath, handler, client))

        client.xmlstream.addObserver = mock_add_observer

        # Call profile_connected
        xep_0356.profile_connected(client)

        # We check that no observers were registered (handler does it instead)
        assert len(observer_calls) == 0

    @ed
    async def test_profile_connected_skips_non_component(self, xep_0356, host):
        """profile_connected skips non-component clients."""
        # Create a non-component client
        non_component_client = MagicMock()
        non_component_client.is_component = False
        non_component_client.jid = jid.JID("user@test.example/resource")

        # Mock addObserver to track calls
        observer_calls = []

        def mock_add_observer(xpath, handler, client=None):
            observer_calls.append((xpath, handler, client))

        non_component_client.xmlstream.addObserver = mock_add_observer

        xep_0356.profile_connected(non_component_client)

        # We check that observer was NOT registered
        assert len(observer_calls) == 0

    @ed
    async def test_on_presence_triggers_async_point(self, xep_0356, client):
        """_on_presence triggers async_point for any presence."""
        # Create presence stanza
        presence = domish.Element((None, "presence"))
        presence["from"] = "user@test.example"

        trigger_point_calls = []

        async def mock_async_point(point_name, *args):
            trigger_point_calls.append((point_name, args))

        G.host.trigger.async_point = mock_async_point

        await xep_0356.on_presence(presence, client)

        assert len(trigger_point_calls) == 1
        point_name, args = trigger_point_calls[0]
        assert point_name == "XEP-0356_presence_received"
        assert len(args) == 3
        assert args[0] == client
        assert args[1] == jid.JID("user@test.example")
        assert args[2] == presence

    @ed
    async def test_on_presence_with_unavailable_type(self, xep_0356, client):
        """_on_presence triggers async_point for unavailable presence."""
        presence = domish.Element((None, "presence"))
        presence["from"] = "user.test.example"
        presence["type"] = "unavailable"

        trigger_point_calls = []

        async def mock_async_point(point_name, *args):
            trigger_point_calls.append((point_name, args))

        G.host.trigger.async_point = mock_async_point

        await xep_0356.on_presence(presence, client)

        # We check that async_point was triggered
        assert len(trigger_point_calls) == 1
        assert trigger_point_calls[0][0] == "XEP-0356_presence_received"

    @ed
    async def test_presence_tracking_add_and_remove_resource(self, xep_0356, client):
        """_track_presence adds and removes resources from the presence data."""
        data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData)
        bare_jid = jid.JID("user@test.example")

        assert len(data.presence_data.online_resources) == 0

        presence_available = domish.Element((None, "presence"))
        presence_available["from"] = "user@test.example/resource1"
        await xep_0356.on_presence(presence_available, client)

        assert bare_jid in data.presence_data.online_resources
        assert "resource1" in data.presence_data.online_resources[bare_jid]

        presence_unavailable = domish.Element((None, "presence"))
        presence_unavailable["from"] = "user@test.example/resource1"
        presence_unavailable["type"] = "unavailable"
        await xep_0356.on_presence(presence_unavailable, client)

        assert bare_jid in data.presence_data.online_resources
        assert "resource1" not in data.presence_data.online_resources[bare_jid]

    @ed
    async def test_presence_tracking_multiple_resources(self, xep_0356, client):
        """_track_presence tracks multiple resources for the same bare JID."""
        data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData)
        bare_jid = jid.JID("user@test.example")

        assert len(data.presence_data.online_resources) == 0

        presence1 = domish.Element((None, "presence"))
        presence1["from"] = "user@test.example/resource1"
        await xep_0356.on_presence(presence1, client)

        assert bare_jid in data.presence_data.online_resources
        assert "resource1" in data.presence_data.online_resources[bare_jid]

        presence2 = domish.Element((None, "presence"))
        presence2["from"] = "user@test.example/resource2"
        await xep_0356.on_presence(presence2, client)

        assert "resource1" in data.presence_data.online_resources[bare_jid]
        assert "resource2" in data.presence_data.online_resources[bare_jid]

    @ed
    async def test_presence_tracking_ignores_bare_jid(self, xep_0356, client):
        """_track_presence ignores presence from bare JIDs (no resource)."""
        data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData)
        bare_jid = jid.JID("user@test.example")

        assert len(data.presence_data.online_resources) == 0

        presence = domish.Element((None, "presence"))
        presence["from"] = "user@test.example"

        await xep_0356.on_presence(presence, client)

        assert bare_jid not in data.presence_data.online_resources