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, 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 tx_xmpp.xmppim import RosterItem
from libervia.backend.memory.sqla_mapping import (
    AccessModel,
    Affiliation,
    PublishModel,
    PubsubItem,
    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 prepare_for_tests(host, client):
    """Set up important data for testing the pubsub component.

    This 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, and mark
    it as a component to reflect real workflow.
    """
    xep_0496 = XEP_0496(host)
    host.plugins["XEP-0496"] = xep_0496
    client.is_component = True
    client.host = host
    return xep_0496


@fixture
def pubsub_resource(client):
    """Create a PubsubResource with proper mocking.

    This fixture creates a PubsubResource with a mock service that has
    xep_0355.is_pep configured to return False by default (since
    pubsub.test.example is not a PEP service).
    """
    mock_service = MagicMock(client=client)
    mock_service.component.xep_0355.is_pep.return_value = False
    return PubsubResource(mock_service)


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,
    recipient: jid.JID | None = None,
) -> 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
    request.recipient = recipient
    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,
    pep: bool = False,
) -> 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 []
    node.pep = pep
    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, pubsub_resource):
        """creating a node sets the creator as owner"""
        request = make_pubsub_request(
            sender=jid.JID("user@test.example"),
            nodeIdentifier="test_node",
            recipient=pubsub_resource.service.client.pubsub_service,
        )

        host.memory.storage.set_pubsub_node = AsyncMock()

        await pubsub_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, pubsub_resource):
        """creating a node without ID generates one"""
        request = make_pubsub_request(
            sender=jid.JID("user@test.example"),
            recipient=pubsub_resource.service.client.pubsub_service,
        )

        host.memory.storage.set_pubsub_node = AsyncMock()

        await pubsub_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, pubsub_resource):
        """creating an existing node raises conflict"""
        from libervia.backend.core.exceptions import ConflictError

        request = make_pubsub_request(
            sender=jid.JID("user@test.example"),
            nodeIdentifier="existing_node",
            recipient=pubsub_resource.service.client.pubsub_service,
        )

        # 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 pubsub_resource._create(request)


