libervia-backend/tests/unit/test_component_pubsub.py

#!/usr/bin/env python3

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

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

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

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

"""
Tests for the PubsubComponent implementation.
"""

from unittest.mock import MagicMock, AsyncMock, patch
from pytest import raises
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


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="not-authorized"):
            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

        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)