embeddevio
6de43622fa
- 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
70 lines
2.6 KiB
Python
70 lines
2.6 KiB
Python
#!/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())
|