validate.py 3.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990
  1. """Docker Compose validation functionality."""
  2. from __future__ import annotations
  3. import logging
  4. import shutil
  5. import subprocess
  6. import tempfile
  7. from pathlib import Path
  8. from ...core.validation import KindValidationFailure, KindValidationResult
  9. logger = logging.getLogger(__name__)
  10. class ComposeDockerValidator:
  11. """Kind-specific validator backed by Docker Compose."""
  12. validator_name = "docker compose config"
  13. unavailable_message = "Required command is unavailable: docker compose"
  14. def __init__(self, verbose: bool = False) -> None:
  15. self.verbose = verbose
  16. self._available: bool | None = None
  17. def is_available(self) -> bool:
  18. """Check whether Docker Compose is available locally."""
  19. if self._available is not None:
  20. return self._available
  21. if shutil.which("docker") is None:
  22. self._available = False
  23. return self._available
  24. result = subprocess.run(
  25. ["docker", "compose", "version"],
  26. capture_output=True,
  27. text=True,
  28. check=False,
  29. )
  30. self._available = result.returncode == 0
  31. return self._available
  32. def validate_rendered_files(self, rendered_files: dict[str, str], _case_name: str) -> KindValidationResult:
  33. """Validate rendered Compose files with Docker Compose."""
  34. result = KindValidationResult(validator=self.validator_name, available=self.is_available())
  35. if not result.available:
  36. result.details.append(self.unavailable_message)
  37. return result
  38. compose_files = _find_compose_files(rendered_files)
  39. if not compose_files:
  40. result.warnings.append("No Docker Compose files found")
  41. return result
  42. for filename, content in compose_files:
  43. failure = self._validate_compose_content(filename, content)
  44. if failure is not None:
  45. result.failures.append(failure)
  46. return result
  47. def _validate_compose_content(self, filename: str, content: str) -> KindValidationFailure | None:
  48. with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as tmp_file:
  49. tmp_file.write(content)
  50. tmp_path = tmp_file.name
  51. try:
  52. result = subprocess.run(
  53. ["docker", "compose", "-f", tmp_path, "config", "--quiet"],
  54. capture_output=True,
  55. text=True,
  56. check=False,
  57. )
  58. finally:
  59. Path(tmp_path).unlink(missing_ok=True)
  60. if result.returncode == 0:
  61. return None
  62. message = result.stderr.strip() or result.stdout.strip() or "Docker Compose validation failed"
  63. return KindValidationFailure(file_path=filename, validator=self.validator_name, message=message)
  64. def _find_compose_files(rendered_files: dict[str, str]) -> list[tuple[str, str]]:
  65. return [
  66. (filename, content)
  67. for filename, content in rendered_files.items()
  68. if filename.endswith(("compose.yaml", "compose.yml", "docker-compose.yaml", "docker-compose.yml"))
  69. ]