class TestPubsubResourcePublish:
    """Tests for item publishing."""

    @ed
    async def test_rejects_empty_items(self, host, client, pubsub_resource):
        """publishing empty items raises bad-request"""
        resource = pubsub_resource
        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, pubsub_resource):
        """publishing to nonexistent node raises item_not_found"""
        resource = pubsub_resource
        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, pubsub_resource
    ):
        """publishers model requires publisher or owner affiliation"""
        resource = pubsub_resource
        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, pubsub_resource):
        """publishers model allows owners to publish"""
        resource = pubsub_resource
        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, pubsub_resource):
        """publishers model allows publishers to publish"""
        resource = pubsub_resource
        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, pubsub_resource
    ):
        """subscribers model requires subscription to publish"""
        resource = pubsub_resource
        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, pubsub_resource
    ):
        """subscribers model allows subscribed members to publish"""
        resource = pubsub_resource
        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, pubsub_resource):
        """publishing an item without ID generates one"""
        resource = pubsub_resource
        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, pubsub_resource):
        """publishing sets the publisher attribute on items"""
        resource = pubsub_resource
        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, pubsub_resource):
        """subscribing for another user raises forbidden"""
        resource = pubsub_resource
        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, pubsub_resource):
        """subscribing to nonexistent node raises item_not_found"""
        resource = pubsub_resource
        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, pubsub_resource):
        """outcast users cannot subscribe"""
        resource = pubsub_resource
        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, pubsub_resource):
        """open access model allows anyone to subscribe"""
        resource = pubsub_resource
        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, pubsub_resource
    ):
        """whitelist access model requires member affiliation"""
        resource = pubsub_resource
        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, pubsub_resource):
        """whitelist access model allows members to subscribe"""
        resource = pubsub_resource
        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, pubsub_resource):
        """subscribing when already pending raises not-authorized"""
        resource = pubsub_resource
        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, pubsub_resource):
        """unsubscribing for another user raises forbidden"""
        resource = pubsub_resource
        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, pubsub_resource):
        """unsubscribing from nonexistent node raises item_not_found"""
        resource = pubsub_resource
        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, pubsub_resource):
        """unsubscribing removes the subscription"""
        resource = pubsub_resource
        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, pubsub_resource):
        """retrieving items from nonexistent node raises item_not_found"""
        resource = pubsub_resource
        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, pubsub_resource):
        """open access model allows anyone to retrieve items"""
        resource = pubsub_resource
        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, pubsub_resource
    ):
        """whitelist access model rejects non-members"""
        resource = pubsub_resource
        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, pubsub_resource):
        """whitelist access model allows members to retrieve items"""
        resource = pubsub_resource
        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()

    @ed
    async def test_presence_model_rejects_non_contacts(
        self, host, client, pubsub_resource
    ):
        """presence access model rejects non-contacts"""
        resource = pubsub_resource
        request = make_pubsub_request(
            sender=jid.JID("stranger@test.example"),
            nodeIdentifier="test_node",
        )

        node = make_pubsub_node(
            name="test_node",
            access_model=AccessModel.presence,
            pep=True,
        )
        host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)

        # Mock XEP-0356 roster to not include the stranger
        mock_xep_0356 = MagicMock()
        mock_xep_0356.get_roster = AsyncMock(return_value={})
        host.plugins["XEP-0356"] = mock_xep_0356

        with raises(StanzaError, match="forbidden"):
            await resource._items(request)

    @ed
    async def test_presence_model_rejects_contact_from_subscription(
        self, host, client, pubsub_resource
    ):
        """presence access model rejects when "to" subscription is missing"""
        requestor_jid = jid.JID("contact@test.example")

        resource = pubsub_resource
        request = make_pubsub_request(
            sender=requestor_jid,
            nodeIdentifier="test_node",
        )

        node = make_pubsub_node(
            name="test_node",
            access_model=AccessModel.presence,
            pep=True,
        )
        host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)

        mock_xep_0356 = MagicMock()
        roster_item = RosterItem(
            entity=requestor_jid,
            subscriptionTo=False,
            subscriptionFrom=True,
        )
        mock_xep_0356.get_roster = AsyncMock(
            return_value={requestor_jid: roster_item}
        )
        host.plugins["XEP-0356"] = mock_xep_0356

        with raises(StanzaError, match="forbidden"):
            await resource._items(request)

    @ed
    async def test_presence_model_allows_contact_from_subscription(
        self, host, client, pubsub_resource
    ):
        """presence access model allows when requestor has "from" subscription"""
        resource = pubsub_resource
        requestor_jid = jid.JID("contact@test.example")
        request = make_pubsub_request(
            sender=requestor_jid,
            nodeIdentifier="test_node",
            recipient=client.pubsub_service,
        )

        node = make_pubsub_node(
            name="test_node",
            access_model=AccessModel.presence,
            pep=True,
        )
        host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
        host.memory.storage.get_items = AsyncMock(return_value=([], {}))
        mock_xep_0356 = MagicMock()
        host.plugins["XEP-0356"] = mock_xep_0356
        host.plugins["XEP-0496"]._xep_0356 = mock_xep_0356

        roster_item = RosterItem(
            entity=requestor_jid,
            subscriptionTo=False,
            subscriptionFrom=True,
        )
        mock_xep_0356.get_roster = AsyncMock(
            return_value={requestor_jid: roster_item}
        )

        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, pubsub_resource):
        """retracting without item identifiers raises bad-request"""
        resource = pubsub_resource
        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, pubsub_resource):
        """retracting from nonexistent node raises item_not_found"""
        resource = pubsub_resource
        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, pubsub_resource):
        """owners can retract any item from their node"""
        resource = pubsub_resource
        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, pubsub_resource):
        """publishers can retract any item from their node"""
        resource = pubsub_resource
        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, pubsub_resource):
        """non-owners can only retract their own items"""
        resource = pubsub_resource
        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, pubsub_resource):
        """users can retract their own items"""
        resource = pubsub_resource
        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, pubsub_resource):
        """purging nonexistent node raises item_not_found"""
        resource = pubsub_resource
        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, pubsub_resource):
        """only owners can purge a node"""
        resource = pubsub_resource
        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, pubsub_resource):
        """owners can purge their node"""
        resource = pubsub_resource
        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, pubsub_resource):
        """deleting nonexistent node raises item_not_found"""
        resource = pubsub_resource
        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, pubsub_resource):
        """only owners can delete a node"""
        resource = pubsub_resource
        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, pubsub_resource):
        """owners can delete their node"""
        resource = pubsub_resource
        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, pubsub_resource):
        """only owners can get node configuration"""
        resource = pubsub_resource
        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, pubsub_resource):
        """owners can get node configuration"""
        resource = pubsub_resource
        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, pubsub_resource):
        """only owners can set node configuration"""
        resource = pubsub_resource
        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, pubsub_resource):
        """owners can set node configuration"""
        resource = pubsub_resource
        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, pubsub_resource):
        """only owners can retrieve all affiliations"""
        resource = pubsub_resource
        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, pubsub_resource):
        """only owners can set affiliations"""
        resource = pubsub_resource
        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, pubsub_resource):
        """owners cannot change their own affiliation to non-owner"""
        resource = pubsub_resource
        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, pubsub_resource):
        """only owners can retrieve all subscriptions"""
        resource = pubsub_resource
        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, pubsub_resource):
        """only owners can set subscriptions"""
        resource = pubsub_resource
        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, pubsub_resource):
        """owners can set subscriptions for entities"""
        resource = pubsub_resource
        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, pubsub_resource):
        """owners cannot subscribe outcast entities"""
        resource = pubsub_resource
        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)


