From 8e76f41f6317f10f16a9c350e103b2808b288e3c Mon Sep 17 00:00:00 2001 From: embeddevio Date: Wed, 28 Oct 2026 21:31:07 +0300 Subject: [PATCH] =?UTF-8?q?tests:=20=D0=BD=D0=B5=D0=B3=D0=B0=D1=82=D0=B8?= =?UTF-8?q?=D0=B2=D0=BD=D1=8B=D0=B5=20=D1=82=D0=B5=D1=81=D1=82=D1=8B=20?= =?UTF-8?q?=D0=B2=D0=B0=D0=BB=D0=B8=D0=B4=D0=B0=D1=82=D0=BE=D1=80=D0=B0=20?= =?UTF-8?q?compose-=D0=BC=D0=B0=D0=BD=D0=B8=D1=84=D0=B5=D1=81=D1=82=D0=BE?= =?UTF-8?q?=D0=B2=20(#56)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Каждая инварианта AC #21 ломается по отдельности (порт, тома /config /data, healthcheck, NET_ADMIN, /dev/net/tun, ip_forward, privileged netns-сервис, DEV-ONLY-метка, --require-dockerfile) — 39 тестов, реальные манифесты не трогаются. Refs #21, Refs #56 --- tests/deploy/test_validate_compose.py | 291 ++++++++++++++++++++++++++ 1 file changed, 291 insertions(+) create mode 100644 tests/deploy/test_validate_compose.py diff --git a/tests/deploy/test_validate_compose.py b/tests/deploy/test_validate_compose.py new file mode 100644 index 0000000..fb04594 --- /dev/null +++ b/tests/deploy/test_validate_compose.py @@ -0,0 +1,291 @@ +"""Негативные и граничные тесты tests/deploy/validate_compose.py (задача #21, issue #56). + +test_compose_manifests.py проверяет, что реальные compose-файлы стенда проходят +валидатор. Здесь — обратная сторона: каждая инварианта AC #21 ломается по +отдельности, и валидатор обязан это поймать (иначе гейт `make test-manifests` +пропускает регресс). Файлы-фикстуры создаются во временном каталоге, реальные +манифесты не трогаются. +""" + +from __future__ import annotations + +import contextlib +import io +import pathlib +import tempfile +import unittest +from unittest import mock + +import yaml + +import validate_compose as vc + +HEALTH_CMD = ["CMD-SHELL", "curl -fsS http://127.0.0.1:8080/api/v1/health >/dev/null || exit 1"] + + +def dev_service(**overrides) -> dict: + svc = { + "image": "awg-dash:dev", + "build": {"context": "..", "dockerfile": "deploy/Dockerfile"}, + "ports": ["8180:8080"], + "volumes": [ + "${AWG_DASH_HOST_DIR:-/opt/awg-dash-dev}/config:/config", + "${AWG_DASH_HOST_DIR:-/opt/awg-dash-dev}/data:/data", + ], + "devices": ["/dev/net/tun:/dev/net/tun"], + "cap_add": ["NET_ADMIN", "NET_RAW", "SYS_MODULE"], + "sysctls": {"net.ipv4.ip_forward": "1"}, + "healthcheck": {"test": HEALTH_CMD}, + "logging": {"driver": "json-file"}, + } + svc.update(overrides) + return svc + + +def dev_doc(**overrides) -> dict: + doc = {"name": "awg-dash-dev", "services": {"awg-dash": dev_service(**overrides)}} + doc.update({k: v for k, v in overrides.items() if k in ("name", "services")}) + return doc + + +def netns_doc(**overrides) -> dict: + svc = { + "image": "awg-dash:dev", + "privileged": True, + "working_dir": "/repo", + "entrypoint": ["/bin/bash", "/repo/tests/netns/run.sh"], + "volumes": ["..:/repo:ro", "/tmp/artifacts:/artifacts"], + "tmpfs": ["/tmp:mode=1777"], + } + svc.update(overrides) + return {"name": "awg-dash-test", "services": {"netns": svc}} + + +def stub_doc(**overrides) -> dict: + svc = { + "entrypoint": ["python3", "/opt/awg-dash-stub/api_stub.py"], + "command": ["--port", "8080"], + "volumes": ["../deploy/api_stub.py:/opt/awg-dash-stub/api_stub.py:ro"], + } + svc.update(overrides) + return {"services": {"awg-dash": svc}} + + +@unittest.skipUnless(vc.yaml is not None, "нужен PyYAML (make venv)") +class ValidateComposeTestCase(unittest.TestCase): + """Общая обвязка: временный «репозиторий» и подмена путей модуля.""" + + def setUp(self) -> None: + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + self.root = pathlib.Path(tmp.name) + (self.root / "deploy").mkdir() + (self.root / "tests" / "netns").mkdir(parents=True) + (self.root / "tests" / "netns" / "run.sh").write_text("#!/bin/bash\n", encoding="utf-8") + self.dev = self.root / "deploy" / "docker-compose.dev.yml" + self.test = self.root / "deploy" / "docker-compose.test.yml" + self.stub = self.root / "deploy" / "docker-compose.stub.yml" + for name, value in (("ROOT", self.root), ("DEV", self.dev), ("TEST", self.test), ("STUB", self.stub)): + patcher = mock.patch.object(vc, name, value) + patcher.start() + self.addCleanup(patcher.stop) + + def write(self, path: pathlib.Path, doc: dict | str) -> pathlib.Path: + text = doc if isinstance(doc, str) else yaml.safe_dump(doc, allow_unicode=True) + path.write_text(text, encoding="utf-8") + return path + + def errors(self, **kwargs) -> list[str]: + errors, _warnings = vc.validate(**kwargs) + return errors + + def warnings(self, **kwargs) -> list[str]: + _errors, warnings = vc.validate(**kwargs) + return warnings + + +class MountTargetTest(unittest.TestCase): + def test_env_substitution_with_default_colon(self) -> None: + svc = {"volumes": ["${AWG_DASH_HOST_DIR:-/opt/awg-dash-dev}/config:/config"]} + self.assertEqual(["/config"], vc._mount_targets(svc)) + + def test_mode_suffix_is_stripped(self) -> None: + svc = {"volumes": ["../deploy/api_stub.py:/opt/awg-dash-stub/api_stub.py:ro"]} + self.assertEqual(["/opt/awg-dash-stub/api_stub.py"], vc._mount_targets(svc)) + + def test_missing_volumes_key(self) -> None: + self.assertEqual([], vc._mount_targets({})) + self.assertEqual([], vc._volumes({})) + self.assertEqual([], vc._ports({})) + + def test_jsonish_flattens_lists(self) -> None: + self.assertEqual("a b", vc.jsonish(["a", "b"])) + self.assertEqual("1", vc.jsonish(1)) + + +class LoadTest(unittest.TestCase): + def test_load_without_pyyaml_fails_loudly(self) -> None: + with mock.patch.object(vc, "yaml", None), tempfile.NamedTemporaryFile(suffix=".yml") as tmp: + with self.assertRaises(SystemExit): + vc.load(pathlib.Path(tmp.name)) + + +class ValidateDevTest(ValidateComposeTestCase): + def test_good_manifest_has_no_errors(self) -> None: + self.write(self.dev, dev_doc()) + self.write(self.test, netns_doc()) + self.write(self.stub, stub_doc()) + (self.root / "deploy" / "api_stub.py").write_text("# DEV-ONLY\n", encoding="utf-8") + self.assertEqual([], self.errors()) + + def test_missing_file(self) -> None: + self.assertTrue(any("не найден" in e for e in self.errors())) + + def test_missing_service(self) -> None: + self.write(self.dev, {"name": "x", "services": {"other": {}}}) + self.assertTrue(any("нет сервиса awg-dash" in e for e in self.errors())) + + def test_missing_project_name(self) -> None: + doc = dev_doc() + doc.pop("name") + self.write(self.dev, doc) + self.assertTrue(any("не задан `name`" in e for e in self.errors())) + + def test_port_not_published(self) -> None: + self.write(self.dev, dev_doc(ports=["127.0.0.1:9000:9000"])) + self.assertTrue(any("порт 8080 не публикуется" in e for e in self.errors())) + + def test_state_directories_not_mounted(self) -> None: + self.write(self.dev, dev_doc(volumes=["/tmp:/tmp"])) + errors = " ".join(self.errors()) + self.assertIn("/config", errors) + self.assertIn("/data", errors) + self.assertIn("AC-2", errors) + + def test_healthcheck_must_hit_api_v1_health(self) -> None: + self.write( + self.dev, + dev_doc(healthcheck={"test": ["CMD-SHELL", "curl -fsS http://127.0.0.1:8080/ || exit 1"]}), + ) + self.assertTrue(any("/api/v1/health" in e for e in self.errors())) + + def test_healthcheck_absent(self) -> None: + self.write(self.dev, dev_doc(healthcheck=None)) + self.assertTrue(any("healthcheck не ходит" in e for e in self.errors())) + + def test_requires_net_admin(self) -> None: + self.write(self.dev, dev_doc(cap_add=["NET_RAW"])) + self.assertTrue(any("NET_ADMIN" in e for e in self.errors())) + + def test_requires_tun_device(self) -> None: + self.write(self.dev, dev_doc(devices=["/dev/null:/dev/null"])) + self.assertTrue(any("/dev/net/tun" in e for e in self.errors())) + + def test_requires_ip_forward(self) -> None: + self.write(self.dev, dev_doc(sysctls={})) + self.assertTrue(any("ip_forward" in e for e in self.errors())) + + def test_ip_forward_accepts_boolean_true(self) -> None: + self.write(self.dev, dev_doc(sysctls={"net.ipv4.ip_forward": True})) + self.assertFalse(any("ip_forward" in e for e in self.errors())) + + def test_unbounded_logging_is_a_warning(self) -> None: + self.write(self.dev, dev_doc(logging=None)) + self.assertTrue(any("json-file" in w for w in self.warnings())) + + def test_missing_dockerfile_is_warning_before_issue_18(self) -> None: + self.write(self.dev, dev_doc()) + self.assertTrue(any("dockerfile" in w for w in self.warnings())) + + def test_missing_dockerfile_is_error_when_required(self) -> None: + self.write(self.dev, dev_doc()) + self.assertTrue(any("dockerfile" in e for e in self.errors(require_dockerfile=True))) + + def test_present_dockerfile_produces_no_dockerfile_message(self) -> None: + (self.root / "deploy" / "Dockerfile").write_text("FROM scratch\n", encoding="utf-8") + self.write(self.dev, dev_doc()) + self.assertFalse(any("dockerfile" in m for m in self.errors(require_dockerfile=True))) + + +class ValidateTestTest(ValidateComposeTestCase): + def test_missing_file(self) -> None: + self.write(self.dev, dev_doc()) + self.assertTrue(any("docker-compose.test.yml: файл не найден" in e for e in self.errors())) + + def test_missing_netns_service(self) -> None: + self.write(self.test, {"services": {"app": {}}}) + self.assertTrue(any("нет сервиса netns" in e for e in self.errors())) + + def test_netns_must_be_privileged(self) -> None: + self.write(self.test, netns_doc(privileged=False)) + self.assertTrue(any("не privileged" in e for e in self.errors())) + + def test_entrypoint_must_point_to_run_sh(self) -> None: + self.write(self.test, netns_doc(entrypoint=["python3", "-m", "http.server"])) + self.assertTrue(any("entrypoint" in e for e in self.errors())) + + def test_missing_run_sh_is_an_error(self) -> None: + (self.root / "tests" / "netns" / "run.sh").unlink() + self.write(self.test, netns_doc()) + self.assertTrue(any("нет файла tests/netns/run.sh" in e for e in self.errors())) + + def test_artifacts_and_tmpfs_are_warnings(self) -> None: + self.write(self.test, netns_doc(volumes=["..:/repo:ro"], tmpfs=[])) + warnings = " ".join(self.warnings()) + self.assertIn("/artifacts", warnings) + self.assertIn("/tmp", warnings) + + +class ValidateStubTest(ValidateComposeTestCase): + def test_missing_stub_is_a_warning(self) -> None: + self.assertTrue(any("health-заглушкой отсутствует" in w for w in self.warnings())) + + def test_missing_service(self) -> None: + self.write(self.stub, {"services": {"other": {}}}) + self.assertTrue(any("stub: нет сервиса awg-dash" in e for e in self.errors())) + + def test_entrypoint_must_point_to_api_stub(self) -> None: + self.write(self.stub, stub_doc(entrypoint=["/bin/sh"])) + self.assertTrue(any("api_stub.py" in e for e in self.errors())) + + def test_missing_api_stub_file_is_an_error(self) -> None: + self.write(self.stub, stub_doc()) + self.assertTrue(any("нет файла deploy/api_stub.py" in e for e in self.errors())) + + def test_dev_only_marker_is_a_warning(self) -> None: + (self.root / "deploy" / "api_stub.py").write_text("# заглушка\n", encoding="utf-8") + self.write(self.stub, stub_doc()) + self.assertTrue(any("DEV-ONLY" in w for w in self.warnings())) + + +class MainTest(ValidateComposeTestCase): + def test_ok_manifest_returns_zero(self) -> None: + self.write(self.dev, dev_doc()) + self.write(self.test, netns_doc()) + self.write(self.stub, stub_doc()) + (self.root / "deploy" / "api_stub.py").write_text("# DEV-ONLY\n", encoding="utf-8") + out = io.StringIO() + with contextlib.redirect_stdout(out): + rc = vc.main([]) + self.assertEqual(0, rc) + self.assertIn("ok (", out.getvalue()) + + def test_errors_return_one_and_are_printed(self) -> None: + out = io.StringIO() + with contextlib.redirect_stdout(out): + rc = vc.main(["--require-dockerfile"]) + self.assertEqual(1, rc) + self.assertIn("[error]", out.getvalue()) + self.assertIn("ошибок", out.getvalue()) + + def test_quiet_suppresses_details_but_keeps_exit_code(self) -> None: + out = io.StringIO() + with contextlib.redirect_stdout(out): + rc = vc.main(["--quiet"]) + self.assertEqual(1, rc) + self.assertNotIn("[error]", out.getvalue()) + self.assertIn("ошибок", out.getvalue()) + + +if __name__ == "__main__": + unittest.main(verbosity=2)