embeddevio
8e76f41f63
Каждая инварианта AC #21 ломается по отдельности (порт, тома /config /data, healthcheck, NET_ADMIN, /dev/net/tun, ip_forward, privileged netns-сервис, DEV-ONLY-метка, --require-dockerfile) — 39 тестов, реальные манифесты не трогаются. Refs #21, Refs #56
292 lines
12 KiB
Python
292 lines
12 KiB
Python
"""Негативные и граничные тесты 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)
|