1281 lines
46 KiB
Python
1281 lines
46 KiB
Python
#!/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 the PubsubComponent implementation.
|
|
"""
|
|
|
|
from unittest.mock import MagicMock, AsyncMock, patch
|
|
from pytest import raises, fixture
|
|
from pytest_twisted import ensureDeferred as ed
|
|
from twisted.words.protocols.jabber import jid
|
|
from twisted.words.protocols.jabber.error import StanzaError
|
|
from twisted.words.xish import domish
|
|
from tx_xmpp import pubsub
|
|
from tx_xmpp import rsm
|
|
from libervia.backend.memory.sqla_mapping import (
|
|
AccessModel,
|
|
Affiliation,
|
|
PublishModel,
|
|
PubsubNode,
|
|
PubsubAffiliation,
|
|
PubsubSub,
|
|
SubscriptionState,
|
|
)
|
|
from libervia.backend.plugins.plugin_comp_pubsub import PubsubResource
|
|
from libervia.backend.plugins.plugin_xep_0496 import XEP_0496
|
|
|
|
|
|
@fixture(autouse=True)
|
|
def setup_xep_0496(host, client):
|
|
"""Set up XEP-0496 plugin instance for pubsub tests.
|
|
|
|
This autouse fixture creates a real XEP_0496 instance so that the actual
|
|
permission-checking logic is tested rather than mocked behavior.
|
|
It also links the client to the host so PubsubResource can access the plugin.
|
|
"""
|
|
xep_0496 = XEP_0496(host)
|
|
host.plugins["XEP-0496"] = xep_0496
|
|
client.host = host
|
|
return xep_0496
|
|
|
|
|
|
def make_pubsub_request(
|
|
sender: jid.JID,
|
|
nodeIdentifier: str | None = None,
|
|
subscriber: jid.JID | None = None,
|
|
items: list[domish.Element] | None = None,
|
|
itemIdentifiers: list[str] | None = None,
|
|
options: dict | None = None,
|
|
affiliations: dict[jid.JID, str] | None = None,
|
|
subscriptions: list[pubsub.Subscription] | None = None,
|
|
rsm_params: rsm.RSMRequest | None = None,
|
|
maxItems: int | None = None,
|
|
nodeType: str | None = None,
|
|
notify: bool = False,
|
|
) -> pubsub.PubSubRequest:
|
|
"""Helper to create a PubSubRequest for testing."""
|
|
request = pubsub.PubSubRequest()
|
|
request.sender = sender
|
|
request.nodeIdentifier = nodeIdentifier
|
|
request.subscriber = subscriber
|
|
request.items = items
|
|
request.itemIdentifiers = itemIdentifiers
|
|
request.options = options
|
|
request.affiliations = affiliations
|
|
request.subscriptions = subscriptions
|
|
request.rsm = rsm_params
|
|
request.maxItems = maxItems
|
|
request.nodeType = nodeType
|
|
request.notify = notify
|
|
return request
|
|
|
|
|
|
def make_pubsub_node(
|
|
name: str,
|
|
access_model: AccessModel = AccessModel.open,
|
|
publish_model: PublishModel = PublishModel.publishers,
|
|
affiliations: list[PubsubAffiliation] | None = None,
|
|
subscriptions: list[PubsubSub] | None = None,
|
|
) -> PubsubNode:
|
|
"""Helper to create a PubsubNode for testing."""
|
|
node = PubsubNode(name=name)
|
|
node.access_model = access_model
|
|
node.publish_model = publish_model
|
|
node.affiliations = affiliations or []
|
|
node.subscriptions = subscriptions or []
|
|
return node
|
|
|
|
|
|
def make_affiliation(jid_obj: jid.JID, affiliation: Affiliation) -> PubsubAffiliation:
|
|
"""Helper to create a PubsubAffiliation for testing."""
|
|
aff = PubsubAffiliation()
|
|
aff.entity = jid_obj
|
|
aff.affiliation = affiliation
|
|
return aff
|
|
|
|
|
|
def make_subscription(jid_obj: jid.JID, state: str) -> PubsubSub:
|
|
"""Helper to create a PubsubSub for testing."""
|
|
sub = PubsubSub()
|
|
sub.subscriber = jid_obj
|
|
sub.state = getattr(SubscriptionState, state.upper())
|
|
return sub
|
|
|
|
|
|
def make_item_element(item_id: str, publisher: str | None = None) -> domish.Element:
|
|
"""Helper to create an item element for testing."""
|
|
item = domish.Element((pubsub.NS_PUBSUB, "item"))
|
|
item["id"] = item_id
|
|
if publisher:
|
|
item["publisher"] = publisher
|
|
item.addElement(("content", "test"))
|
|
return item
|
|
|
|
|
|
class TestPubsubResourceCreate:
|
|
"""Tests for node creation."""
|
|
|
|
@ed
|
|
async def test_creates_node_with_owner_affiliation(self, host, client):
|
|
"""creating a node sets the creator as owner"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("user@test.example"),
|
|
nodeIdentifier="test_node",
|
|
)
|
|
|
|
host.memory.storage.set_pubsub_node = AsyncMock()
|
|
|
|
await resource._create(request)
|
|
|
|
# Verify that the creator is set as owner
|
|
call_args = host.memory.storage.set_pubsub_node.call_args
|
|
assert call_args is not None
|
|
affiliations = call_args.kwargs.get("affiliations", [])
|
|
assert len(affiliations) == 1
|
|
assert affiliations[0].entity == jid.JID("user@test.example")
|
|
assert affiliations[0].affiliation == Affiliation.owner
|
|
|
|
@ed
|
|
async def test_generates_node_id_if_not_provided(self, host, client):
|
|
"""creating a node without ID generates one"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(sender=jid.JID("user@test.example"))
|
|
|
|
host.memory.storage.set_pubsub_node = AsyncMock()
|
|
|
|
await resource._create(request)
|
|
|
|
# Verify a node ID was generated and used
|
|
call_args = host.memory.storage.set_pubsub_node.call_args
|
|
assert call_args is not None
|
|
node_id = call_args.kwargs.get("name")
|
|
assert node_id is not None
|
|
assert len(node_id) > 0
|
|
# Node ID should start with "generic/" when auto-generated
|
|
assert node_id.startswith("generic/")
|
|
|
|
@ed
|
|
async def test_raises_conflict_if_node_exists(self, host, client):
|
|
"""creating an existing node raises conflict"""
|
|
from libervia.backend.core.exceptions import ConflictError
|
|
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("user@test.example"),
|
|
nodeIdentifier="existing_node",
|
|
)
|
|
|
|
# The ConflictError is generated by storage, so we actually check that the
|
|
# Exception is propagated correctly.
|
|
host.memory.storage.set_pubsub_node = AsyncMock(side_effect=ConflictError())
|
|
|
|
with raises(StanzaError, match="conflict"):
|
|
await resource._create(request)
|
|
|
|
|
|
class TestPubsubResourcePublish:
|
|
"""Tests for item publishing."""
|
|
|
|
@ed
|
|
async def test_rejects_empty_items(self, host, client):
|
|
"""publishing empty items raises bad-request"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("user@test.example"),
|
|
nodeIdentifier="test_node",
|
|
items=[],
|
|
)
|
|
|
|
with raises(StanzaError, match="bad-request"):
|
|
await resource._publish(request)
|
|
|
|
@ed
|
|
async def test_rejects_nonexistent_node(self, host, client):
|
|
"""publishing to nonexistent node raises item_not_found"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("user@test.example"),
|
|
nodeIdentifier="nonexistent_node",
|
|
items=[make_item_element("item1")],
|
|
)
|
|
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=None)
|
|
|
|
with raises(StanzaError, match="item-not-found"):
|
|
await resource._publish(request)
|
|
|
|
@ed
|
|
async def test_publishers_model_requires_publisher_affiliation(self, host, client):
|
|
"""publishers model requires publisher or owner affiliation"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("user@test.example"),
|
|
nodeIdentifier="test_node",
|
|
items=[make_item_element("item1")],
|
|
)
|
|
|
|
node = make_pubsub_node(
|
|
name="test_node",
|
|
publish_model=PublishModel.publishers,
|
|
affiliations=[
|
|
make_affiliation(jid.JID("user@test.example"), Affiliation.member)
|
|
],
|
|
)
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
|
|
|
|
with raises(StanzaError, match="forbidden"):
|
|
await resource._publish(request)
|
|
|
|
@ed
|
|
async def test_publishers_model_allows_owner(self, host, client):
|
|
"""publishers model allows owners to publish"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("owner@test.example"),
|
|
nodeIdentifier="test_node",
|
|
items=[make_item_element("item1")],
|
|
)
|
|
|
|
node = make_pubsub_node(
|
|
name="test_node",
|
|
publish_model=PublishModel.publishers,
|
|
affiliations=[
|
|
make_affiliation(jid.JID("owner@test.example"), Affiliation.owner)
|
|
],
|
|
)
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
|
|
host.memory.storage.cache_pubsub_items = AsyncMock()
|
|
|
|
await resource._publish(request)
|
|
|
|
host.memory.storage.cache_pubsub_items.assert_called_once()
|
|
|
|
@ed
|
|
async def test_publishers_model_allows_publisher(self, host, client):
|
|
"""publishers model allows publishers to publish"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("publisher@test.example"),
|
|
nodeIdentifier="test_node",
|
|
items=[make_item_element("item1")],
|
|
)
|
|
|
|
node = make_pubsub_node(
|
|
name="test_node",
|
|
publish_model=PublishModel.publishers,
|
|
affiliations=[
|
|
make_affiliation(jid.JID("publisher@test.example"), Affiliation.publisher)
|
|
],
|
|
)
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
|
|
host.memory.storage.cache_pubsub_items = AsyncMock()
|
|
|
|
await resource._publish(request)
|
|
|
|
host.memory.storage.cache_pubsub_items.assert_called_once()
|
|
|
|
@ed
|
|
async def test_subscribers_model_requires_subscription(self, host, client):
|
|
"""subscribers model requires subscription to publish"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("user@test.example"),
|
|
nodeIdentifier="test_node",
|
|
items=[make_item_element("item1")],
|
|
)
|
|
|
|
node = make_pubsub_node(
|
|
name="test_node",
|
|
publish_model=PublishModel.subscribers,
|
|
affiliations=[
|
|
make_affiliation(jid.JID("user@test.example"), Affiliation.member)
|
|
],
|
|
)
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
|
|
|
|
with raises(StanzaError, match="forbidden"):
|
|
await resource._publish(request)
|
|
|
|
@ed
|
|
async def test_subscribers_model_allows_subscribed_member(self, host, client):
|
|
"""subscribers model allows subscribed members to publish"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("user@test.example"),
|
|
nodeIdentifier="test_node",
|
|
items=[make_item_element("item1")],
|
|
)
|
|
|
|
node = make_pubsub_node(
|
|
name="test_node",
|
|
publish_model=PublishModel.subscribers,
|
|
affiliations=[
|
|
make_affiliation(jid.JID("user@test.example"), Affiliation.member)
|
|
],
|
|
subscriptions=[make_subscription(jid.JID("user@test.example"), "subscribed")],
|
|
)
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
|
|
host.memory.storage.cache_pubsub_items = AsyncMock()
|
|
|
|
await resource._publish(request)
|
|
|
|
host.memory.storage.cache_pubsub_items.assert_called_once()
|
|
|
|
@ed
|
|
async def test_generates_item_id_if_missing(self, host, client):
|
|
"""publishing an item without ID generates one"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
item = make_item_element("") # Empty ID
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("owner@test.example"),
|
|
nodeIdentifier="test_node",
|
|
items=[item],
|
|
)
|
|
|
|
node = make_pubsub_node(
|
|
name="test_node",
|
|
affiliations=[
|
|
make_affiliation(jid.JID("owner@test.example"), Affiliation.owner)
|
|
],
|
|
)
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
|
|
host.memory.storage.cache_pubsub_items = AsyncMock()
|
|
|
|
await resource._publish(request)
|
|
|
|
# Verify item was processed and has an ID generated
|
|
call_args = host.memory.storage.cache_pubsub_items.call_args
|
|
assert call_args is not None
|
|
# cache_pubsub_items is called with (client, node, items_data)
|
|
items_data = call_args.args[2]
|
|
assert len(items_data) == 1
|
|
assert items_data[0].getAttribute("id") is not None
|
|
assert len(items_data[0].getAttribute("id")) > 0
|
|
|
|
@ed
|
|
async def test_sets_publisher_attribute(self, host, client):
|
|
"""publishing sets the publisher attribute on items"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
item = make_item_element("item1")
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("user@test.example/resource"),
|
|
nodeIdentifier="test_node",
|
|
items=[item],
|
|
)
|
|
|
|
node = make_pubsub_node(
|
|
name="test_node",
|
|
affiliations=[
|
|
make_affiliation(jid.JID("user@test.example"), Affiliation.owner)
|
|
],
|
|
)
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
|
|
host.memory.storage.cache_pubsub_items = AsyncMock()
|
|
|
|
await resource._publish(request)
|
|
|
|
# Verify the item has publisher attribute set
|
|
call_args = host.memory.storage.cache_pubsub_items.call_args
|
|
assert call_args is not None
|
|
# cache_pubsub_items is called with (client, node, items_data)
|
|
items_data = call_args.args[2]
|
|
assert len(items_data) == 1
|
|
published_item = items_data[0]
|
|
assert published_item.getAttribute("publisher") == "user@test.example/resource"
|
|
|
|
|
|
class TestPubsubResourceSubscribe:
|
|
"""Tests for subscription management."""
|
|
|
|
@ed
|
|
async def test_rejects_subscribing_for_others(self, host, client):
|
|
"""subscribing for another user raises forbidden"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("user@test.example"),
|
|
nodeIdentifier="test_node",
|
|
subscriber=jid.JID("other@test.example"),
|
|
)
|
|
|
|
with raises(StanzaError, match="forbidden"):
|
|
await resource._subscribe(request)
|
|
|
|
@ed
|
|
async def test_rejects_nonexistent_node(self, host, client):
|
|
"""subscribing to nonexistent node raises item_not_found"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("user@test.example"),
|
|
nodeIdentifier="nonexistent_node",
|
|
subscriber=jid.JID("user@test.example"),
|
|
)
|
|
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=None)
|
|
|
|
with raises(StanzaError, match="item-not-found"):
|
|
await resource._subscribe(request)
|
|
|
|
@ed
|
|
async def test_rejects_outcast_users(self, host, client):
|
|
"""outcast users cannot subscribe"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("outcast@test.example"),
|
|
nodeIdentifier="test_node",
|
|
subscriber=jid.JID("outcast@test.example"),
|
|
)
|
|
|
|
node = make_pubsub_node(
|
|
name="test_node",
|
|
affiliations=[
|
|
make_affiliation(jid.JID("outcast@test.example"), Affiliation.outcast)
|
|
],
|
|
)
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
|
|
|
|
with raises(StanzaError, match="forbidden"):
|
|
await resource._subscribe(request)
|
|
|
|
@ed
|
|
async def test_open_model_allows_anyone(self, host, client):
|
|
"""open access model allows anyone to subscribe"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("user@test.example"),
|
|
nodeIdentifier="test_node",
|
|
subscriber=jid.JID("user@test.example"),
|
|
)
|
|
|
|
node = make_pubsub_node(
|
|
name="test_node",
|
|
access_model=AccessModel.open,
|
|
)
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
|
|
host.memory.storage.set_pubsub_subscriptions = AsyncMock()
|
|
|
|
await resource._subscribe(request)
|
|
|
|
host.memory.storage.set_pubsub_subscriptions.assert_called_once()
|
|
call_args = host.memory.storage.set_pubsub_subscriptions.call_args
|
|
subscriptions_dict = call_args.args[1]
|
|
assert (
|
|
subscriptions_dict[jid.JID("user@test.example")]
|
|
== SubscriptionState.SUBSCRIBED
|
|
)
|
|
|
|
@ed
|
|
async def test_whitelist_model_requires_membership(self, host, client):
|
|
"""whitelist access model requires member affiliation"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("user@test.example"),
|
|
nodeIdentifier="test_node",
|
|
subscriber=jid.JID("user@test.example"),
|
|
)
|
|
|
|
node = make_pubsub_node(
|
|
name="test_node",
|
|
access_model=AccessModel.whitelist,
|
|
affiliations=[], # No affiliation for this user
|
|
)
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
|
|
|
|
with raises(StanzaError, match="forbidden"):
|
|
await resource._subscribe(request)
|
|
|
|
@ed
|
|
async def test_whitelist_model_allows_member(self, host, client):
|
|
"""whitelist access model allows members to subscribe"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("member@test.example"),
|
|
nodeIdentifier="test_node",
|
|
subscriber=jid.JID("member@test.example"),
|
|
)
|
|
|
|
node = make_pubsub_node(
|
|
name="test_node",
|
|
access_model=AccessModel.whitelist,
|
|
affiliations=[
|
|
make_affiliation(jid.JID("member@test.example"), Affiliation.member)
|
|
],
|
|
)
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
|
|
host.memory.storage.set_pubsub_subscriptions = AsyncMock()
|
|
|
|
await resource._subscribe(request)
|
|
|
|
host.memory.storage.set_pubsub_subscriptions.assert_called_once()
|
|
call_args = host.memory.storage.set_pubsub_subscriptions.call_args
|
|
subscriptions_dict = call_args.args[1]
|
|
assert (
|
|
subscriptions_dict[jid.JID("member@test.example")]
|
|
== SubscriptionState.SUBSCRIBED
|
|
)
|
|
|
|
@ed
|
|
async def test_rejects_duplicate_pending(self, host, client):
|
|
"""subscribing when already pending raises not-authorized"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("user@test.example"),
|
|
nodeIdentifier="test_node",
|
|
subscriber=jid.JID("user@test.example"),
|
|
)
|
|
|
|
node = make_pubsub_node(
|
|
name="test_node",
|
|
access_model=AccessModel.open,
|
|
)
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
|
|
pending_sub = make_subscription(jid.JID("user@test.example"), "pending")
|
|
host.memory.storage.get_pubsub_subscriptions = AsyncMock(
|
|
return_value=[pending_sub]
|
|
)
|
|
|
|
with raises(StanzaError, match="not-authorized"):
|
|
await resource._subscribe(request)
|
|
|
|
|
|
class TestPubsubResourceUnsubscribe:
|
|
"""Tests for unsubscription."""
|
|
|
|
@ed
|
|
async def test_rejects_unsubscribing_for_others(self, host, client):
|
|
"""unsubscribing for another user raises forbidden"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("user@test.example"),
|
|
nodeIdentifier="test_node",
|
|
subscriber=jid.JID("other@test.example"),
|
|
)
|
|
|
|
with raises(StanzaError, match="forbidden"):
|
|
await resource._unsubscribe(request)
|
|
|
|
@ed
|
|
async def test_rejects_nonexistent_node(self, host, client):
|
|
"""unsubscribing from nonexistent node raises item_not_found"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("user@test.example"),
|
|
nodeIdentifier="nonexistent_node",
|
|
subscriber=jid.JID("user@test.example"),
|
|
)
|
|
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=None)
|
|
|
|
with raises(StanzaError, match="item-not-found"):
|
|
await resource._unsubscribe(request)
|
|
|
|
@ed
|
|
async def test_removes_subscription(self, host, client):
|
|
"""unsubscribing removes the subscription"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("user@test.example"),
|
|
nodeIdentifier="test_node",
|
|
subscriber=jid.JID("user@test.example"),
|
|
)
|
|
|
|
node = make_pubsub_node(
|
|
name="test_node",
|
|
subscriptions=[make_subscription(jid.JID("user@test.example"), "subscribed")],
|
|
)
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
|
|
host.memory.storage.set_pubsub_subscriptions = AsyncMock()
|
|
|
|
await resource._unsubscribe(request)
|
|
|
|
# Verify that the subscription was set to None (removed)
|
|
host.memory.storage.set_pubsub_subscriptions.assert_called_once()
|
|
call_args = host.memory.storage.set_pubsub_subscriptions.call_args
|
|
assert call_args is not None
|
|
subscriptions_dict = call_args.args[1]
|
|
assert jid.JID("user@test.example") in subscriptions_dict
|
|
assert subscriptions_dict[jid.JID("user@test.example")] is None
|
|
|
|
|
|
class TestPubsubResourceItems:
|
|
"""Tests for item retrieval."""
|
|
|
|
@ed
|
|
async def test_rejects_nonexistent_node(self, host, client):
|
|
"""retrieving items from nonexistent node raises item_not_found"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("user@test.example"),
|
|
nodeIdentifier="nonexistent_node",
|
|
)
|
|
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=None)
|
|
|
|
with raises(StanzaError, match="item-not-found"):
|
|
await resource._items(request)
|
|
|
|
@ed
|
|
async def test_open_model_allows_anyone(self, host, client):
|
|
"""open access model allows anyone to retrieve items"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("user@test.example"),
|
|
nodeIdentifier="test_node",
|
|
)
|
|
|
|
node = make_pubsub_node(
|
|
name="test_node",
|
|
access_model=AccessModel.open,
|
|
)
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
|
|
host.memory.storage.get_items = AsyncMock(return_value=([], {}))
|
|
|
|
await resource._items(request)
|
|
|
|
host.memory.storage.get_items.assert_called_once()
|
|
|
|
@ed
|
|
async def test_whitelist_model_rejects_non_members(self, host, client):
|
|
"""whitelist access model rejects non-members"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("user@test.example"),
|
|
nodeIdentifier="test_node",
|
|
)
|
|
|
|
node = make_pubsub_node(
|
|
name="test_node",
|
|
access_model=AccessModel.whitelist,
|
|
affiliations=[], # No affiliation for this user
|
|
)
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
|
|
|
|
with raises(StanzaError, match="forbidden"):
|
|
await resource._items(request)
|
|
|
|
@ed
|
|
async def test_whitelist_model_allows_member(self, host, client):
|
|
"""whitelist access model allows members to retrieve items"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("member@test.example"),
|
|
nodeIdentifier="test_node",
|
|
)
|
|
|
|
node = make_pubsub_node(
|
|
name="test_node",
|
|
access_model=AccessModel.whitelist,
|
|
affiliations=[
|
|
make_affiliation(jid.JID("member@test.example"), Affiliation.member)
|
|
],
|
|
)
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
|
|
host.memory.storage.get_items = AsyncMock(return_value=([], {}))
|
|
|
|
await resource._items(request)
|
|
|
|
host.memory.storage.get_items.assert_called_once()
|
|
|
|
|
|
class TestPubsubResourceRetract:
|
|
"""Tests for item retraction."""
|
|
|
|
@ed
|
|
async def test_rejects_empty_identifiers(self, host, client):
|
|
"""retracting without item identifiers raises bad-request"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("user@test.example"),
|
|
nodeIdentifier="test_node",
|
|
itemIdentifiers=[],
|
|
)
|
|
|
|
with raises(StanzaError, match="bad-request"):
|
|
await resource._retract(request)
|
|
|
|
@ed
|
|
async def test_rejects_nonexistent_node(self, host, client):
|
|
"""retracting from nonexistent node raises item_not_found"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("user@test.example"),
|
|
nodeIdentifier="nonexistent_node",
|
|
itemIdentifiers=["item1"],
|
|
)
|
|
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=None)
|
|
|
|
with raises(StanzaError, match="item-not-found"):
|
|
await resource._retract(request)
|
|
|
|
@ed
|
|
async def test_owner_can_retract_any_item(self, host, client):
|
|
"""owners can retract any item from their node"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("owner@test.example"),
|
|
nodeIdentifier="test_node",
|
|
itemIdentifiers=["item1"],
|
|
)
|
|
|
|
node = make_pubsub_node(
|
|
name="test_node",
|
|
affiliations=[
|
|
make_affiliation(jid.JID("owner@test.example"), Affiliation.owner)
|
|
],
|
|
)
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
|
|
|
|
# Create an item published by someone else
|
|
item_elt = make_item_element("item1")
|
|
item_elt["publisher"] = "other@test.example"
|
|
item_wrapper = MagicMock()
|
|
item_wrapper.data = item_elt
|
|
item_wrapper.name = "item1"
|
|
host.memory.storage.get_items = AsyncMock(return_value=([item_wrapper], {}))
|
|
host.memory.storage.delete_pubsub_items = AsyncMock()
|
|
|
|
with patch.object(resource, "_send_notifications", new=AsyncMock()):
|
|
await resource._retract(request)
|
|
|
|
host.memory.storage.delete_pubsub_items.assert_called_once()
|
|
call_args = host.memory.storage.delete_pubsub_items.call_args
|
|
assert call_args.args[0] == node
|
|
assert call_args.args[1] == ["item1"]
|
|
|
|
@ed
|
|
async def test_publisher_can_retract_any_item(self, host, client):
|
|
"""publishers can retract any item from their node"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("publisher@test.example"),
|
|
nodeIdentifier="test_node",
|
|
itemIdentifiers=["item1"],
|
|
)
|
|
|
|
node = make_pubsub_node(
|
|
name="test_node",
|
|
affiliations=[
|
|
make_affiliation(jid.JID("publisher@test.example"), Affiliation.publisher)
|
|
],
|
|
)
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
|
|
|
|
item_elt = make_item_element("item1")
|
|
item_elt["publisher"] = "other@test.example"
|
|
item_wrapper = MagicMock()
|
|
item_wrapper.data = item_elt
|
|
item_wrapper.name = "item1"
|
|
host.memory.storage.get_items = AsyncMock(return_value=([item_wrapper], {}))
|
|
host.memory.storage.delete_pubsub_items = AsyncMock()
|
|
|
|
with patch.object(resource, "_send_notifications", new=AsyncMock()):
|
|
await resource._retract(request)
|
|
|
|
host.memory.storage.delete_pubsub_items.assert_called_once()
|
|
call_args = host.memory.storage.delete_pubsub_items.call_args
|
|
assert call_args.args[0] == node
|
|
assert call_args.args[1] == ["item1"]
|
|
|
|
@ed
|
|
async def test_user_can_only_retract_own_items(self, host, client):
|
|
"""non-owners can only retract their own items"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("user@test.example"),
|
|
nodeIdentifier="test_node",
|
|
itemIdentifiers=["item1"],
|
|
)
|
|
|
|
node = make_pubsub_node(
|
|
name="test_node",
|
|
affiliations=[
|
|
make_affiliation(jid.JID("user@test.example"), Affiliation.member)
|
|
],
|
|
)
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
|
|
host.memory.is_admin_jid.return_value = False
|
|
|
|
# Item published by someone else
|
|
item_elt = make_item_element("item1")
|
|
item_elt["publisher"] = "other@test.example"
|
|
item_wrapper = MagicMock()
|
|
item_wrapper.data = item_elt
|
|
item_wrapper.name = "item1"
|
|
host.memory.storage.get_items = AsyncMock(return_value=([item_wrapper], {}))
|
|
|
|
with raises(StanzaError, match="forbidden"):
|
|
await resource._retract(request)
|
|
|
|
@ed
|
|
async def test_user_can_retract_own_items(self, host, client):
|
|
"""users can retract their own items"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("user@test.example"),
|
|
nodeIdentifier="test_node",
|
|
itemIdentifiers=["item1"],
|
|
)
|
|
|
|
node = make_pubsub_node(
|
|
name="test_node",
|
|
affiliations=[
|
|
make_affiliation(jid.JID("user@test.example"), Affiliation.publisher)
|
|
],
|
|
)
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
|
|
|
|
# Item published by the user
|
|
item_elt = make_item_element("item1")
|
|
item_elt["publisher"] = "user@test.example/resource"
|
|
item_wrapper = MagicMock()
|
|
item_wrapper.data = item_elt
|
|
item_wrapper.name = "item1"
|
|
host.memory.storage.get_items = AsyncMock(return_value=([item_wrapper], {}))
|
|
host.memory.storage.delete_pubsub_items = AsyncMock()
|
|
|
|
with patch.object(resource, "_send_notifications", new=AsyncMock()):
|
|
await resource._retract(request)
|
|
|
|
host.memory.storage.delete_pubsub_items.assert_called_once()
|
|
call_args = host.memory.storage.delete_pubsub_items.call_args
|
|
assert call_args.args[0] == node
|
|
assert call_args.args[1] == ["item1"]
|
|
|
|
|
|
class TestPubsubResourcePurge:
|
|
"""Tests for node purging."""
|
|
|
|
@ed
|
|
async def test_rejects_nonexistent_node(self, host, client):
|
|
"""purging nonexistent node raises item_not_found"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("user@test.example"),
|
|
nodeIdentifier="nonexistent_node",
|
|
)
|
|
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=None)
|
|
|
|
with raises(StanzaError, match="item-not-found"):
|
|
await resource._purge(request)
|
|
|
|
@ed
|
|
async def test_only_owners_can_purge(self, host, client):
|
|
"""only owners can purge a node"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("user@test.example"),
|
|
nodeIdentifier="test_node",
|
|
)
|
|
|
|
node = make_pubsub_node(
|
|
name="test_node",
|
|
affiliations=[
|
|
make_affiliation(jid.JID("user@test.example"), Affiliation.publisher)
|
|
],
|
|
)
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
|
|
host.memory.is_admin_jid.return_value = False
|
|
|
|
with raises(StanzaError, match="forbidden"):
|
|
await resource._purge(request)
|
|
|
|
@ed
|
|
async def test_owner_can_purge(self, host, client):
|
|
"""owners can purge their node"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("owner@test.example"),
|
|
nodeIdentifier="test_node",
|
|
)
|
|
|
|
node = make_pubsub_node(
|
|
name="test_node",
|
|
affiliations=[
|
|
make_affiliation(jid.JID("owner@test.example"), Affiliation.owner)
|
|
],
|
|
)
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
|
|
host.memory.storage.delete_pubsub_items = AsyncMock()
|
|
host.memory.is_admin_jid.return_value = False
|
|
|
|
with patch.object(resource, "_send_notifications", new=AsyncMock()):
|
|
await resource._purge(request)
|
|
|
|
host.memory.storage.delete_pubsub_items.assert_called_once()
|
|
call_args = host.memory.storage.delete_pubsub_items.call_args
|
|
assert call_args.args[0] == node
|
|
assert call_args.kwargs.get("items_names") is None
|
|
|
|
|
|
class TestPubsubResourceDelete:
|
|
"""Tests for node deletion."""
|
|
|
|
@ed
|
|
async def test_rejects_nonexistent_node(self, host, client):
|
|
"""deleting nonexistent node raises item_not_found"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("user@test.example"),
|
|
nodeIdentifier="nonexistent_node",
|
|
)
|
|
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=None)
|
|
|
|
with raises(StanzaError, match="item-not-found"):
|
|
await resource._delete(request)
|
|
|
|
@ed
|
|
async def test_only_owners_can_delete(self, host, client):
|
|
"""only owners can delete a node"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("user@test.example"),
|
|
nodeIdentifier="test_node",
|
|
)
|
|
|
|
node = make_pubsub_node(
|
|
name="test_node",
|
|
affiliations=[
|
|
make_affiliation(jid.JID("user@test.example"), Affiliation.publisher)
|
|
],
|
|
)
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
|
|
host.memory.is_admin_jid.return_value = False
|
|
|
|
with raises(StanzaError, match="forbidden"):
|
|
await resource._delete(request)
|
|
|
|
@ed
|
|
async def test_owner_can_delete(self, host, client):
|
|
"""owners can delete their node"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("owner@test.example"),
|
|
nodeIdentifier="test_node",
|
|
)
|
|
|
|
node = make_pubsub_node(
|
|
name="test_node",
|
|
affiliations=[
|
|
make_affiliation(jid.JID("owner@test.example"), Affiliation.owner)
|
|
],
|
|
)
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
|
|
host.memory.storage.delete_pubsub_node = AsyncMock()
|
|
host.memory.is_admin_jid.return_value = False
|
|
# Mock methods that would fail with test node (no id, storage.session() issues)
|
|
resource.xep_0496.get_linking_nodes_by_id = AsyncMock(return_value=[])
|
|
resource.xep_0496.get_descendant_nodes = AsyncMock(return_value=[])
|
|
|
|
with patch.object(resource, "_send_notifications", new=AsyncMock()):
|
|
await resource._delete(request)
|
|
|
|
host.memory.storage.delete_pubsub_node.assert_called_once()
|
|
|
|
|
|
class TestPubsubResourceConfigure:
|
|
"""Tests for node configuration."""
|
|
|
|
@ed
|
|
async def test_only_owners_can_get_config(self, host, client):
|
|
"""only owners can get node configuration"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("user@test.example"),
|
|
nodeIdentifier="test_node",
|
|
)
|
|
|
|
node = make_pubsub_node(
|
|
name="test_node",
|
|
affiliations=[
|
|
make_affiliation(jid.JID("user@test.example"), Affiliation.member)
|
|
],
|
|
)
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
|
|
host.memory.is_admin_jid.return_value = False
|
|
|
|
with raises(StanzaError, match="forbidden"):
|
|
await resource._configureGet(request)
|
|
|
|
@ed
|
|
async def test_owner_can_get_config(self, host, client):
|
|
"""owners can get node configuration"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("owner@test.example"),
|
|
nodeIdentifier="test_node",
|
|
)
|
|
|
|
node = make_pubsub_node(
|
|
name="test_node",
|
|
access_model=AccessModel.whitelist,
|
|
publish_model=PublishModel.subscribers,
|
|
affiliations=[
|
|
make_affiliation(jid.JID("owner@test.example"), Affiliation.owner)
|
|
],
|
|
)
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
|
|
host.memory.is_admin_jid.return_value = False
|
|
|
|
result = await resource._configureGet(request)
|
|
|
|
assert result["pubsub#access_model"] == "whitelist"
|
|
assert result["pubsub#publish_model"] == "subscribers"
|
|
|
|
@ed
|
|
async def test_only_owners_can_set_config(self, host, client):
|
|
"""only owners can set node configuration"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("user@test.example"),
|
|
nodeIdentifier="test_node",
|
|
options={"pubsub#access_model": "whitelist"},
|
|
)
|
|
|
|
node = make_pubsub_node(
|
|
name="test_node",
|
|
affiliations=[
|
|
make_affiliation(jid.JID("user@test.example"), Affiliation.member)
|
|
],
|
|
)
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
|
|
host.memory.is_admin_jid.return_value = False
|
|
|
|
with raises(StanzaError, match="forbidden"):
|
|
await resource._configureSet(request)
|
|
|
|
@ed
|
|
async def test_owner_can_set_config(self, host, client):
|
|
"""owners can set node configuration"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("owner@test.example"),
|
|
nodeIdentifier="test_node",
|
|
options={"pubsub#access_model": "whitelist"},
|
|
)
|
|
|
|
node = make_pubsub_node(
|
|
name="test_node",
|
|
affiliations=[
|
|
make_affiliation(jid.JID("owner@test.example"), Affiliation.owner)
|
|
],
|
|
)
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
|
|
host.memory.storage.update_pubsub_node = AsyncMock()
|
|
host.memory.is_admin_jid.return_value = False
|
|
|
|
await resource._configureSet(request)
|
|
|
|
# Verify that the configuration was updated with the correct values
|
|
host.memory.storage.update_pubsub_node.assert_called_once()
|
|
call_args = host.memory.storage.update_pubsub_node.call_args
|
|
assert call_args is not None
|
|
# update_pubsub_node is called with (node, update_values)
|
|
update_values = call_args.args[1]
|
|
assert "access_model" in update_values
|
|
assert update_values["access_model"] == AccessModel.whitelist
|
|
|
|
|
|
class TestPubsubResourceAffiliations:
|
|
"""Tests for affiliation management."""
|
|
|
|
@ed
|
|
async def test_only_owners_can_get_affiliations(self, host, client):
|
|
"""only owners can retrieve all affiliations"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("user@test.example"),
|
|
nodeIdentifier="test_node",
|
|
)
|
|
|
|
node = make_pubsub_node(
|
|
name="test_node",
|
|
affiliations=[
|
|
make_affiliation(jid.JID("user@test.example"), Affiliation.member)
|
|
],
|
|
)
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
|
|
host.memory.is_admin_jid.return_value = False
|
|
|
|
with raises(StanzaError, match="forbidden"):
|
|
await resource._affiliationsGet(request)
|
|
|
|
@ed
|
|
async def test_only_owners_can_set_affiliations(self, host, client):
|
|
"""only owners can set affiliations"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("user@test.example"),
|
|
nodeIdentifier="test_node",
|
|
affiliations={jid.JID("other@test.example"): "member"},
|
|
)
|
|
|
|
node = make_pubsub_node(
|
|
name="test_node",
|
|
affiliations=[
|
|
make_affiliation(jid.JID("user@test.example"), Affiliation.member)
|
|
],
|
|
)
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
|
|
host.memory.is_admin_jid.return_value = False
|
|
|
|
with raises(StanzaError, match="forbidden"):
|
|
await resource._affiliationsSet(request)
|
|
|
|
@ed
|
|
async def test_owner_cannot_demote_self(self, host, client):
|
|
"""owners cannot change their own affiliation to non-owner"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("owner@test.example"),
|
|
nodeIdentifier="test_node",
|
|
affiliations={jid.JID("owner@test.example"): "publisher"},
|
|
)
|
|
|
|
node = make_pubsub_node(
|
|
name="test_node",
|
|
affiliations=[
|
|
make_affiliation(jid.JID("owner@test.example"), Affiliation.owner)
|
|
],
|
|
)
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
|
|
host.memory.is_admin_jid.return_value = False
|
|
|
|
with raises(StanzaError, match="forbidden"):
|
|
await resource._affiliationsSet(request)
|
|
|
|
|
|
class TestPubsubResourceSubscriptionsGetSet:
|
|
"""Tests for subscription management by owners."""
|
|
|
|
@ed
|
|
async def test_only_owners_can_get_subscriptions(self, host, client):
|
|
"""only owners can retrieve all subscriptions"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("user@test.example"),
|
|
nodeIdentifier="test_node",
|
|
)
|
|
|
|
node = make_pubsub_node(
|
|
name="test_node",
|
|
affiliations=[
|
|
make_affiliation(jid.JID("user@test.example"), Affiliation.member)
|
|
],
|
|
)
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
|
|
host.memory.is_admin_jid.return_value = False
|
|
|
|
with raises(StanzaError, match="forbidden"):
|
|
await resource._subscriptionsGet(request)
|
|
|
|
@ed
|
|
async def test_only_owners_can_set_subscriptions(self, host, client):
|
|
"""only owners can set subscriptions"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("user@test.example"),
|
|
nodeIdentifier="test_node",
|
|
subscriptions=[
|
|
pubsub.Subscription(
|
|
nodeIdentifier="test_node",
|
|
subscriber=jid.JID("other@test.example"),
|
|
state="subscribed",
|
|
)
|
|
],
|
|
)
|
|
|
|
node = make_pubsub_node(
|
|
name="test_node",
|
|
affiliations=[
|
|
make_affiliation(jid.JID("user@test.example"), Affiliation.member)
|
|
],
|
|
)
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
|
|
host.memory.is_admin_jid.return_value = False
|
|
|
|
with raises(StanzaError, match="forbidden"):
|
|
await resource._subscriptionsSet(request)
|
|
|
|
@ed
|
|
async def test_owner_can_set_subscriptions(self, host, client):
|
|
"""owners can set subscriptions for entities"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("owner@test.example"),
|
|
nodeIdentifier="test_node",
|
|
subscriptions=[
|
|
pubsub.Subscription(
|
|
nodeIdentifier="test_node",
|
|
subscriber=jid.JID("user@test.example"),
|
|
state="subscribed",
|
|
)
|
|
],
|
|
)
|
|
|
|
node = make_pubsub_node(
|
|
name="test_node",
|
|
affiliations=[
|
|
make_affiliation(jid.JID("owner@test.example"), Affiliation.owner),
|
|
],
|
|
)
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
|
|
host.memory.storage.set_pubsub_subscriptions = AsyncMock()
|
|
host.memory.is_admin_jid.return_value = False
|
|
|
|
await resource._subscriptionsSet(request)
|
|
|
|
host.memory.storage.set_pubsub_subscriptions.assert_called_once()
|
|
call_args = host.memory.storage.set_pubsub_subscriptions.call_args
|
|
subscriptions_dict = call_args.args[1]
|
|
assert (
|
|
subscriptions_dict[jid.JID("user@test.example")]
|
|
== SubscriptionState.SUBSCRIBED
|
|
)
|
|
|
|
@ed
|
|
async def test_cannot_subscribe_outcast(self, host, client):
|
|
"""owners cannot subscribe outcast entities"""
|
|
resource = PubsubResource(MagicMock(client=client))
|
|
request = make_pubsub_request(
|
|
sender=jid.JID("owner@test.example"),
|
|
nodeIdentifier="test_node",
|
|
subscriptions=[
|
|
pubsub.Subscription(
|
|
nodeIdentifier="test_node",
|
|
subscriber=jid.JID("outcast@test.example"),
|
|
state="subscribed",
|
|
)
|
|
],
|
|
)
|
|
|
|
node = make_pubsub_node(
|
|
name="test_node",
|
|
affiliations=[
|
|
make_affiliation(jid.JID("owner@test.example"), Affiliation.owner),
|
|
make_affiliation(jid.JID("outcast@test.example"), Affiliation.outcast),
|
|
],
|
|
)
|
|
host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
|
|
host.memory.is_admin_jid.return_value = False
|
|
|
|
with raises(StanzaError, match="not-acceptable"):
|
|
await resource._subscriptionsSet(request)
|