diff --git a/libervia/backend/plugins/plugin_xep_0356.py b/libervia/backend/plugins/plugin_xep_0356.py
new file mode 100644
index 00000000..40acc340
--- /dev/null
+++ b/libervia/backend/plugins/plugin_xep_0356.py
@@ -0,0 +1,581 @@
+#!/usr/bin/env python3
+
+# Libervia plugin for Privileged Entity (XEP-0356).
+# 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 .
+
+from dataclasses import dataclass, field
+from textwrap import dedent
+from typing import TYPE_CHECKING, Final, Literal, cast
+import uuid
+
+from twisted.internet import reactor
+from twisted.words.protocols.jabber import error, jid
+from twisted.words.protocols.jabber.xmlstream import XMPPHandler
+from twisted.words.xish import domish
+from tx_xmpp import disco, itx_xmpp
+from tx_xmpp.xmppim import RosterItem
+from zope.interface import implementer
+
+from libervia.backend import G
+from libervia.backend.core import exceptions
+from libervia.backend.core.constants import Const as C
+from libervia.backend.core.core_types import SatXMPPComponent, SatXMPPEntity
+from libervia.backend.core.i18n import _
+from libervia.backend.core.log import getLogger
+from libervia.backend.plugins.plugin_xep_0297 import NS_FORWARD, XEP_0297
+
+if TYPE_CHECKING:
+ from libervia.backend.core.main import LiberviaBackend
+
+log = getLogger(__name__)
+
+PLUGIN_INFO = {
+ C.PI_NAME: "Privileged Entity",
+ C.PI_IMPORT_NAME: "XEP-0356",
+ C.PI_TYPE: "XEP",
+ C.PI_MODES: [C.PLUG_MODE_COMPONENT],
+ C.PI_PROTOCOLS: ["XEP-0356"],
+ C.PI_DEPENDENCIES: ["XEP-0297"],
+ C.PI_RECOMMENDATIONS: [],
+ C.PI_MAIN: "XEP_0356",
+ C.PI_HANDLER: "yes",
+ C.PI_DESCRIPTION: dedent(_("""\
+ Privileged entity implementation for components.
+
+ This plugin provides privileged entity capabilities to components, allowing them to:
+ - Access and modify user rosters (with roster push support)
+ - Send messages on behalf of server
+ - Send IQ stanzas on behalf of server
+ - Receive presence information for managed entities and roster contacts
+ """)),
+}
+
+NS_PRIVILEGE: Final = "urn:xmpp:privilege:2"
+NS_ROSTER = "jabber:iq:roster"
+ROSTER_TTL = 3600
+IMPORT_NAME = "XEP-0356"
+
+RosterPrivilegeType = Literal["none", "get", "set", "both"]
+MessagePrivilegeType = Literal["none", "outgoing"]
+PresencePrivilegeType = Literal["none", "managed_entity", "roster"]
+
+
+@dataclass
+class IQPermission:
+ """IQ permission for a namespace."""
+
+ get: bool
+ set: bool
+
+
+@dataclass
+class RosterCacheEntry:
+ """Cached roster data for a managed entity."""
+
+ roster: dict[jid.JID, RosterItem]
+ version: str | None
+ timestamp: float
+
+
+@dataclass
+class PrivilegeData:
+ """Data structure for privilege tracking."""
+
+ roster_access: RosterPrivilegeType = "none"
+ roster_push: bool = False
+ message_privilege: MessagePrivilegeType = "none"
+ presence_privilege: PresencePrivilegeType = "none"
+ iq_permissions: dict[str, IQPermission] = field(default_factory=dict)
+
+
+class PrivilegeError(exceptions.PermissionError):
+ """A privileged operation failed"""
+
+ pass
+
+
+@dataclass
+class ClientPrivilegeData:
+ privileges: PrivilegeData = field(default_factory=lambda: PrivilegeData())
+ roster_cache: dict[jid.JID, RosterCacheEntry] = field(default_factory=dict)
+
+
+class XEP_0356:
+ """XEP-0356: Privileged Entity handler."""
+
+ namespace = NS_PRIVILEGE
+
+ def __init__(self, host: "LiberviaBackend") -> None:
+ log.info(f"plugin {PLUGIN_INFO[C.PI_NAME]!r} initialization")
+ self._forward = cast(XEP_0297, host.plugins["XEP-0297"])
+ host.register_namespace("privilege", NS_PRIVILEGE)
+
+ def get_handler(self, client: SatXMPPEntity) -> "XEP0356Handler":
+ """Return the handler for this plugin.
+
+ @param client: The client entity.
+ @return: The XEP0356Handler instance.
+ """
+ return XEP0356Handler(self)
+
+ def profile_connected(self, client: SatXMPPEntity) -> None:
+ """Called when a client is connected.
+
+ Observers are now registered in XEP0356Handler.connectionInitialized()
+ to avoid race conditions with privilege advertisements.
+
+ @param client: The client connected.
+ """
+ if not client.is_component:
+ return
+ log.debug(f"XEP-0356 profile_connected called for {client.profile}")
+
+ def can_access_roster(self, client: SatXMPPComponent) -> bool:
+ """Check if this entity can access user rosters."""
+ data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData)
+ return data.privileges.roster_access in ("get", "both")
+
+ def can_modify_roster(self, client: SatXMPPComponent) -> bool:
+ """Check if this entity can modify user rosters."""
+ data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData)
+ return data.privileges.roster_access in ("set", "both")
+
+ def can_receive_roster_pushes(self, client: SatXMPPComponent) -> bool:
+ """Check if this entity receives roster pushes."""
+ data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData)
+ return data.privileges.roster_push
+
+ def can_send_messages(self, client: SatXMPPComponent) -> bool:
+ """Check if this entity can send privileged messages."""
+ data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData)
+ return data.privileges.message_privilege == "outgoing"
+
+ def can_receive_managed_presence(self, client: SatXMPPComponent) -> bool:
+ """Check if this entity can receive presence for managed entities."""
+ data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData)
+ return data.privileges.presence_privilege in ("managed_entity", "roster")
+
+ def can_receive_roster_presence(self, client: SatXMPPComponent) -> bool:
+ """Check if this entity can receive presence for roster contacts."""
+ data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData)
+ return data.privileges.presence_privilege == "roster"
+
+ def can_send_iq(self, client: SatXMPPComponent, namespace: str, iq_type: str) -> bool:
+ """Check if this entity can send privileged IQ for a namespace/type.
+
+ @param client: The client to check.
+ @param namespace: The namespace of the IQ payload.
+ @param iq_type: The type of IQ ("get" or "set").
+ @return: True if the entity can send this IQ.
+ """
+ data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData)
+ perms = data.privileges.iq_permissions.get(namespace)
+ if perms is None:
+ return False
+ return perms.get if iq_type == "get" else perms.set
+
+ def on_advertise(self, msg: domish.Element, client: SatXMPPComponent) -> None:
+ """Handle privilege advertisement from the server.
+
+ @param msg: The message stanza containing privilege information.
+ @param client: The client receiving the advertisement.
+ """
+ try:
+ privilege_elt = next(msg.elements(NS_PRIVILEGE, "privilege"))
+ except StopIteration:
+ log.warning("Privilege advertisement without privilege element")
+ return
+
+ server_jid = client.server_jid
+ if jid.JID(msg["from"]) != server_jid:
+ log.warning(
+ f"SECURITY: privilege advertisement from unexpected server: {msg['from']}"
+ )
+ raise error.StanzaError("not-allowed")
+
+ data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData)
+ data.privileges = PrivilegeData()
+
+ for perm in privilege_elt.elements(NS_PRIVILEGE, "perm"):
+ try:
+ access = perm["access"]
+ except KeyError:
+ log.warning(f"perm element without access attribute: {perm.toXml()}")
+ continue
+
+ if access == "roster":
+ self._handle_roster_perm(data, perm)
+ elif access == "message":
+ self._handle_message_perm(data, perm)
+ elif access == "presence":
+ self._handle_presence_perm(data, perm)
+ elif access == "iq":
+ self._handle_iq_perm(data, perm)
+ else:
+ log.warning(f"Unknown privilege access type: {access}")
+
+ log.info(
+ f"Privileges updated for {client.profile}:\n"
+ f" roster: {data.privileges.roster_access} "
+ f"(push={data.privileges.roster_push})\n"
+ f" message: {data.privileges.message_privilege}\n"
+ f" presence: {data.privileges.presence_privilege}\n"
+ f" iq: {list(data.privileges.iq_permissions.keys())}"
+ )
+
+ def _handle_roster_perm(
+ self, data: ClientPrivilegeData, perm: domish.Element
+ ) -> None:
+ """Handle roster permission element."""
+ try:
+ perm_type = perm["type"]
+ except KeyError:
+ log.warning(f"roster perm without type: {perm.toXml()}")
+ return
+
+ if perm_type not in ("none", "get", "set", "both"):
+ log.warning(f"Invalid roster permission type: {perm_type}")
+ return
+
+ data.privileges.roster_access = cast(RosterPrivilegeType, perm_type)
+
+ push_attr = perm.getAttribute("push")
+ if push_attr is not None:
+ data.privileges.roster_push = push_attr.lower() == "true"
+ else:
+ data.privileges.roster_push = perm_type in ("get", "both")
+
+ def _handle_message_perm(
+ self, data: ClientPrivilegeData, perm: domish.Element
+ ) -> None:
+ """Handle message permission element."""
+ try:
+ perm_type = perm["type"]
+ except KeyError:
+ log.warning(f"message perm without type: {perm.toXml()}")
+ return
+
+ if perm_type not in ("none", "outgoing"):
+ log.warning(f"Invalid message permission type: {perm_type}")
+ return
+
+ data.privileges.message_privilege = cast(MessagePrivilegeType, perm_type)
+
+ def _handle_presence_perm(
+ self, data: ClientPrivilegeData, perm: domish.Element
+ ) -> None:
+ """Handle presence permission element."""
+ try:
+ perm_type = perm["type"]
+ except KeyError:
+ log.warning(f"presence perm without type: {perm.toXml()}")
+ return
+
+ if perm_type not in ("none", "managed_entity", "roster"):
+ log.warning(f"Invalid presence permission type: {perm_type}")
+ return
+
+ data.privileges.presence_privilege = cast(PresencePrivilegeType, perm_type)
+
+ def _handle_iq_perm(self, data: ClientPrivilegeData, perm: domish.Element) -> None:
+ """Handle IQ permission element."""
+ for ns_elt in perm.elements(NS_PRIVILEGE, "namespace"):
+ try:
+ ns = ns_elt["ns"]
+ perm_type = ns_elt["type"]
+ except KeyError:
+ log.warning(
+ f"namespace element without required attributes: {ns_elt.toXml()}"
+ )
+ continue
+
+ if perm_type not in ("get", "set", "both"):
+ log.warning(f"Invalid IQ permission type: {perm_type}")
+ continue
+
+ data.privileges.iq_permissions[ns] = IQPermission(
+ get=perm_type in ("get", "both"),
+ set=perm_type in ("set", "both"),
+ )
+
+ def on_roster_push(self, client: SatXMPPComponent, iq_elt: domish.Element) -> None:
+ """Handle roster push from server.
+
+ @param client: The client receiving the roster push.
+ @param iq_elt: The roster push IQ stanza.
+ """
+ if not self.can_receive_roster_pushes(client):
+ log.debug("Roster push received but not configured")
+ return
+
+ from_jid_str = iq_elt.getAttribute("from")
+ if not from_jid_str:
+ log.warning(f"Roster push without 'from' attribute: {iq_elt.toXml()}")
+ return
+
+ from_jid = jid.JID(from_jid_str)
+ if from_jid.resource:
+ log.warning(
+ f"Roster push from full JID instead of bare JID: {iq_elt.toXml()}"
+ )
+ return
+
+ data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData)
+ cache = data.roster_cache
+
+ # Parse roster item using tx-xmpp
+ query = next(iq_elt.elements(NS_ROSTER, "query"))
+ item_elt = next(query.elements(NS_ROSTER, "item"))
+ roster_item = RosterItem.fromElement(item_elt)
+ item_jid = roster_item.entity
+
+ cached_entry = cache.get(from_jid)
+ roster = cached_entry.roster if cached_entry else {}
+ if roster_item.remove:
+ roster.pop(item_jid, None)
+ else:
+ roster[item_jid] = roster_item
+
+ version = query.getAttribute("ver")
+ cache[from_jid] = RosterCacheEntry(
+ roster=roster,
+ version=version,
+ timestamp=reactor.seconds(),
+ )
+
+ log.debug(f"Roster push for {from_jid}: {item_jid}")
+
+ async def get_roster(
+ self, client: SatXMPPComponent, user_jid: jid.JID
+ ) -> dict[jid.JID, RosterItem]:
+ """Get the roster of a managed entity.
+
+ @param client: The client making the request.
+ @param user_jid: The bare JID of the user whose roster to retrieve.
+ @return: Dict mapping JIDs to RosterItem objects.
+ @raise PrivilegeError: If roster access is not permitted.
+ """
+ if not self.can_access_roster(client):
+ raise PrivilegeError(f"Roster access not permitted for {user_jid}")
+
+ data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData)
+ cache = data.roster_cache
+ cached = cache.get(user_jid)
+ if cached and (reactor.seconds() - cached.timestamp) < ROSTER_TTL:
+ return cached.roster
+
+ iq = client.IQ("get")
+ iq["to"] = user_jid.full()
+ query = domish.Element((NS_ROSTER, "query"))
+ iq.addChild(query)
+
+ result = await iq.send()
+
+ roster = {}
+ for item_elt in result.elements(NS_ROSTER, "item"):
+ roster_item = RosterItem.fromElement(item_elt)
+ roster[roster_item.entity] = roster_item
+
+ version = query.getAttribute("ver")
+ cache[user_jid] = RosterCacheEntry(
+ roster=roster,
+ version=version,
+ timestamp=reactor.seconds(),
+ )
+
+ return roster
+
+ async def send_privileged_message(
+ self, client: SatXMPPComponent, message_elt: domish.Element
+ ) -> None:
+ """Send a message on behalf of a user.
+
+ @param client: The client sending the message.
+ @param message_elt: The message stanza to send (will be wrapped in forwarded).
+ @raise PrivilegeError: If message privilege is not permitted.
+ """
+ if not self.can_send_messages(client):
+ raise PrivilegeError("Message privilege not permitted")
+
+ try:
+ from_attr = message_elt["from"]
+ except KeyError:
+ raise PrivilegeError('Message must have a "from" attribute')
+
+ from_jid = jid.JID(from_attr)
+ if from_jid.resource or jid.JID(from_jid.host) != client.server_jid:
+ raise PrivilegeError(
+ f"from_jid must be a bare JID from our server: {from_jid}"
+ )
+
+ wrapper = domish.Element((None, "message"))
+ wrapper["from"] = client.jid.full()
+ wrapper["to"] = client.server_jid.full()
+ wrapper["id"] = str(uuid.uuid4())
+
+ privilege = domish.Element((NS_PRIVILEGE, "privilege"))
+ forwarded = self._forward.build_forwarded(message_elt)
+ privilege.addChild(forwarded)
+ wrapper.addChild(privilege)
+
+ client.xmlstream.send(wrapper)
+
+ async def send_privileged_iq(
+ self, client: SatXMPPComponent, iq_elt: domish.Element
+ ) -> domish.Element:
+ """Send an IQ stanza on behalf of a user.
+
+ @param client: The client sending the IQ.
+ @param iq_elt: The IQ stanza to send (will be wrapped).
+ @return: The IQ response.
+ @raise PrivilegeError: If IQ privilege is not permitted.
+ """
+ data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData)
+ if not data.privileges.iq_permissions:
+ raise PrivilegeError("IQ privilege not permitted")
+
+ to_attr = iq_elt.getAttribute("to")
+ if not to_attr:
+ raise PrivilegeError("IQ must have a 'to' attribute")
+
+ user_jid = jid.JID(to_attr)
+ if user_jid.resource or jid.JID(user_jid.host) != client.server_jid:
+ raise PrivilegeError(
+ f"user_jid must be a bare JID from our server: {user_jid}"
+ )
+
+ iq_type = cast(str, iq_elt.getAttribute("type", ""))
+ first_child = iq_elt.firstChildElement()
+ if first_child is None:
+ raise PrivilegeError("IQ has no payload")
+
+ namespace = first_child.uri
+ if not self.can_send_iq(client, namespace, iq_type):
+ raise PrivilegeError(
+ f"IQ privilege not permitted for namespace {namespace} type {iq_type}"
+ )
+
+ wrapper = client.IQ(iq_type)
+ wrapper["to"] = user_jid.full()
+
+ priv_iq = domish.Element((NS_PRIVILEGE, "privileged_iq"))
+ iq_copy = domish.Element(("jabber:client", "iq"))
+ iq_copy["type"] = iq_type
+ iq_copy["id"] = iq_elt.getAttribute("id", str(uuid.uuid4()))
+ if iq_elt.getAttribute("to"):
+ iq_copy["to"] = iq_elt["to"]
+ iq_copy.addChild(first_child)
+ priv_iq.addChild(iq_copy)
+ wrapper.addChild(priv_iq)
+
+ result = await wrapper.send()
+ return self._unwrap_privileged_iq_result(result)
+
+ def _unwrap_privileged_iq_result(
+ self, iq_result_elt: domish.Element
+ ) -> domish.Element:
+ """Unwrap a privileged IQ response.
+
+ @param iq_result_elt: The IQ result stanza.
+ @return: The unwrapped inner IQ.
+ @raise PrivilegeError: If the response format is invalid.
+ """
+ try:
+ privilege = next(iq_result_elt.elements(NS_PRIVILEGE, "privilege"))
+ forwarded = next(privilege.elements(NS_FORWARD, "forwarded"))
+ return next(forwarded.elements("jabber:client", "iq"))
+ except StopIteration:
+ log.error(f"Invalid privileged IQ response: {iq_result_elt.toXml()}")
+ raise PrivilegeError("Invalid response format")
+
+ def _on_presence(
+ self, presence_elt: domish.Element, client: SatXMPPComponent
+ ) -> None:
+ """Handle incoming presence stanza.
+
+ @param presence_elt: The presence stanza.
+ @param client: The client instance.
+ """
+ try:
+ from_jid = jid.JID(presence_elt["from"])
+ except:
+ log.exception("Can't parse 'from' attribute.")
+ raise
+
+ G.host.trigger.point("XEP-0356_presence_received", client, from_jid, presence_elt)
+
+
+@implementer(itx_xmpp.IDisco)
+class XEP0356Handler(XMPPHandler):
+
+ def __init__(self, plugin_parent: XEP_0356) -> None:
+ self.plugin_parent = plugin_parent
+
+ @property
+ def client(self) -> SatXMPPComponent:
+ """Get the client (component) this handler is attached to."""
+ assert self.parent is not None
+ return cast(SatXMPPComponent, self.parent)
+
+ def connectionInitialized(self) -> None:
+ """Register observers after stream initialization.
+
+ This is called immediately after authentication, ensuring we don't miss
+ privilege advertisements that may be sent by the server right after
+ the component authenticates.
+ """
+ if not self.client.is_component:
+ return
+
+ self.client.xmlstream.addObserver(
+ f"/message/privilege[@xmlns='{NS_PRIVILEGE}']",
+ self.plugin_parent.on_advertise,
+ client=self.client,
+ )
+ self.client.xmlstream.addObserver(
+ "/presence", self.plugin_parent._on_presence, client=self.client
+ )
+ log.debug(f"XEP-0356 observers registered for {self.client.profile}")
+
+ def getDiscoInfo(
+ self,
+ requestor: jid.JID,
+ target: jid.JID,
+ nodeIdentifier: str = "",
+ ) -> list[disco.DiscoFeature | disco.DiscoIdentity]:
+ """Get disco info.
+
+ @param requestor: JID of the requesting entity
+ @param target: JID of the target entity
+ @param nodeIdentifier: Node identifier
+ @return: Disco features
+ """
+ return [disco.DiscoFeature(NS_PRIVILEGE)]
+
+ def getDiscoItems(
+ self,
+ requestor: jid.JID,
+ target: jid.JID,
+ nodeIdentifier: str = "",
+ ) -> list[disco.DiscoItem]:
+ """Get disco items.
+
+ @param requestor: JID of the requesting entity
+ @param target: JID of the target entity
+ @param nodeIdentifier: Node identifier
+ @return: Disco items
+ """
+ return []
diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py
index f30a84ae..5fe73b7d 100644
--- a/tests/unit/conftest.py
+++ b/tests/unit/conftest.py
@@ -20,6 +20,7 @@ from unittest.mock import MagicMock, AsyncMock
from pytest import fixture
from twisted.internet import defer
from twisted.words.protocols.jabber import jid
+from twisted.words.xish import domish
from libervia.backend import G
from tx_xmpp.data_form import NS_X_DATA
from libervia.backend.core.main import LiberviaBackend
@@ -94,6 +95,49 @@ def host(bridge, storage):
return host
+@fixture
+def client_IQ():
+ """Factory fixture to create a mock IQ method for testing.
+
+ Usage:
+ def test_something(client, client_IQ):
+ response_elt = domish.Element((None, "iq"))
+ response_elt["type"] = "result"
+ client.IQ = client_IQ(response_elt=response_elt)
+
+ # Later in test:
+ assert len(client.IQ.sent_iqs) == 1
+
+ @param response: Optional response to return from send(). If None, returns empty IQ.
+ @return: The mock IQ method with attached sent_iqs list.
+ """
+
+ def factory(response_elt=None):
+ sent_iqs = []
+
+ def client_IQ(type_="set", timeout=60):
+ iq_elt = domish.Element((None, "iq"))
+ iq_elt["type"] = type_
+
+ async def mock_send():
+ sent_iqs.append(iq_elt)
+ if response_elt is not None:
+ return response_elt
+ # Return empty IQ by default
+ empty_response = domish.Element((None, "iq"))
+ empty_response["type"] = "result"
+ return empty_response
+
+ iq_elt.send = mock_send
+ return iq_elt
+
+ # Attach list for test assertions
+ client_IQ.sent_iqs = sent_iqs
+ return client_IQ
+
+ return factory
+
+
@fixture
def client():
client = MagicMock()
diff --git a/tests/unit/test_xep_0356.py b/tests/unit/test_xep_0356.py
new file mode 100644
index 00000000..567033a3
--- /dev/null
+++ b/tests/unit/test_xep_0356.py
@@ -0,0 +1,708 @@
+#!/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 .
+
+"""
+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 import disco
+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 = []
+
+ def mock_trigger_point(point_name, *args):
+ trigger_point_calls.append((point_name, args))
+
+ G.host.trigger.point = mock_trigger_point
+
+ 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 = []
+
+ def mock_trigger_point(point_name, *args):
+ trigger_point_calls.append((point_name, args))
+
+ G.host.trigger.point = mock_trigger_point
+
+ 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"