validate.py 3.0 KB

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