awg-dash/tests/deploy/test_validate_compose.py

"""Негативные и граничные тесты 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)