mirror of
https://github.com/agessaman/meshcore-bot.git
synced 2026-10-27 08:12:20 +00:00
test(characterization): pin packaging coverage and the web viewer route table
The packaging test checks pyproject.toml against git ls-files: every directory under modules/ with Python files must be a regular package, and every tracked runtime data file must match a package-data glob, since include-package-data is off and a new subpackage or data file would otherwise silently miss the wheel. The endpoint inventory records all 104 routes with their methods and endpoint names, plus the status, content type, redirect target, script-src CSP (nonce normalized) and framing headers of anonymous requests with and without a password, and of anonymous mutating requests when locked. Endpoint names feed the CSP nonce allowlist and url_for, so a helper extraction that renames or reroutes anything shows up here.
This commit is contained in:
1 parent
eda2f12e1d
commit
f829ed9ddd
5 files changed
+2048
-2
No files matched your search
File diff suppressed because it is too large.
Load diff
@@ -53,7 +53,7 @@ def _diff(exp: Any, got: Any, path: str = "") -> list[str]:
|
||||
if len(exp) != len(got):
|
||||
return [f"{path}: length {len(exp)} -> {len(got)}"]
|
||||
out = []
|
||||
for i, (a, b) in enumerate(zip(exp, got)):
|
||||
for i, (a, b) in enumerate(zip(exp, got, strict=True)):
|
||||
out.extend(_diff(a, b, f"{path}[{i}]"))
|
||||
return out
|
||||
return [] if exp == got else [f"{path}: {exp!r} -> {got!r}"]
|
||||
@@ -35,7 +35,7 @@ class _Cmd(BaseCommand):
|
||||
def _shapes():
|
||||
yield "self_info=None", None
|
||||
for combo in itertools.product(VALUES, repeat=len(FIELDS)):
|
||||
present = {f: (f"{v}-{f}" if v == "Dev" else v) for f, v in zip(FIELDS, combo) if v is not None}
|
||||
present = {f: (f"{v}-{f}" if v == "Dev" else v) for f, v in zip(FIELDS, combo, strict=True) if v is not None}
|
||||
label = ",".join(f"{k}={v!r}" for k, v in present.items()) or "empty"
|
||||
yield f"dict[{label}]", dict(present)
|
||||
yield f"obj[{label}]", SimpleNamespace(**present)
|
||||
|
||||
@@ -0,0 +1,69 @@
|
||||
"""Every tracked runtime file must ship in the wheel.
|
||||
|
||||
``pyproject.toml`` uses ``include-package-data = false`` and lists package data
|
||||
explicitly, and package discovery is the regular (non-namespace) kind, so a new
|
||||
subpackage without ``__init__.py`` or a new data file outside the globs is
|
||||
silently left out of an installed bot. This checks the configuration against
|
||||
``git ls-files`` without building a wheel (the dev venv has no setuptools).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import fnmatch
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
tomllib = pytest.importorskip("tomllib") # Python 3.11+; CI's 3.11-3.13 jobs cover it
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
|
||||
|
||||
def _tracked(prefix: str) -> list[str]:
|
||||
try:
|
||||
out = subprocess.run(
|
||||
["git", "ls-files", prefix], cwd=ROOT, capture_output=True, text=True, check=True
|
||||
).stdout
|
||||
except (OSError, subprocess.CalledProcessError):
|
||||
pytest.skip("not a git checkout")
|
||||
return [line for line in out.splitlines() if line]
|
||||
|
||||
|
||||
def _config():
|
||||
return tomllib.loads((ROOT / "pyproject.toml").read_text(encoding="utf-8"))["tool"]["setuptools"]
|
||||
|
||||
|
||||
def test_every_python_package_directory_is_a_regular_package():
|
||||
missing = sorted(
|
||||
{
|
||||
str(Path(f).parent)
|
||||
for f in _tracked("modules")
|
||||
if f.endswith(".py") and not (ROOT / Path(f).parent / "__init__.py").exists()
|
||||
}
|
||||
)
|
||||
assert missing == [], f"directories with modules but no __init__.py (not packaged): {missing}"
|
||||
|
||||
|
||||
def test_every_non_python_runtime_file_matches_a_package_data_glob():
|
||||
package_data: dict[str, list[str]] = _config()["package-data"]
|
||||
uncovered = []
|
||||
for prefix, package in (("modules", "modules"), ("translations", "translations"), ("data/randomlines", "data.randomlines")):
|
||||
globs = package_data.get(package, [])
|
||||
base = ROOT / package.replace(".", "/")
|
||||
for f in _tracked(prefix):
|
||||
if f.endswith((".py", ".pyc", ".md")) or "/__pycache__/" in f:
|
||||
continue
|
||||
rel = (ROOT / f).relative_to(base).as_posix()
|
||||
# setuptools globs: '*' does not cross '/', '**' does.
|
||||
if not any(_glob_match(rel, g) for g in globs):
|
||||
uncovered.append(f)
|
||||
assert uncovered == [], f"runtime files not covered by [tool.setuptools.package-data]: {uncovered}"
|
||||
|
||||
|
||||
def _glob_match(rel: str, pattern: str) -> bool:
|
||||
if "**" in pattern:
|
||||
head, _, tail = pattern.partition("**/")
|
||||
return rel.startswith(head) and fnmatch.fnmatch(rel[len(head):].split("/")[-1], tail) if tail else rel.startswith(head)
|
||||
rel_parts, pat_parts = rel.split("/"), pattern.split("/")
|
||||
return len(rel_parts) == len(pat_parts) and all(fnmatch.fnmatch(r, p) for r, p in zip(rel_parts, pat_parts, strict=True))
|
||||
@@ -0,0 +1,94 @@
|
||||
"""Characterization: the web viewer's route table and how each route answers.
|
||||
|
||||
Records every rule (path, methods, endpoint name) and, for parameterless GET
|
||||
routes, the status, content type and security-relevant headers of a request
|
||||
made anonymously with and without a password configured. It also records, for
|
||||
every mutating route, what an anonymous request gets when a password is set.
|
||||
Helper extraction or route regrouping in ``app.py`` must leave this unchanged:
|
||||
endpoint names feed the CSP nonce allowlist and ``url_for``, and auth is
|
||||
path-based.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import configparser
|
||||
import re
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from tests.characterization.golden_util import assert_golden
|
||||
|
||||
SKIP_GET = {
|
||||
# Streams or long-polls; not a single request/response.
|
||||
"/socket.io/",
|
||||
}
|
||||
|
||||
|
||||
def _viewer(tmp_path, password: str):
|
||||
from modules.web_viewer.app import BotDataViewer
|
||||
|
||||
tmp_path.mkdir(parents=True, exist_ok=True)
|
||||
config = configparser.ConfigParser()
|
||||
config.add_section("Bot")
|
||||
config.set("Bot", "db_path", str(tmp_path / "meshcore_bot.db"))
|
||||
config.set("Bot", "bot_name", "InvBot")
|
||||
config.add_section("Web_Viewer")
|
||||
for key, value in [
|
||||
("host", "127.0.0.1"), ("port", "8080"), ("enabled", "false"),
|
||||
("auto_start", "false"), ("debug", "false"),
|
||||
("cors_allowed_origins", "*"), ("web_viewer_password", password),
|
||||
]:
|
||||
config.set("Web_Viewer", key, value)
|
||||
config_path = str(tmp_path / "config.ini")
|
||||
with open(config_path, "w") as handle:
|
||||
config.write(handle)
|
||||
with patch.object(BotDataViewer, "_start_database_polling"), \
|
||||
patch.object(BotDataViewer, "_start_log_tailing"), \
|
||||
patch.object(BotDataViewer, "_start_cleanup_scheduler"), \
|
||||
patch.object(BotDataViewer, "_start_dashboard_refresher"), \
|
||||
patch.object(BotDataViewer, "_setup_socketio_handlers"), \
|
||||
patch("modules.web_viewer.app.RepeaterManager"):
|
||||
viewer = BotDataViewer(db_path=str(tmp_path / "meshcore_bot.db"), config_path=config_path)
|
||||
viewer.app.testing = True
|
||||
return viewer
|
||||
|
||||
|
||||
def _headers(resp):
|
||||
csp = resp.headers.get("Content-Security-Policy", "")
|
||||
script_src = next((p.strip() for p in csp.split(";") if p.strip().startswith("script-src")), "")
|
||||
return {
|
||||
"status": resp.status_code,
|
||||
"content_type": (resp.headers.get("Content-Type") or "").split(";")[0],
|
||||
"location": re.sub(r"https?://[^/]+", "", resp.headers.get("Location", "")),
|
||||
"csp_script_src": re.sub(r"'nonce-[^']+'", "'nonce-X'", script_src),
|
||||
"x_frame_options": resp.headers.get("X-Frame-Options"),
|
||||
"x_content_type_options": resp.headers.get("X-Content-Type-Options"),
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.timeout(120)
|
||||
def test_route_table_and_anonymous_responses(tmp_path):
|
||||
open_viewer = _viewer(tmp_path / "open", "")
|
||||
locked_viewer = _viewer(tmp_path / "locked", "s3cret-pass")
|
||||
rules = sorted(open_viewer.app.url_map.iter_rules(), key=lambda r: (r.rule, r.endpoint))
|
||||
inventory = {}
|
||||
open_client = open_viewer.app.test_client()
|
||||
locked_client = locked_viewer.app.test_client()
|
||||
for rule in rules:
|
||||
if rule.endpoint == "static":
|
||||
continue
|
||||
methods = sorted(m for m in rule.methods if m not in ("HEAD", "OPTIONS"))
|
||||
entry = {"endpoint": rule.endpoint, "methods": methods}
|
||||
has_params = bool(rule.arguments)
|
||||
if "GET" in methods and not has_params and rule.rule not in SKIP_GET:
|
||||
entry["GET open"] = _headers(open_client.get(rule.rule))
|
||||
entry["GET locked anon"] = _headers(locked_client.get(rule.rule))
|
||||
for method in methods:
|
||||
if method == "GET":
|
||||
continue
|
||||
path = re.sub(r"<(?:[^:>]+:)?[^>]+>", "1", rule.rule)
|
||||
resp = locked_client.open(path, method=method, json={})
|
||||
entry[f"{method} locked anon"] = _headers(resp)
|
||||
inventory[f"{rule.rule}"] = entry
|
||||
assert_golden("web_endpoint_inventory", inventory)
|
||||
Reference in new issue
Block a user