mirror of
https://git.quad4.io/RNS-Things/MeshChatX.git
synced 2026-05-29 20:14:18 +00:00
feat(tests): add HTTP API contract tests and helpers for route validation
This commit is contained in:
@@ -0,0 +1,70 @@
|
||||
"""Helpers for HTTP API route contract checks (meshchat.py vs frontend)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
_ROUTE_DECORATOR = re.compile(
|
||||
r'@routes\.(get|post|patch|delete|put)\(\s*(?:\n\s*)?["\']([^"\']+)["\']',
|
||||
re.MULTILINE,
|
||||
)
|
||||
|
||||
|
||||
def extract_meshchat_http_routes(meshchat_py: Path) -> list[dict[str, str]]:
|
||||
text = meshchat_py.read_text(encoding="utf-8")
|
||||
rows: list[dict[str, str]] = []
|
||||
for m in _ROUTE_DECORATOR.finditer(text):
|
||||
rows.append({"method": m.group(1).upper(), "path": m.group(2)})
|
||||
rows.sort(key=lambda x: (x["path"], x["method"]))
|
||||
return rows
|
||||
|
||||
|
||||
def path_matches_aiohttp_route(route: str, path: str) -> bool:
|
||||
pattern = ""
|
||||
i = 0
|
||||
while i < len(route):
|
||||
if route[i] == "{":
|
||||
j = route.find("}", i)
|
||||
if j == -1:
|
||||
return False
|
||||
pattern += "[^/]+"
|
||||
i = j + 1
|
||||
else:
|
||||
pattern += re.escape(route[i])
|
||||
i += 1
|
||||
return re.fullmatch(pattern, path) is not None
|
||||
|
||||
|
||||
def extract_frontend_api_paths(frontend_root: Path) -> set[str]:
|
||||
out: set[str] = set()
|
||||
for path in list(frontend_root.rglob("*.vue")) + list(frontend_root.rglob("*.js")):
|
||||
if "node_modules" in path.parts:
|
||||
continue
|
||||
text = path.read_text(encoding="utf-8")
|
||||
for m in re.finditer(r"`(/api/v1[^`]+)`", text):
|
||||
s = m.group(1).split("?")[0]
|
||||
s = re.sub(r"\$\{[^}]+\}", "a", s)
|
||||
out.add(s)
|
||||
for m in re.finditer(r'["\'](/api/v1[^"\']+)["\']', text):
|
||||
s = m.group(1).split("?")[0]
|
||||
if "${" in s:
|
||||
continue
|
||||
out.add(s)
|
||||
return out
|
||||
|
||||
|
||||
def load_route_fixture(fixture_path: Path) -> list[dict[str, str]]:
|
||||
data = json.loads(fixture_path.read_text(encoding="utf-8"))
|
||||
routes = data["routes"]
|
||||
routes.sort(key=lambda x: (x["path"], x["method"]))
|
||||
return routes
|
||||
|
||||
|
||||
def write_route_fixture(fixture_path: Path, routes: list[dict[str, str]]) -> None:
|
||||
fixture_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
fixture_path.write_text(
|
||||
json.dumps({"routes": routes}, indent=2) + "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
@@ -0,0 +1,42 @@
|
||||
"""Contract tests: aiohttp routes in meshchat.py vs checked-in manifest; frontend /api/v1 usage."""
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from tests.backend.http_api_contract_helpers import (
|
||||
extract_frontend_api_paths,
|
||||
extract_meshchat_http_routes,
|
||||
load_route_fixture,
|
||||
path_matches_aiohttp_route,
|
||||
write_route_fixture,
|
||||
)
|
||||
|
||||
_REPO_ROOT = Path(__file__).resolve().parents[2]
|
||||
_MESHCHAT_PY = _REPO_ROOT / "meshchatx" / "meshchat.py"
|
||||
_FRONTEND_ROOT = _REPO_ROOT / "meshchatx" / "src" / "frontend"
|
||||
_FIXTURE = Path(__file__).resolve().parent / "fixtures" / "http_api_routes.json"
|
||||
|
||||
|
||||
def test_meshchat_http_routes_match_fixture():
|
||||
live = extract_meshchat_http_routes(_MESHCHAT_PY)
|
||||
if os.environ.get("UPDATE_HTTP_API_ROUTES") == "1":
|
||||
write_route_fixture(_FIXTURE, live)
|
||||
pytest.skip("UPDATE_HTTP_API_ROUTES=1: fixture updated; re-run without the env var")
|
||||
expected = load_route_fixture(_FIXTURE)
|
||||
assert live == expected, (
|
||||
"HTTP route list drifted. If you added or renamed routes, run: "
|
||||
"UPDATE_HTTP_API_ROUTES=1 poetry run pytest tests/backend/test_http_api_contract.py -k "
|
||||
"meshchat_http_routes_match_fixture"
|
||||
)
|
||||
|
||||
|
||||
def test_frontend_api_paths_exist_on_backend():
|
||||
backend_paths = [r["path"] for r in extract_meshchat_http_routes(_MESHCHAT_PY)]
|
||||
frontend_paths = extract_frontend_api_paths(_FRONTEND_ROOT)
|
||||
missing = []
|
||||
for fp in sorted(frontend_paths):
|
||||
if not any(path_matches_aiohttp_route(br, fp) for br in backend_paths):
|
||||
missing.append(fp)
|
||||
assert not missing, f"Frontend references unknown HTTP paths: {missing}"
|
||||
@@ -0,0 +1,42 @@
|
||||
"""
|
||||
Optional live Reticulum smoke test.
|
||||
|
||||
Reticulum is a process-wide singleton; this test runs a short script in a
|
||||
subprocess so it does not interfere with other tests.
|
||||
|
||||
Enable with: MESHCHAT_LIVE_RETICULUM=1
|
||||
|
||||
This does not replace multi-node mesh interoperability testing; it only
|
||||
verifies that a default-config Reticulum instance can start and exit cleanly
|
||||
in an isolated config directory (useful after RNS upgrades or OS changes).
|
||||
"""
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
import pytest
|
||||
|
||||
_RUN = os.environ.get("MESHCHAT_LIVE_RETICULUM") == "1"
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1 to run live Reticulum subprocess smoke test")
|
||||
def test_reticulum_subprocess_start_and_exit():
|
||||
script = r"""
|
||||
import tempfile
|
||||
import RNS
|
||||
|
||||
tmpdir = tempfile.mkdtemp(prefix="meshchat_rns_live_")
|
||||
try:
|
||||
RNS.Reticulum(configdir=tmpdir, loglevel=RNS.LOG_ERROR)
|
||||
finally:
|
||||
RNS.exit(0)
|
||||
"""
|
||||
result = subprocess.run(
|
||||
[sys.executable, "-c", script],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=120,
|
||||
)
|
||||
assert result.returncode == 0, result.stderr + result.stdout
|
||||
Reference in New Issue
Block a user