tests: негативные тесты валидатора compose-манифестов (#56)
Каждая инварианта AC #21 ломается по отдельности (порт, тома /config /data, healthcheck, NET_ADMIN, /dev/net/tun, ip_forward, privileged netns-сервис, DEV-ONLY-метка, --require-dockerfile) — 39 тестов, реальные манифесты не трогаются. Refs #21, Refs #56
This commit is contained in:
@@ -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)
|
||||
Reference in New Issue
Block a user