Sprint 0 / #21: dev/staging стенд — docker-compose, тома, netns-харнесс #53

Merged
embeddevio merged 30 commits from feat/21-dev-stand into main 2026-10-28 22:12:03 +03:00
2 changed files with 33 additions and 7 deletions
Showing only changes of commit c7f2ea128e - Show all commits
+17 -2
View File
@@ -15,9 +15,12 @@ DASH_PORT ?= 8180
NETNS_ARGS ?=
VERIFY_ARGS ?=
MANIFEST_ARGS ?=
# Режим входа в netns внутри тестового контейнера: auto|ip|nsenter (issue #56).
NETNS_EXEC_MODE ?= auto
.PHONY: help venv lint fmt test build dev-up dev-up-app dev-up-stub dev-down dev-logs dev-verify \
dev-verify-stub dev-reset test-netns test-netns-keep test-netns-ci test-manifests lint-shell
dev-verify-stub dev-reset test-netns test-netns-keep test-netns-ci test-manifests lint-shell \
test-deploy
help:
@grep -E '^[a-zA-Z_-]+:' $(MAKEFILE_LIST) | sed 's/:.*//' | sort -u
@@ -106,13 +109,25 @@ test-netns-keep:
bash tests/netns/run.sh --keep-topology --keep-logs $(NETNS_ARGS)
# Тот же стенд в привилегированном контейнере — путь для CI (#17).
# Внутри контейнера `ip netns exec` падает с "mount of /sys failed: Operation
# not permitted" (iproute2 6.x), поэтому харнесс сам переключается на
# `nsenter --net=/var/run/netns/<ns>` (issue #56). AWG_NS_EXEC_MODE форсирует
# режим: ip|nsenter|auto.
test-netns-ci:
AWG_DASH_IMAGE=$(IMAGE) $(COMPOSE) -f $(COMPOSE_TEST) run --rm netns
AWG_DASH_IMAGE=$(IMAGE) AWG_NS_EXEC_MODE=$(NETNS_EXEC_MODE) $(COMPOSE) -f $(COMPOSE_TEST) run --rm netns
# --- статические проверки манифестов и скриптов ------------------------------
test-manifests:
$(PY) tests/deploy/validate_compose.py $(MANIFEST_ARGS)
# Тесты стенда с покрытием: гейт DoD «тесты ≥70 %» для #21/#56.
# Отдельный tests/deploy/pytest.ini обязателен: без него `pytest tests/deploy`
# наследует addopts корневого pyproject.toml (--cov=awg_agent --cov=awg_api),
# покрытие выходит 0.00 % и RC=1 — гейт приходилось гонять с --no-cov.
test-deploy:
@test -d tests/deploy || { echo 'tests/deploy ещё нет — задача #21'; exit 1; }
$(PY) -m pytest tests/deploy -c tests/deploy/pytest.ini
lint-shell:
@command -v shellcheck >/dev/null 2>&1 || { echo 'shellcheck не установлен (apt-get install shellcheck)'; exit 1; }
shellcheck -x -S style tests/netns/*.sh deploy/*.sh
+16 -5
View File
@@ -25,6 +25,8 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
STARTED_AT = time.time()
STUB_ENV = os.environ.get("AWG_API_STUB", "0")
HEALTH_PATHS = ("/api/v1/health", "/healthz")
class Handler(BaseHTTPRequestHandler):
server_version = "awg-dash-stub/0.1"
@@ -40,7 +42,7 @@ class Handler(BaseHTTPRequestHandler):
def do_GET(self) -> None: # noqa: N802 (HTTP-метод)
path = self.path.split("?", 1)[0]
if path in ("/api/v1/health", "/healthz"):
if path in HEALTH_PATHS:
self._json(
200,
{
@@ -56,7 +58,7 @@ class Handler(BaseHTTPRequestHandler):
self._json(404, {"error": "not_implemented", "issue": "#31", "path": path})
def do_HEAD(self) -> None: # noqa: N802
self.send_response(200 if self.path.split("?", 1)[0] in ("/api/v1/health", "/healthz") else 404)
self.send_response(200 if self.path.split("?", 1)[0] in HEALTH_PATHS else 404)
self.send_header("Content-Length", "0")
self.end_headers()
@@ -64,19 +66,28 @@ class Handler(BaseHTTPRequestHandler):
print(f"[api-stub] {self.address_string()} {fmt % args}", flush=True)
def main() -> int:
def make_server(host: str, port: int) -> ThreadingHTTPServer:
"""HTTP-сервер заглушки.
Вынесен из main(), чтобы тесты (tests/deploy/test_api_stub.py) поднимали
настоящий HTTP-сервер на свободном порту (port=0) и проверяли коды ответов.
"""
return ThreadingHTTPServer((host, port), Handler)
def main(argv: list[str] | None = None) -> int:
ap = argparse.ArgumentParser(description="DEV-ONLY заглушка health для dev-стенда (#21)")
ap.add_argument("--host", default="0.0.0.0")
ap.add_argument("--port", type=int, default=8080)
ap.add_argument("--domain", default="/config", help="каталог, доступность которого проверяем (для лога)")
args = ap.parse_args()
args = ap.parse_args(argv)
print(
f"[api-stub] DEV-ONLY заглушка на {args.host}:{args.port}; каталог {args.domain} "
f"{'доступен' if os.path.isdir(args.domain) else 'НЕ доступен'}",
flush=True,
)
ThreadingHTTPServer((args.host, args.port), Handler).serve_forever()
make_server(args.host, args.port).serve_forever()
return 0