class TestPubsubAutoSubscribe:
    """Tests for PEP auto-subscription."""

    @ed
    async def test_auto_subscribe_skips_no_notify_capability(
        self, host, client, pubsub_resource
    ):
        """auto-subscription skips contacts without +notify capability"""
        client.is_component = True

        contact_jid = jid.JID("contact@test.example/resource1")
        presence_elt = domish.Element((None, "presence"))
        presence_elt["from"] = contact_jid.full()

        # Mock caps hash
        caps_hash = "abc123"
        host.memory.get_entity_datum = MagicMock(return_value=caps_hash)

        # Mock disco info WITHOUT +notify feature
        mock_disco_info = MagicMock()
        mock_disco_info.features = [
            "http://jabber.org/protocol/caps",
            "http://jabber.org/protocol/tune",  # No +notify
        ]
        host.memory.disco.hashes = {caps_hash: mock_disco_info}

        await pubsub_resource._on_presence_received(client, contact_jid, presence_elt)

        # Check that no query was made (no +notify)
        host.memory.storage.get_pep_last_items.assert_not_called()

    @ed
    async def test_auto_subscribe_skips_no_presence_subscription(
        self, host, client, pubsub_resource
    ):
        """auto-subscription skips contacts without presence subscription"""
        client.is_component = True

        contact_jid = jid.JID("stranger@test.example/resource1")
        requestor_jid = contact_jid.userhostJID()
        presence_elt = domish.Element((None, "presence"))
        presence_elt["from"] = contact_jid.full()

        # Mock caps hash with +notify
        caps_hash = "abc123"
        host.memory.get_entity_datum = MagicMock(return_value=caps_hash)
        mock_disco_info = MagicMock()
        mock_disco_info.features = ["http://jabber.org/protocol/tune+notify"]
        host.memory.disco.hashes = {caps_hash: mock_disco_info}

        # Mock get_roster to return empty roster (no presence subscription)
        mock_xep_0356 = MagicMock()
        mock_xep_0356.get_roster = AsyncMock(return_value={})
        host.plugins["XEP-0356"] = mock_xep_0356
        pubsub_resource.xep_0356 = mock_xep_0356

        host.memory.storage.get_pep_last_items = AsyncMock(return_value=[])

        await pubsub_resource._on_presence_received(client, contact_jid, presence_elt)

        mock_xep_0356.get_roster.assert_called_once_with(client, requestor_jid)

        # Verify get_pep_last_items was called with only self-subscription, not the
        # stranger
        host.memory.storage.get_pep_last_items.assert_called_once()
        call_args = host.memory.storage.get_pep_last_items.call_args
        # We want the second positional argument (list of publishers)
        publishers_arg = call_args[0][1]
        assert publishers_arg == [requestor_jid], (
            f"Expected only self-subscription, got {publishers_arg}"
        )