Stefan-Sanger
aa822329dd
The test was skipping because it depended on the second_client fixture, which uses XMPP_JID2 (a distinct account). Carbons require both resources to share the same bare JID. Rewrite to create a second VNCXmppClient inline with the same bare JID as xmpp_client but a different resource (/carbon-<random>), so the test runs regardless of whether XMPP_JID2 is configured or distinct. Part-of: <http://gitlab.vnc.biz/uxf/vnctalk-prosody/-/merge_requests/3>
311 lines
14 KiB
Python
311 lines
14 KiB
Python
"""Core XMPP server tests: connectivity, auth, disco, MAM, carbons, REST roundtrips."""
|
|
import asyncio
|
|
import uuid
|
|
from datetime import datetime, timezone, timedelta
|
|
import pytest
|
|
import slixmpp
|
|
import xml.etree.ElementTree as ET
|
|
|
|
from conftest import VNCXmppClient
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestCoreXmpp:
|
|
|
|
async def test_connect_and_auth(self, xmpp_client):
|
|
"""Client must authenticate successfully."""
|
|
assert xmpp_client.session_started_event.is_set()
|
|
assert xmpp_client.boundjid.bare
|
|
|
|
async def test_disco_info_server(self, xmpp_client, xmpp_config):
|
|
"""Server domain must advertise basic disco features.
|
|
|
|
MAM and stanza-id are account-level features (hooked on
|
|
account-disco-info in mod_mam.lua), not server domain features.
|
|
"""
|
|
res = await xmpp_client.disco_info(to=xmpp_config["domain"])
|
|
features = [f.get("var") for f in res.xml.findall(".//{http://jabber.org/protocol/disco#info}feature")]
|
|
required = [
|
|
"http://jabber.org/protocol/disco#info",
|
|
"http://jabber.org/protocol/disco#items",
|
|
]
|
|
for f in required:
|
|
assert f in features, f"Missing required disco feature: {f}"
|
|
|
|
async def test_disco_info_account(self, xmpp_client):
|
|
"""Account disco must advertise MAM and stanza-id.
|
|
|
|
mod_mam.lua hooks account-disco-info to inject these.
|
|
"""
|
|
res = await xmpp_client.disco_info(to=xmpp_client.boundjid.bare)
|
|
features = [f.get("var") for f in res.xml.findall(".//{http://jabber.org/protocol/disco#info}feature")]
|
|
assert "urn:xmpp:mam:2" in features, "MAM feature missing from account disco"
|
|
assert "urn:xmpp:sid:0" in features, "stanza-id feature missing from account disco"
|
|
|
|
async def test_mam_available(self, xmpp_client):
|
|
"""MAM prefs query must succeed."""
|
|
res = await xmpp_client.get_mam_prefs()
|
|
assert res.xml.find("{urn:xmpp:mam:2}prefs") is not None
|
|
|
|
async def test_carbons_enable(self, xmpp_client):
|
|
"""Carbons enable IQ must succeed."""
|
|
await xmpp_client.enable_carbons()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestBoshAndWebsocket:
|
|
|
|
async def test_bosh_url_reachable(self, xmpp_config):
|
|
import aiohttp
|
|
url = xmpp_config.get("bosh_url")
|
|
if not url:
|
|
pytest.skip("No BOSH URL configured")
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as resp:
|
|
assert resp.status in (200, 404, 403)
|
|
|
|
async def test_websocket_url_reachable(self, xmpp_config):
|
|
import aiohttp
|
|
url = xmpp_config.get("ws_url")
|
|
if not url:
|
|
pytest.skip("No WebSocket URL configured")
|
|
async with aiohttp.ClientSession() as session:
|
|
try:
|
|
async with session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as resp:
|
|
assert resp.status in (200, 400, 426, 404)
|
|
except aiohttp.ClientResponseError as e:
|
|
assert e.status in (400, 426)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestRestInjection:
|
|
"""mod_http_rest endpoint behaviour and round-trips."""
|
|
|
|
async def test_rest_accepts_xml(self, rest_injector):
|
|
body = '<message to="test@example.com" from="admin@example.com" type="chat"><body>hello</body></message>'
|
|
status, _ = await rest_injector.inject(body)
|
|
# 201 means accepted and injected; 422 means parseable but maybe not routable
|
|
assert status in (201, 422), f"Expected 201/422, got {status}"
|
|
|
|
async def test_rest_rejects_non_xml(self, rest_injector):
|
|
status, _ = await rest_injector.inject("not xml")
|
|
# 415 = Unsupported Media Type; 422 = Unprocessable Entity ( Prosody returns this for invalid XML)
|
|
assert status in (415, 422), f"Expected 415 or 422 for non-XML, got {status}"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestRestToMAMRoundtrip:
|
|
"""Priority 1 from AUDIT: verify REST-injected messages land in the sender's MAM archive."""
|
|
|
|
async def _poll_mam_for_id(self, xmpp_client, target_id, max_attempts=10, delay=1.0, start=None):
|
|
"""Poll MAM until target_id appears or max_attempts exhausted.
|
|
|
|
Archiving via REST is asynchronous; the HTTP 201 response fires the
|
|
vnc-rest-message event, but archive:append may complete after the
|
|
REST endpoint has already responded. We retry instead of failing
|
|
immediately on an empty first page.
|
|
|
|
start: ISO 8601 timestamp to filter MAM results. Pass a value from
|
|
just before injection so the query returns only the tail of the archive
|
|
rather than the oldest page, which would never contain a fresh message
|
|
when the archive is large.
|
|
"""
|
|
for attempt in range(max_attempts):
|
|
results = await xmpp_client.query_mam_and_collect(timeout=10, start=start)
|
|
ids = []
|
|
for r in results:
|
|
forwarded = r.find("{urn:xmpp:forward:0}forwarded")
|
|
if forwarded is not None:
|
|
msg_el = next(
|
|
(c for c in forwarded if c.tag in (
|
|
"{jabber:client}message",
|
|
"{urn:xmpp:forward:0}message",
|
|
"message",
|
|
)),
|
|
None,
|
|
)
|
|
if msg_el is not None:
|
|
ids.append(msg_el.get("id"))
|
|
if target_id in ids:
|
|
return True
|
|
if attempt < max_attempts - 1:
|
|
await asyncio.sleep(delay)
|
|
return False
|
|
|
|
async def test_rest_message_archived_in_sender_mam(self, xmpp_client, xmpp_config, rest_injector):
|
|
"""POST /rest a message, then query sender's MAM and find it."""
|
|
import xml.etree.ElementTree as ET
|
|
from slixmpp.xmlstream.handler import Callback
|
|
from slixmpp.xmlstream.matcher import MatchXPath
|
|
msg_id = f"rest-mam-{uuid.uuid4().hex[:12]}"
|
|
from_jid = xmpp_client.boundjid.bare
|
|
to_jid = f"nonexistent-{uuid.uuid4().hex[:8]}@{xmpp_config['domain']}"
|
|
body = f"REST injected test message {msg_id}"
|
|
|
|
start_time = (datetime.now(timezone.utc) - timedelta(seconds=5)).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
injected_xml = (
|
|
f'<message id="{msg_id}" from="{from_jid}" to="{to_jid}" type="chat">'
|
|
f"<body>{body}</body>"
|
|
f"</message>"
|
|
)
|
|
print(f"\n[REST] injecting: {injected_xml}")
|
|
status, resp_text = await rest_injector.inject(injected_xml)
|
|
print(f"[REST] response: status={status} body={resp_text!r}")
|
|
assert status in (201, 422), f"REST injection failed with {status}"
|
|
|
|
print("[wait] sleeping 30s for archive to settle ...")
|
|
await asyncio.sleep(30)
|
|
|
|
query_id = f"q-{uuid.uuid4().hex[:8]}"
|
|
results = []
|
|
|
|
# MAM result wrappers have no top-level <body>, so slixmpp's high-level
|
|
# "message" event never fires for them (its matcher requires a <body>).
|
|
# Register a low-level stream handler matching the result element instead.
|
|
def on_mam_result(stanza):
|
|
try:
|
|
print(f"[mam result stanza] {stanza}", flush=True)
|
|
res = stanza.xml.find("{urn:xmpp:mam:2}result")
|
|
if res is not None and res.get("queryid") == query_id:
|
|
results.append(res)
|
|
except Exception as exc:
|
|
print(f"[mam result ERROR] {exc}", flush=True)
|
|
|
|
handler_name = f"mam_collect_{query_id}"
|
|
xmpp_client.register_handler(Callback(
|
|
handler_name,
|
|
MatchXPath(f"{{{xmpp_client.default_ns}}}message/{{urn:xmpp:mam:2}}result"),
|
|
on_mam_result,
|
|
))
|
|
try:
|
|
iq = xmpp_client.make_iq_set()
|
|
query = ET.SubElement(iq.xml, "{urn:xmpp:mam:2}query")
|
|
query.set("queryid", query_id)
|
|
xmpp_client._add_mam_query_form(query, start=start_time)
|
|
query_xml = ET.tostring(iq.xml, encoding="unicode")
|
|
print(f"[MAM query] {query_xml}")
|
|
fin_iq = await asyncio.wait_for(iq.send(), timeout=10)
|
|
await asyncio.sleep(0.5)
|
|
print(f"[MAM fin] {ET.tostring(fin_iq.xml, encoding='unicode')}")
|
|
print(f"[mam results received] {len(results)}")
|
|
finally:
|
|
xmpp_client.remove_handler(handler_name)
|
|
|
|
print(f"[MAM results] {len(results)} result(s) received")
|
|
ids = []
|
|
for r in results:
|
|
print(f" result: {ET.tostring(r, encoding='unicode')}")
|
|
forwarded = r.find("{urn:xmpp:forward:0}forwarded")
|
|
if forwarded is not None:
|
|
# Prosody may omit xmlns='jabber:client' on the inner <message>,
|
|
# causing it to inherit the forwarded namespace instead.
|
|
# Match by local name to handle both cases.
|
|
msg_el = next(
|
|
(c for c in forwarded if c.tag in (
|
|
"{jabber:client}message",
|
|
"{urn:xmpp:forward:0}message",
|
|
"message",
|
|
)),
|
|
None,
|
|
)
|
|
if msg_el is not None:
|
|
ids.append(msg_el.get("id"))
|
|
print(f" -> inner message id={msg_el.get('id')} tag={msg_el.tag}")
|
|
print(f"[MAM results] message ids found: {ids}")
|
|
print(f"[MAM results] looking for: {msg_id}")
|
|
|
|
assert msg_id in ids, f"REST-injected message {msg_id} not found in sender MAM archive"
|
|
|
|
async def test_rest_message_to_non_roster_contact_is_archived(self, xmpp_client, xmpp_config, rest_injector):
|
|
"""shall_store always returns true, so even non-roster messages are archived."""
|
|
msg_id = f"rest-noroster-{uuid.uuid4().hex[:12]}"
|
|
from_jid = xmpp_client.boundjid.bare
|
|
# Use a clearly foreign domain to ensure no roster entry exists
|
|
to_jid = f"foreign-{uuid.uuid4().hex[:8]}@foreign.example.com"
|
|
body = f"Non-roster test {msg_id}"
|
|
|
|
start_time = (datetime.now(timezone.utc) - timedelta(seconds=5)).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
status, _ = await rest_injector.inject_message(msg_id, from_jid, to_jid, body)
|
|
assert status in (201, 422)
|
|
|
|
found = await self._poll_mam_for_id(xmpp_client, msg_id, start=start_time)
|
|
assert found, f"Non-roster REST message {msg_id} should be archived (shall_store=true contract broken)"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestRestToCarbonsRoundtrip:
|
|
"""Priority 2 from AUDIT: REST-injected messages are carbon-copied."""
|
|
|
|
async def test_rest_message_carbon_copied_to_second_resource(
|
|
self, xmpp_client, xmpp_config, rest_injector
|
|
):
|
|
"""Enable carbons on two resources of the same account. Inject via REST.
|
|
Second resource must receive a carbon copy.
|
|
|
|
Creates a second client with the same bare JID as xmpp_client but a
|
|
different resource, so this test works even when XMPP_JID2 is a
|
|
distinct account (or not configured at all).
|
|
"""
|
|
bare_jid = xmpp_client.boundjid.bare
|
|
password = xmpp_config["password"]
|
|
second_jid = f"{bare_jid}/carbon-{uuid.uuid4().hex[:4]}"
|
|
|
|
second_client = VNCXmppClient(
|
|
second_jid, password,
|
|
host=xmpp_config["host"], port=xmpp_config["port"],
|
|
use_ssl=True, verify_ssl=xmpp_config["verify_ssl"],
|
|
)
|
|
try:
|
|
await second_client.async_connect(timeout=30)
|
|
|
|
await xmpp_client.enable_carbons()
|
|
await second_client.enable_carbons()
|
|
await asyncio.sleep(0.5)
|
|
|
|
carbon_received = asyncio.Event()
|
|
carbon_bodies = []
|
|
|
|
# A REST-injected message enters through the sender-side (c2s)
|
|
# carbons path (the mod_carbons patch hooks vnc-rest-message as a
|
|
# c2s handler), so other resources of the *sender* receive <sent>
|
|
# carbons; <received> would only appear on recipient resources.
|
|
# slixmpp's "message" event only fires for messages with a <body>;
|
|
# carbon wrappers have none, so a raw stream handler is needed.
|
|
from slixmpp.xmlstream.handler import Callback
|
|
from slixmpp.xmlstream.matcher import MatchXPath
|
|
|
|
def on_carbon_wrapper(msg):
|
|
for tag in ("sent", "received"):
|
|
carb = msg.xml.find("{urn:xmpp:carbons:2}" + tag)
|
|
if carb is not None:
|
|
fwd = carb.find("{urn:xmpp:forward:0}forwarded")
|
|
if fwd is not None:
|
|
body_el = fwd.find("{jabber:client}message/{jabber:client}body")
|
|
if body_el is not None and body_el.text:
|
|
carbon_bodies.append(body_el.text)
|
|
carbon_received.set()
|
|
|
|
second_client.register_handler(Callback(
|
|
"carbon-probe", MatchXPath("{jabber:client}message"), on_carbon_wrapper
|
|
))
|
|
try:
|
|
msg_id = f"rest-carbons-{uuid.uuid4().hex[:12]}"
|
|
from_jid = bare_jid
|
|
to_jid = bare_jid
|
|
body = f"carbon test {msg_id}"
|
|
|
|
status, _ = await rest_injector.inject_message(msg_id, from_jid, to_jid, body)
|
|
assert status in (201, 422)
|
|
|
|
try:
|
|
await asyncio.wait_for(carbon_received.wait(), timeout=8)
|
|
except asyncio.TimeoutError:
|
|
pytest.fail("Carbon copy not received on second resource after REST injection")
|
|
|
|
assert any(body in b for b in carbon_bodies), "Carbon body does not match injected message"
|
|
finally:
|
|
second_client.remove_handler("carbon-probe")
|
|
finally:
|
|
if second_client.session_started_event.is_set():
|
|
await second_client.async_disconnect()
|