erebion/libervia-backend
Watch
1
0
Fork
You've already forked libervia-backend
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

tests (unit/component pubsub): Fix tests

Browse source 
rel 476
This commit is contained in:
Goffi 2026-04-10 23:16:10 +02:00
commit 31b55fbbe7
1 changed files with 20 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

22
tests/unit/test_component_pubsub.py
View file

@ -21,7 +21,7 @@ Tests for the PubsubComponent implementation.
"""
from unittest.mock import MagicMock, AsyncMock, patch
from pytest import raises
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
@ -38,6 +38,21 @@ from libervia.backend.memory.sqla_mapping import (
SubscriptionState,
)
from libervia.backend.plugins.plugin_comp_pubsub import PubsubResource
from libervia.backend.plugins.plugin_xep_0496 import XEP_0496
@fixture(autouse=True)
def setup_xep_0496(host, client):
"""Set up XEP-0496 plugin instance for pubsub tests.
This autouse fixture creates a real XEP_0496 instance so that the actual
permission-checking logic is tested rather than mocked behavior.
It also links the client to the host so PubsubResource can access the plugin.
"""
xep_0496 = XEP_0496(host)
host.plugins["XEP-0496"] = xep_0496
client.host = host
return xep_0496
def make_pubsub_request(
@ -482,7 +497,7 @@ class TestPubsubResourceSubscribe:
)
host.memory.storage.get_pubsub_node = AsyncMock(return_value=node)
with raises(StanzaError, match="not-authorized"):
with raises(StanzaError, match="forbidden"):
await resource._subscribe(request)
@ed
@ -966,6 +981,9 @@ class TestPubsubResourceDelete:
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)