Stefan-Sanger
913c6be84f
Always sending Host: <xmpp domain> to /rest broke when REST was fronted by an ingress (TLS hostname mismatch -> 431). Add a --rest-host-header option (REST_HOST_HEADER env, default 'auto') that sends the XMPP domain Host only for IP/localhost URLs where prosody needs it for vhost routing; 'none' disables, any other value is sent verbatim. Documents the option in tests/README.md. Part-of: <http://gitlab.vnc.biz/uxf/vnctalk-prosody/-/merge_requests/3>
600 lines
24 KiB
Python
600 lines
24 KiB
Python
"""Test configuration and shared fixtures for vnctalk-prosody verification."""
|
|
import asyncio
|
|
import os
|
|
import ssl
|
|
import uuid
|
|
import pytest
|
|
import pytest_asyncio
|
|
import slixmpp
|
|
import xml.etree.ElementTree as ET
|
|
|
|
|
|
class VNCXmppClient(slixmpp.ClientXMPP):
|
|
"""Async-friendly XMPP client for vnctalk tests."""
|
|
|
|
def __init__(self, jid, password, host=None, port=5222, use_ssl=True, verify_ssl=False):
|
|
ssl_context = None
|
|
if use_ssl and not verify_ssl:
|
|
ssl_context = ssl.create_default_context()
|
|
ssl_context.check_hostname = False
|
|
ssl_context.verify_mode = ssl.CERT_NONE
|
|
super().__init__(jid, password, host=host or '', port=port, ssl_context=ssl_context)
|
|
self.connected_event = asyncio.Event()
|
|
self.disconnected_event = asyncio.Event()
|
|
self.session_started_event = asyncio.Event()
|
|
self.add_event_handler("session_start", self._on_session_start)
|
|
self.add_event_handler("disconnected", self._on_disconnected)
|
|
self._use_ssl = use_ssl
|
|
# Honor the configured host/port: slixmpp's connect() otherwise does
|
|
# SRV/A resolution on the JID domain, which breaks against a local
|
|
# compose stack whose domain (e.g. example.com) doesn't resolve to it.
|
|
self._connect_host = host or None
|
|
self._connect_port = port
|
|
|
|
def _on_session_start(self, event):
|
|
self.session_started_event.set()
|
|
|
|
def _on_disconnected(self, event):
|
|
self.disconnected_event.set()
|
|
|
|
async def async_connect(self, timeout=30):
|
|
await self.connect(self._connect_host, self._connect_port)
|
|
await asyncio.wait_for(self.session_started_event.wait(), timeout=timeout)
|
|
# announce presence so the server routes directed messages to this
|
|
# resource; without it, chats to our bare JID go to offline storage
|
|
self.send_presence()
|
|
await asyncio.sleep(0.2)
|
|
|
|
async def async_disconnect(self):
|
|
self.disconnect()
|
|
await asyncio.wait_for(self.disconnected_event.wait(), timeout=10)
|
|
|
|
async def enable_carbons(self):
|
|
iq = self.make_iq_set()
|
|
iq.append(ET.Element("{urn:xmpp:carbons:2}enable"))
|
|
await iq.send()
|
|
|
|
async def get_mam_prefs(self):
|
|
iq = self.make_iq_get()
|
|
iq.append(ET.Element("{urn:xmpp:mam:2}prefs"))
|
|
res = await iq.send()
|
|
return res
|
|
|
|
def _add_mam_query_form(self, query, with_jid=None, start=None, end=None):
|
|
"""Add a valid XEP-0313 data form to a MAM query element."""
|
|
x = ET.SubElement(query, "{jabber:x:data}x")
|
|
x.set("type", "submit")
|
|
ft = ET.SubElement(x, "{jabber:x:data}field")
|
|
ft.set("type", "hidden")
|
|
ft.set("var", "FORM_TYPE")
|
|
ET.SubElement(ft, "{jabber:x:data}value").text = "urn:xmpp:mam:2"
|
|
if with_jid:
|
|
f = ET.SubElement(x, "{jabber:x:data}field")
|
|
f.set("var", "with")
|
|
ET.SubElement(f, "{jabber:x:data}value").text = with_jid
|
|
if start:
|
|
f = ET.SubElement(x, "{jabber:x:data}field")
|
|
f.set("var", "start")
|
|
ET.SubElement(f, "{jabber:x:data}value").text = start
|
|
if end:
|
|
f = ET.SubElement(x, "{jabber:x:data}field")
|
|
f.set("var", "end")
|
|
ET.SubElement(f, "{jabber:x:data}value").text = end
|
|
|
|
async def query_mam(self, with_jid=None, start=None, end=None):
|
|
"""Query MAM and return the full IQ result (including forwarded results)."""
|
|
iq = self.make_iq_set()
|
|
query = ET.SubElement(iq.xml, "{urn:xmpp:mam:2}query")
|
|
query.set("queryid", f"q-{uuid.uuid4().hex[:8]}")
|
|
self._add_mam_query_form(query, with_jid=with_jid, start=start, end=end)
|
|
res = await iq.send()
|
|
return res
|
|
|
|
async def query_mam_and_collect(self, timeout=10, with_jid=None, start=None, to=None):
|
|
"""Query MAM and collect all <result> stanzas.
|
|
|
|
Prosody sends <result> elements as individual <message> stanzas and the
|
|
<fin> element inside the IQ response.
|
|
|
|
to: target archive. Omit for the user's own archive; pass a room JID to
|
|
query a MUC's archive (MUC MAM lives at the room, not the user).
|
|
|
|
NOTE: we cannot use the high-level "message" event here. slixmpp only
|
|
fires "message" for stanzas with a top-level <body> (its matcher is
|
|
'{jabber:client}message/{jabber:client}body'). A MAM result wrapper has
|
|
no <body> of its own — its only child is <result xmlns='urn:xmpp:mam:2'>
|
|
— so we must register a low-level stream handler that matches the result
|
|
element directly. (slixmpp's own xep_0313 plugin does the same thing.)
|
|
"""
|
|
from slixmpp.xmlstream.handler import Callback
|
|
from slixmpp.xmlstream.matcher import MatchXPath
|
|
|
|
query_id = f"q-{uuid.uuid4().hex[:8]}"
|
|
results = []
|
|
|
|
def on_mam_result(stanza):
|
|
res = stanza.xml.find("{urn:xmpp:mam:2}result")
|
|
if res is not None and res.get("queryid") == query_id:
|
|
results.append(res)
|
|
|
|
handler_name = f"mam_collect_{query_id}"
|
|
self.register_handler(Callback(
|
|
handler_name,
|
|
MatchXPath(f"{{{self.default_ns}}}message/{{urn:xmpp:mam:2}}result"),
|
|
on_mam_result,
|
|
))
|
|
try:
|
|
iq = self.make_iq_set(ito=to)
|
|
query = ET.SubElement(iq.xml, "{urn:xmpp:mam:2}query")
|
|
query.set("queryid", query_id)
|
|
self._add_mam_query_form(query, with_jid=with_jid, start=start)
|
|
# IQ response contains <fin>; result messages arrive before it
|
|
await asyncio.wait_for(iq.send(), timeout=timeout)
|
|
# Grace period for any trailing <result> messages
|
|
await asyncio.sleep(0.5)
|
|
return results
|
|
finally:
|
|
self.remove_handler(handler_name)
|
|
|
|
async def get_vcard(self, to=None):
|
|
iq = self.make_iq_get(ito=to)
|
|
iq.append(ET.Element("{vcard-temp}vCard"))
|
|
res = await iq.send()
|
|
return res
|
|
|
|
async def set_vcard(self, vcard_xml):
|
|
iq = self.make_iq_set()
|
|
iq.append(vcard_xml)
|
|
res = await iq.send()
|
|
return res
|
|
|
|
async def disco_info(self, to=None, timeout=10):
|
|
iq = self.make_iq_get(ito=to)
|
|
iq.append(ET.Element("{http://jabber.org/protocol/disco#info}query"))
|
|
# bound the wait: a non-responding server would otherwise hang forever
|
|
res = await iq.send(timeout=timeout)
|
|
return res
|
|
|
|
async def disco_items(self, to=None, timeout=10):
|
|
iq = self.make_iq_get(ito=to)
|
|
iq.append(ET.Element("{http://jabber.org/protocol/disco#items}query"))
|
|
res = await iq.send(timeout=timeout)
|
|
return res
|
|
|
|
async def join_muc(self, room_jid, nick, timeout=5):
|
|
"""Join a MUC room and wait for self-presence (code 110)."""
|
|
joined = asyncio.Event()
|
|
presences = []
|
|
|
|
def on_presence(pres):
|
|
# pres["from"] is a slixmpp JID object, not a str
|
|
if str(pres["from"]).startswith(room_jid):
|
|
x = pres.xml.find("{http://jabber.org/protocol/muc#user}x")
|
|
if x is not None:
|
|
for status in x.findall("{http://jabber.org/protocol/muc#user}status"):
|
|
if status.get("code") == "110":
|
|
joined.set()
|
|
presences.append(pres)
|
|
|
|
self.add_event_handler("presence", on_presence)
|
|
try:
|
|
presence = self.make_presence(pto=f"{room_jid}/{nick}")
|
|
ET.SubElement(presence.xml, "{http://jabber.org/protocol/muc}x")
|
|
# presence.send() writes to the stream and returns None (not awaitable)
|
|
presence.send()
|
|
await asyncio.wait_for(joined.wait(), timeout=timeout)
|
|
return presences
|
|
finally:
|
|
self.del_event_handler("presence", on_presence)
|
|
|
|
async def configure_muc(self, room_jid, fields=None, timeout=8):
|
|
"""Submit the MUC owner config form to unlock a freshly-created room.
|
|
|
|
A newly created Prosody room is locked until its owner submits a config
|
|
form; while locked, other users cannot join. With no fields this submits
|
|
an "instant room" (empty submit), accepting defaults and unlocking it.
|
|
"""
|
|
iq = self.make_iq_set(ito=room_jid)
|
|
query = ET.SubElement(iq.xml, "{http://jabber.org/protocol/muc#owner}query")
|
|
x = ET.SubElement(query, "{jabber:x:data}x")
|
|
x.set("type", "submit")
|
|
if fields:
|
|
# a non-empty config submit must declare the form type, otherwise
|
|
# Prosody rejects it ("Form is not of type room configuration")
|
|
ft = ET.SubElement(x, "{jabber:x:data}field")
|
|
ft.set("type", "hidden")
|
|
ft.set("var", "FORM_TYPE")
|
|
ET.SubElement(ft, "{jabber:x:data}value").text = (
|
|
"http://jabber.org/protocol/muc#roomconfig"
|
|
)
|
|
for var, value in fields.items():
|
|
f = ET.SubElement(x, "{jabber:x:data}field")
|
|
f.set("var", var)
|
|
ET.SubElement(f, "{jabber:x:data}value").text = value
|
|
await iq.send(timeout=timeout)
|
|
|
|
async def leave_muc(self, room_jid, nick):
|
|
presence = self.make_presence(pto=f"{room_jid}/{nick}", ptype="unavailable")
|
|
# presence.send() writes to the stream and returns None (not awaitable)
|
|
presence.send()
|
|
|
|
async def destroy_muc(self, room_jid):
|
|
"""Send owner destroy request."""
|
|
iq = self.make_iq_set(ito=room_jid)
|
|
query = ET.SubElement(iq.xml, "{http://jabber.org/protocol/muc#owner}query")
|
|
ET.SubElement(query, "{http://jabber.org/protocol/muc#owner}destroy")
|
|
try:
|
|
await iq.send()
|
|
except Exception:
|
|
pass # may fail if not owner
|
|
|
|
async def get_muc_affiliations(self, room_jid, affiliation="member"):
|
|
iq = self.make_iq_get(ito=room_jid)
|
|
query = ET.SubElement(iq.xml, "{http://jabber.org/protocol/muc#admin}query")
|
|
item = ET.SubElement(query, "{http://jabber.org/protocol/muc#admin}item")
|
|
item.set("affiliation", affiliation)
|
|
res = await iq.send()
|
|
return res.xml.findall(".//{http://jabber.org/protocol/muc#admin}item")
|
|
|
|
async def set_muc_affiliation(self, room_jid, target_jid, affiliation, reason=""):
|
|
iq = self.make_iq_set(ito=room_jid)
|
|
query = ET.SubElement(iq.xml, "{http://jabber.org/protocol/muc#admin}query")
|
|
item = ET.SubElement(query, "{http://jabber.org/protocol/muc#admin}item")
|
|
item.set("affiliation", affiliation)
|
|
item.set("jid", target_jid)
|
|
if reason:
|
|
ET.SubElement(item, "{http://jabber.org/protocol/muc#admin}reason").text = reason
|
|
await iq.send()
|
|
|
|
|
|
class RESTInjector:
|
|
"""Helper to inject stanzas via mod_http_rest.
|
|
|
|
If auth credentials are provided, they are sent as HTTP Basic Auth headers
|
|
because the /rest endpoint is typically protected by a reverse proxy.
|
|
"""
|
|
|
|
def __init__(self, url, auth_user=None, auth_password=None, host_header=None):
|
|
self.url = url
|
|
self.auth_user = auth_user
|
|
self.auth_password = auth_password
|
|
# See _resolve_rest_host_header: only direct (IP/localhost) prosody
|
|
# access needs a Host: <xmpp domain> override for vhost routing.
|
|
self.host_header = host_header
|
|
|
|
def _headers(self):
|
|
headers = {"Content-Type": "text/xml"}
|
|
if self.host_header:
|
|
headers["Host"] = self.host_header
|
|
if self.auth_user and self.auth_password:
|
|
import base64
|
|
creds = base64.b64encode(
|
|
f"{self.auth_user}:{self.auth_password}".encode()
|
|
).decode()
|
|
headers["Authorization"] = f"Basic {creds}"
|
|
return headers
|
|
|
|
async def inject(self, stanza_xml: str):
|
|
import aiohttp
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.post(
|
|
self.url,
|
|
data=stanza_xml,
|
|
headers=self._headers(),
|
|
timeout=aiohttp.ClientTimeout(total=10)
|
|
) as resp:
|
|
return resp.status, await resp.text()
|
|
|
|
async def inject_message(self, msg_id, from_jid, to_jid, body, type_="chat"):
|
|
xml = (
|
|
f'<message id="{msg_id}" from="{from_jid}" to="{to_jid}" type="{type_}">'
|
|
f"<body>{body}</body>"
|
|
f"</message>"
|
|
)
|
|
return await self.inject(xml)
|
|
|
|
|
|
def pytest_addoption(parser):
|
|
parser.addoption(
|
|
"--xmpp-host", action="store", default=os.getenv("XMPP_HOST", "localhost"),
|
|
help="XMPP server hostname/IP"
|
|
)
|
|
parser.addoption(
|
|
"--xmpp-port", action="store", type=int, default=int(os.getenv("XMPP_PORT", "5222")),
|
|
help="XMPP C2S port"
|
|
)
|
|
parser.addoption(
|
|
"--xmpp-jid", action="store", default=os.getenv("XMPP_JID", ""),
|
|
help="Test account JID (bare or full)"
|
|
)
|
|
parser.addoption(
|
|
"--xmpp-password", action="store", default=os.getenv("XMPP_PASSWORD", ""),
|
|
help="Test account password"
|
|
)
|
|
parser.addoption(
|
|
"--xmpp-jid2", action="store", default=os.getenv("XMPP_JID2", ""),
|
|
help="Second distinct test account JID (for affiliation tests)"
|
|
)
|
|
parser.addoption(
|
|
"--xmpp-password2", action="store", default=os.getenv("XMPP_PASSWORD2", ""),
|
|
help="Second distinct test account password"
|
|
)
|
|
parser.addoption(
|
|
"--xmpp-domain", action="store", default=os.getenv("XMPP_DOMAIN", "example.com"),
|
|
help="XMPP domain"
|
|
)
|
|
parser.addoption(
|
|
"--bosh-url", action="store", default=os.getenv("BOSH_URL", ""),
|
|
help="BOSH URL to test (e.g. http://localhost:5280/http-bind)"
|
|
)
|
|
parser.addoption(
|
|
"--ws-url", action="store", default=os.getenv("WS_URL", ""),
|
|
help="WebSocket URL to test (e.g. ws://localhost:5280/xmpp-websocket)"
|
|
)
|
|
parser.addoption(
|
|
"--rest-url", action="store", default=os.getenv("REST_URL", ""),
|
|
help="mod_http_rest URL (e.g. http://localhost:5280/rest)"
|
|
)
|
|
parser.addoption(
|
|
"--rest-user", action="store", default=os.getenv("REST_USER", ""),
|
|
help="HTTP Basic Auth user for /rest endpoint (reverse proxy credential)"
|
|
)
|
|
parser.addoption(
|
|
"--rest-password", action="store", default=os.getenv("REST_PASSWORD", ""),
|
|
help="HTTP Basic Auth password for /rest endpoint (reverse proxy credential)"
|
|
)
|
|
parser.addoption(
|
|
"--rest-host-header", action="store", default=os.getenv("REST_HOST_HEADER", "auto"),
|
|
help=(
|
|
"Host header to send with /rest requests. 'auto' (default): send the XMPP "
|
|
"domain only when the REST URL host is an IP or localhost (direct prosody "
|
|
"access needs it for vhost routing; ingress-fronted hostnames rewrite Host "
|
|
"themselves and reject a mismatching override, e.g. with 431). "
|
|
"'none': never send. Any other value: send it verbatim."
|
|
)
|
|
)
|
|
parser.addoption(
|
|
"--muc-domain", action="store", default=os.getenv("MUC_DOMAIN", ""),
|
|
help="MUC component domain (e.g. conference.example.com)"
|
|
)
|
|
parser.addoption(
|
|
"--admin-telnet-host", action="store", default=os.getenv("ADMIN_TELNET_HOST", "127.0.0.1"),
|
|
help="Admin telnet host"
|
|
)
|
|
parser.addoption(
|
|
"--admin-telnet-port", action="store", type=int, default=int(os.getenv("ADMIN_TELNET_PORT", "5582")),
|
|
help="Admin telnet port"
|
|
)
|
|
parser.addoption(
|
|
"--pg-host", action="store", default=os.getenv("PG_HOST", ""),
|
|
help="PostgreSQL host (for DB verification)"
|
|
)
|
|
parser.addoption(
|
|
"--pg-port", action="store", type=int, default=int(os.getenv("PG_PORT", "5432")),
|
|
help="PostgreSQL port"
|
|
)
|
|
parser.addoption(
|
|
"--pg-db", action="store", default=os.getenv("PG_DB", "prosody"),
|
|
help="PostgreSQL database name"
|
|
)
|
|
parser.addoption(
|
|
"--pg-user", action="store", default=os.getenv("PG_USER", ""),
|
|
help="PostgreSQL user"
|
|
)
|
|
parser.addoption(
|
|
"--pg-password", action="store", default=os.getenv("PG_PASSWORD", ""),
|
|
help="PostgreSQL password"
|
|
)
|
|
parser.addoption(
|
|
"--mock-url", action="store", default=os.getenv("MOCK_URL", ""),
|
|
help="Base URL of the test mock service (for HTTP side-effect assertions)"
|
|
)
|
|
parser.addoption(
|
|
"--verify-ssl", action="store_true", default=False,
|
|
help="Verify TLS certificates (default: False, because test containers use self-signed certs)"
|
|
)
|
|
parser.addoption(
|
|
"--skip-live", action="store_true", default=False,
|
|
help="Skip tests that require a live XMPP connection"
|
|
)
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def xmpp_config(request):
|
|
return {
|
|
"host": request.config.getoption("--xmpp-host"),
|
|
"port": request.config.getoption("--xmpp-port"),
|
|
"jid": request.config.getoption("--xmpp-jid"),
|
|
"password": request.config.getoption("--xmpp-password"),
|
|
"jid2": request.config.getoption("--xmpp-jid2"),
|
|
"password2": request.config.getoption("--xmpp-password2"),
|
|
"domain": request.config.getoption("--xmpp-domain"),
|
|
"bosh_url": request.config.getoption("--bosh-url"),
|
|
"ws_url": request.config.getoption("--ws-url"),
|
|
"rest_url": request.config.getoption("--rest-url"),
|
|
"rest_user": request.config.getoption("--rest-user"),
|
|
"rest_password": request.config.getoption("--rest-password"),
|
|
"rest_host_header": request.config.getoption("--rest-host-header"),
|
|
"muc_domain": request.config.getoption("--muc-domain"),
|
|
"admin_telnet_host": request.config.getoption("--admin-telnet-host"),
|
|
"admin_telnet_port": request.config.getoption("--admin-telnet-port"),
|
|
"pg_host": request.config.getoption("--pg-host"),
|
|
"pg_port": request.config.getoption("--pg-port"),
|
|
"pg_db": request.config.getoption("--pg-db"),
|
|
"pg_user": request.config.getoption("--pg-user"),
|
|
"pg_password": request.config.getoption("--pg-password"),
|
|
"mock_url": request.config.getoption("--mock-url"),
|
|
"verify_ssl": request.config.getoption("--verify-ssl"),
|
|
"skip_live": request.config.getoption("--skip-live"),
|
|
}
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def xmpp_client(xmpp_config):
|
|
"""Yield a connected XMPP client."""
|
|
cfg = xmpp_config
|
|
if cfg["skip_live"] or not cfg["jid"] or not cfg["password"]:
|
|
pytest.skip("Live XMPP tests disabled or credentials missing")
|
|
|
|
client = VNCXmppClient(
|
|
cfg["jid"], cfg["password"],
|
|
host=cfg["host"], port=cfg["port"], use_ssl=True, verify_ssl=cfg["verify_ssl"]
|
|
)
|
|
try:
|
|
await client.async_connect(timeout=30)
|
|
yield client
|
|
finally:
|
|
if client.session_started_event.is_set():
|
|
await client.async_disconnect()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def second_client(xmpp_config):
|
|
"""Yield a second connected client. Uses a distinct JID if configured, otherwise same bare JID with /test2 resource."""
|
|
cfg = xmpp_config
|
|
if cfg["skip_live"]:
|
|
pytest.skip("Live XMPP tests disabled")
|
|
|
|
if cfg["jid2"] and cfg["password2"]:
|
|
jid = cfg["jid2"]
|
|
password = cfg["password2"]
|
|
elif cfg["jid"] and cfg["password"]:
|
|
# Fallback: same account, different resource
|
|
bare = cfg["jid"].split("/")[0]
|
|
jid = f"{bare}/test2"
|
|
password = cfg["password"]
|
|
else:
|
|
pytest.skip("No credentials for second client")
|
|
|
|
client = VNCXmppClient(
|
|
jid, password,
|
|
host=cfg["host"], port=cfg["port"], use_ssl=True, verify_ssl=cfg["verify_ssl"]
|
|
)
|
|
try:
|
|
await client.async_connect(timeout=30)
|
|
yield client
|
|
finally:
|
|
if client.session_started_event.is_set():
|
|
await client.async_disconnect()
|
|
|
|
|
|
def _resolve_rest_host_header(mode, url, domain):
|
|
"""Decide which Host header (if any) to send with /rest requests.
|
|
|
|
Direct prosody access (IP/localhost URL) needs Host: <xmpp domain> because
|
|
prosody routes HTTP to a vhost by Host and has no default_host. An
|
|
ingress-fronted DNS name rewrites Host for the upstream itself, and
|
|
rejects a request whose Host doesn't match the TLS name (observed: 431
|
|
with an empty body) -- so there the override must not be sent.
|
|
"""
|
|
if mode == "none":
|
|
return None
|
|
if mode and mode != "auto":
|
|
return mode
|
|
import ipaddress
|
|
from urllib.parse import urlparse
|
|
host = urlparse(url).hostname or ""
|
|
try:
|
|
ipaddress.ip_address(host)
|
|
is_direct = True
|
|
except ValueError:
|
|
is_direct = host == "localhost"
|
|
return domain if is_direct else None
|
|
|
|
|
|
@pytest.fixture
|
|
def rest_injector(xmpp_config):
|
|
url = xmpp_config.get("rest_url")
|
|
if not url:
|
|
pytest.skip("No REST URL configured")
|
|
return RESTInjector(
|
|
url,
|
|
auth_user=xmpp_config.get("rest_user"),
|
|
auth_password=xmpp_config.get("rest_password"),
|
|
host_header=_resolve_rest_host_header(
|
|
xmpp_config.get("rest_host_header", "auto"), url, xmpp_config.get("domain")
|
|
),
|
|
)
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def pg_connection(xmpp_config):
|
|
"""Yield a PostgreSQL connection if DB params are configured."""
|
|
cfg = xmpp_config
|
|
if not cfg.get("pg_host") or not cfg.get("pg_user"):
|
|
pytest.skip("PostgreSQL connection params not configured")
|
|
import asyncpg
|
|
try:
|
|
# bound the connect: a half-open tunnel accepts TCP but never completes
|
|
# the PG handshake, which would otherwise hang the whole suite
|
|
conn = await asyncio.wait_for(
|
|
asyncpg.connect(
|
|
host=cfg["pg_host"],
|
|
port=cfg["pg_port"],
|
|
database=cfg["pg_db"],
|
|
user=cfg["pg_user"],
|
|
password=cfg["pg_password"],
|
|
),
|
|
timeout=10,
|
|
)
|
|
except (asyncio.TimeoutError, OSError) as e:
|
|
pytest.skip(f"PostgreSQL not reachable at {cfg['pg_host']}:{cfg['pg_port']}: {e}")
|
|
try:
|
|
yield conn
|
|
finally:
|
|
await conn.close()
|
|
|
|
|
|
class MockClient:
|
|
"""Client for the tests/mocks aiohttp capture service.
|
|
|
|
Exposes reset()/captured()/wait_for() helpers so HTTP side-effect tests can
|
|
assert deterministically on what Prosody POSTed/PUTed to the mock backends.
|
|
"""
|
|
|
|
def __init__(self, base_url, session):
|
|
self.base_url = base_url.rstrip("/")
|
|
self.session = session
|
|
|
|
async def reset(self):
|
|
async with self.session.delete(f"{self.base_url}/__requests") as resp:
|
|
assert resp.status == 204, f"mock reset failed: {resp.status}"
|
|
|
|
async def captured(self, path=None):
|
|
"""Return the captured request list, optionally filtered by path prefix."""
|
|
url = f"{self.base_url}/__requests"
|
|
if path:
|
|
url += f"?path={path}"
|
|
async with self.session.get(url) as resp:
|
|
return await resp.json()
|
|
|
|
async def wait_for(self, path, count=1, timeout=5, interval=0.2):
|
|
"""Poll until at least `count` captured requests match `path` prefix."""
|
|
import time
|
|
deadline = time.monotonic() + timeout
|
|
while time.monotonic() < deadline:
|
|
items = await self.captured(path)
|
|
if len(items) >= count:
|
|
return items
|
|
await asyncio.sleep(interval)
|
|
return await self.captured(path)
|
|
|
|
async def mark_fcm_error(self, token):
|
|
"""Tell the mock to return NotRegistered for the next notify of `token`."""
|
|
async with self.session.get(f"{self.base_url}/__fcm_error", params={"token": token}) as resp:
|
|
return await resp.json()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def mock_client(xmpp_config):
|
|
"""Yield a MockClient if MOCK_URL is configured, else skip."""
|
|
base = xmpp_config.get("mock_url")
|
|
if not base:
|
|
pytest.skip("No MOCK_URL configured")
|
|
import aiohttp
|
|
session = aiohttp.ClientSession()
|
|
try:
|
|
yield MockClient(base, session)
|
|
finally:
|
|
await session.close()
|