test(#21): утилиты стенда, раннер с JSON-отчётом и README харнесса

- netns_common.py — обёртки над ip/nft/iptables: проверка netns и линков, счётчики правил меток и kill-switch (nft -j и iptables -v -x), чтение JSONL-логов мока и детектора, отправка трафика через probe.py; - run_tests.py — прогон unittest с JSON-отчётом (pytest не нужен: работает и на хосте, и в alpine-образе); - README.md — схема топологии, таблица тестов, переменные окружения и pitfalls (root/CAP_NET_ADMIN, rp_filter, отсутствие nft и ping в образе). Refs #21 Refs #15
2026-10-21 23:48:22 +03:00
parent 18b911d582
commit 6de43622fa
2 changed files with 317 additions and 0 deletions
@@ -0,0 +1,69 @@
+#!/usr/bin/env python3
+"""run_tests.py — прогон netns-тестов с JSON-отчётом (задача #21). Только stdlib.
+
+Запускается из tests/netns/run.sh. Не требует pytest: unittest есть в stdlib,
+поэтому стенд работает и на хосте, и в alpine-образе приложения.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import pathlib
+import sys
+import time
+import unittest
+
+
+def main(argv: list[str] | None = None) -> int:
+    ap = argparse.ArgumentParser(description="Прогон netns-тестов awg-dash (#21)")
+    ap.add_argument("--report", default=os.environ.get("AWG_NETNS_REPORT", ""))
+    ap.add_argument("--pattern", default="test_*.py")
+    ap.add_argument("--verbosity", type=int, default=2)
+    args = ap.parse_args(argv)
+
+    here = pathlib.Path(__file__).resolve().parent
+    # тесты стенда пропускаются вне run.sh (нужны root + поднятый стенд)
+    os.environ.setdefault("AWG_NETNS_RUN", "1")
+
+    loader = unittest.TestLoader()
+    loader.sortTestMethodsUsing = staticmethod(lambda a, b: (a > b) - (a < b))  # стабильный алфавитный порядок
+    suite = loader.discover(str(here), pattern=args.pattern, top_level_dir=str(here))
+
+    started = time.time()
+    result = unittest.TextTestRunner(verbosity=args.verbosity).run(suite)
+    duration = round(time.time() - started, 3)
+
+    details = [
+        {"test": str(test), "status": "failed", "message": message}
+        for test, message in list(result.failures) + list(result.errors)
+    ]
+    details += [{"test": str(test), "status": "skipped", "message": message} for test, message in result.skipped]
+
+    payload = {
+        "suite": "tests/netns",
+        "tests": result.testsRun,
+        "passed": result.testsRun - len(result.failures) - len(result.errors) - len(result.skipped),
+        "failures": len(result.failures),
+        "errors": len(result.errors),
+        "skipped": len(result.skipped),
+        "duration_s": duration,
+        "details": details,
+        "returncode": 0 if result.wasSuccessful() else 1,
+    }
+
+    if args.report:
+        path = pathlib.Path(args.report)
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+
+    print(
+        "[netns] тесты: {tests} всего, {passed} ok, {failures} fail, {errors} err, "
+        "{skipped} skip, {duration_s}s".format(**payload)
+    )
+    return int(payload["returncode"])
+
+
+if __name__ == "__main__":
+    sys.exit(main())