Stefan-Sanger
35161b9b6d
1. REST 404 "Unknown host: prosody": the tester reaches prosody by service name (http://prosody:5280), so aiohttp sends Host: prosody, which Prosody rejects as an unknown vhost. The auto host-header heuristic only overrides for IP/localhost. Set REST_HOST_HEADER= example.com explicitly in the tester env so HTTP routing lands on the example.com VirtualHost that serves mod_http_rest /rest. 2. healthcheck.sh not found: the static test_04_infra checks resolve ../config/healthcheck.sh (= /config/healthcheck.sh) but the tester image only ships /tests. Mount ./config:/config:ro so the path resolves inside the container. 3. telnet non-loopback banner: read only 256 bytes, capturing just the ASCII-art top and never the literal "Prosody" text. Bump to 1024 to match test_04_infra.test_telnet_banner, which passes with the larger read. Part-of: <http://gitlab.vnc.biz/uxf/vnctalk-prosody/-/merge_requests/3>
172 lines
7.2 KiB
Python
172 lines
7.2 KiB
Python
"""Low-risk smoke tests (test-improvement.md Phase 7).
|
|
|
|
One positive smoke per module that otherwise has no dedicated coverage:
|
|
* filter_chatstates — drops chatstate-only messages to a CSI-inactive client
|
|
* idlecompat — injects <idle/> into presence carrying jabber:iq:last
|
|
* http_altconnect — /.well-known/host-meta.json advertises alt-connections
|
|
* webpresence — /status/<jid> returns an image
|
|
* admin_telnet — reachable on a non-loopback address (proves "*" bind)
|
|
"""
|
|
import asyncio
|
|
import uuid
|
|
import xml.etree.ElementTree as ET
|
|
from urllib.parse import urlparse
|
|
|
|
import pytest
|
|
|
|
|
|
def _http_base(xmpp_config):
|
|
"""Derive an http://<host>:<port> base for Prosody's HTTP port."""
|
|
for key in ("bosh_url", "ws_url"):
|
|
url = xmpp_config.get(key)
|
|
if url:
|
|
p = urlparse(url)
|
|
if p.hostname and p.port:
|
|
return f"{p.scheme or 'http'}://{p.hostname}:{p.port}", p.hostname
|
|
return None, None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestFilterChatstates:
|
|
async def test_chatstate_dropped_when_inactive(self, xmpp_client, second_client, xmpp_config):
|
|
if xmpp_client.boundjid.bare == second_client.boundjid.bare:
|
|
pytest.xfail("Need --xmpp-jid2 with a distinct account")
|
|
|
|
from slixmpp.xmlstream.handler import Callback
|
|
from slixmpp.xmlstream.matcher import MatchXPath
|
|
|
|
seen = []
|
|
|
|
def on_message(stanza):
|
|
if stanza.xml.find("{http://jabber.org/protocol/chatstates}composing") is not None:
|
|
seen.append("composing")
|
|
|
|
# slixmpp's "message" event only fires for stanzas with a <body>, but a
|
|
# chatstate-only message has none — register a low-level stream handler.
|
|
handler_name = "cs_composing_smoke"
|
|
second_client.register_handler(Callback(
|
|
handler_name,
|
|
MatchXPath(f"{{{second_client.default_ns}}}message"),
|
|
on_message,
|
|
))
|
|
try:
|
|
# CSI inactive → filter_chatstates strips chatstates; a chatstate-only
|
|
# message has no other tags and is dropped entirely.
|
|
second_client.send_raw('<inactive xmlns="urn:xmpp:csi:0"/>')
|
|
await asyncio.sleep(0.5)
|
|
|
|
m1 = xmpp_client.make_message(mto=second_client.boundjid.bare, mtype="chat")
|
|
ET.SubElement(m1.xml, "{http://jabber.org/protocol/chatstates}composing")
|
|
m1.send()
|
|
await asyncio.sleep(2)
|
|
assert not seen, "chatstate delivered to CSI-inactive client — filter_chatstates did not drop it"
|
|
|
|
# CSI active → chatstates pass through again.
|
|
second_client.send_raw('<active xmlns="urn:xmpp:csi:0"/>')
|
|
await asyncio.sleep(0.5)
|
|
|
|
m2 = xmpp_client.make_message(mto=second_client.boundjid.bare, mtype="chat")
|
|
ET.SubElement(m2.xml, "{http://jabber.org/protocol/chatstates}composing")
|
|
m2.send()
|
|
for _ in range(20):
|
|
if seen:
|
|
break
|
|
await asyncio.sleep(0.25)
|
|
assert seen, "chatstate not delivered to CSI-active client — filter_chatstates kept filtering"
|
|
finally:
|
|
second_client.remove_handler(handler_name)
|
|
second_client.send_raw('<active xmlns="urn:xmpp:csi:0"/>')
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestIdlecompat:
|
|
async def test_idle_injected_into_last_activity_presence(self, xmpp_client, second_client):
|
|
if xmpp_client.boundjid.bare == second_client.boundjid.bare:
|
|
pytest.xfail("Need --xmpp-jid2 with a distinct account")
|
|
|
|
got = asyncio.Event()
|
|
|
|
def on_presence(pres):
|
|
if pres.xml.find("{urn:xmpp:idle:1}idle") is not None:
|
|
got.set()
|
|
|
|
second_client.add_event_handler("presence", on_presence)
|
|
try:
|
|
# Directed presence carrying jabber:iq:last → mod_idlecompat must
|
|
# add an <idle xmlns='urn:xmpp:idle:1'/> child.
|
|
p = xmpp_client.make_presence(pto=second_client.boundjid.bare)
|
|
ET.SubElement(p.xml, "{jabber:iq:last}query").set("seconds", "42")
|
|
p.send()
|
|
await asyncio.wait_for(got.wait(), timeout=8)
|
|
finally:
|
|
second_client.del_event_handler("presence", on_presence)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestHttpAltconnect:
|
|
async def test_host_meta_json_advertises_alt_connections(self, xmpp_config):
|
|
import aiohttp
|
|
base, host = _http_base(xmpp_config)
|
|
if not base:
|
|
pytest.skip("No BOSH/WS URL configured")
|
|
domain = xmpp_config.get("domain") or host
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(
|
|
f"{base}/.well-known/host-meta.json",
|
|
headers={"Host": domain},
|
|
timeout=aiohttp.ClientTimeout(total=8),
|
|
) as resp:
|
|
assert resp.status == 200, f"host-meta.json returned {resp.status}"
|
|
data = await resp.json(content_type=None)
|
|
links = [l.get("rel") for l in data.get("links", [])]
|
|
# The WebSocket alt-connection is always advertised; BOSH is only
|
|
# present when mod_bosh's http URL resolves on the vhost, which depends
|
|
# on the deployment, so assert the reliable one and soft-check BOSH.
|
|
assert "urn:xmpp:alt-connections:websocket" in links, (
|
|
f"WebSocket alt-connection missing from host-meta.json: {links}"
|
|
)
|
|
assert any("alt-connections" in r for r in links), (
|
|
f"no xmpp:alt-connections advertised in host-meta.json: {links}"
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestWebpresence:
|
|
async def test_presence_endpoint_returns_image(self, xmpp_config):
|
|
import aiohttp
|
|
base, host = _http_base(xmpp_config)
|
|
if not base:
|
|
pytest.skip("No BOSH/WS URL configured")
|
|
domain = xmpp_config.get("domain") or host
|
|
jid = xmpp_config.get("jid") or f"user1@{domain}"
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(
|
|
f"{base}/status/{jid}",
|
|
headers={"Host": domain},
|
|
timeout=aiohttp.ClientTimeout(total=8),
|
|
) as resp:
|
|
assert resp.status == 200, f"/status/{jid} returned {resp.status}"
|
|
ctype = resp.headers.get("Content-Type", "")
|
|
assert ctype.startswith("image/"), (
|
|
f"webpresence returned Content-Type {ctype!r}, expected image/*"
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestAdminTelnetNonLoopback:
|
|
async def test_telnet_reachable_on_non_loopback(self, xmpp_config):
|
|
host = xmpp_config["admin_telnet_host"]
|
|
port = xmpp_config["admin_telnet_port"]
|
|
if host in ("127.0.0.1", "localhost", "::1"):
|
|
pytest.skip("ADMIN_TELNET_HOST is loopback — cannot prove wildcard bind")
|
|
try:
|
|
reader, writer = await asyncio.wait_for(
|
|
asyncio.open_connection(host, port), timeout=5
|
|
)
|
|
data = await asyncio.wait_for(reader.read(1024), timeout=5)
|
|
writer.close()
|
|
await writer.wait_closed()
|
|
except (OSError, asyncio.TimeoutError) as e:
|
|
pytest.fail(f"admin telnet not reachable on {host}:{port}: {e}")
|
|
assert b"Prosody" in data or b"\x00" in data, "no telnet banner from non-loopback address"
|