plugin XEP-0356: Privileged Entity implementation:
This is the second XEP needed to implement PEP in Pubsub Component. This implementation should be more or less complete. rel 476
This commit is contained in:
Goffi
parent
ba9a2fc6d5
commit
6605e523d0
3 changed files with 1333 additions and 0 deletions
|
|
@ -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()
|
||||
|
|
|
|||
708
tests/unit/test_xep_0356.py
Normal file
708
tests/unit/test_xep_0356.py
Normal file
|
|
@ -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 <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 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"
|
||||
Loading…
Reference in a new issue