""" Docker container checks module """ from typing import Dict, Optional from tests.utils.logger import Logger from tests.utils.docker_utils import DockerUtils class DockerChecks: """Check and manage Docker containers""" def __init__(self, logger: Logger, project_dir: str, version: str = "unknown"): self.logger = logger self.version = version self.docker_utils = DockerUtils(logger, version=version) self.docker_utils.set_project_dir(project_dir) self.containers = [ "postgres", "redis", "backend", "daemon", "frontend", "caddy", "nginx", "backup" ] def run_all_checks(self, auto_fix: bool = True) -> bool: """Run all Docker checks and optionally fix issues""" self.logger.subheader(f"Docker Checks (v{self.version})") # Check Docker installation if not self._check_docker_installation(): return False # Check Docker Compose installation if not self._check_docker_compose_installation(): return False # Check container statuses if not self._check_container_statuses(auto_fix): return False # Check network if not self._check_network(): return False # Check volumes if not self._check_volumes(): return False self.logger.success("DOCKER", "All Docker checks passed") return True def _check_docker_installation(self) -> bool: """Check if Docker is installed""" self.logger.info("DOCKER", "Checking Docker installation...") if self.docker_utils.check_docker_installed(): self.logger.success("DOCKER", "Docker is installed") return True else: self.logger.error("DOCKER", "Docker is not installed or not accessible") return False def _check_docker_compose_installation(self) -> bool: """Check if Docker Compose is installed""" self.logger.info("DOCKER", "Checking Docker Compose installation...") if self.docker_utils.check_docker_compose_installed(): self.logger.success("DOCKER", "Docker Compose is installed") return True else: self.logger.error("DOCKER", "Docker Compose is not installed or not accessible") return False def _check_container_statuses(self, auto_fix: bool) -> bool: """Check status of all containers and optionally fix issues""" self.logger.info("DOCKER", "Checking container statuses...") statuses = self.docker_utils.get_all_container_statuses() all_ok = True for container in self.containers: status = statuses.get(container, "not_found") if status == "running": self.logger.success("DOCKER", f"{container}: {status}") elif status == "not_found": self.logger.error("DOCKER", f"{container}: not found - container may not exist") all_ok = False if auto_fix: self.logger.info("DOCKER", f"Attempting to start containers with docker-compose up...") if self.docker_utils.compose_up(): self.logger.success("DOCKER", "Containers started successfully") else: self.logger.error("DOCKER", "Failed to start containers") return False else: self.logger.warning("DOCKER", f"{container}: {status}") if auto_fix: self.logger.info("DOCKER", f"Attempting to fix {container}...") # Get actual container name from docker compose ps container_name = self.docker_utils.get_container_name_by_service(container) if not container_name: container_name = f"streama2026-{container}" # Fallback to default if status == "exited": if self.docker_utils.start_container(container_name): self.logger.success("DOCKER", f"{container} started") else: self.logger.error("DOCKER", f"Failed to start {container}") all_ok = False else: # Try to restart if self.docker_utils.restart_container(container_name): self.logger.success("DOCKER", f"{container} restarted") else: self.logger.error("DOCKER", f"Failed to restart {container}") all_ok = False return all_ok def _check_network(self) -> bool: """Check if Docker network exists""" self.logger.info("DOCKER", "Checking Docker network...") if self.docker_utils.check_network_exists("streama-network"): self.logger.success("DOCKER", "Docker network exists") return True else: # Try to create network self.logger.info("DOCKER", "Docker network not found - attempting to create...") if self.docker_utils.create_network("streama-network"): self.logger.success("DOCKER", "Docker network created") return True else: self.logger.info("DOCKER", "Docker network will be created on next compose up") return True # Not critical, will be created automatically def _check_volumes(self) -> bool: """Check if Docker volumes exist""" self.logger.info("DOCKER", "Checking Docker volumes...") volumes = [ "streama2026_postgres_data", "streama2026_redis_data", "streama2026_stream_storage", "streama2026_ffmpeg_cache" ] all_ok = True for volume in volumes: if self.docker_utils.check_volume_exists(volume): self.logger.success("DOCKER", f"Volume {volume} exists") else: self.logger.warning("DOCKER", f"Volume {volume} not found - will be created on first use") return True # Not critical, volumes will be created automatically def rebuild_containers(self, services: Optional[list] = None) -> bool: """Rebuild specific containers or all""" self.logger.info("DOCKER", "Rebuilding containers...") if services is None: services = ["backend", "frontend"] if self.docker_utils.rebuild_containers(services): self.logger.success("DOCKER", f"Containers rebuilt: {', '.join(services)}") return True else: self.logger.error("DOCKER", "Failed to rebuild containers") return False def wait_for_health(self, timeout: int = 60) -> bool: """Wait for all containers to become healthy with active polling""" self.logger.info("HEALTH", "Waiting for health checks (active polling)...") container_names = { "postgres": "streama2026-postgres", "redis": "streama2026-redis", "backend": "streama2026-backend" } all_ok = True for service, container_name in container_names.items(): self.logger.info("HEALTH", f"Waiting for {service} to become healthy...") if self.docker_utils.wait_for_health(container_name, timeout): self.logger.success("HEALTH", f"{service}: healthy") else: # Backend container health check might fail but API still works if service == "backend": self.logger.info("HEALTH", f"{service}: failed to become healthy (but API works)") else: self.logger.error("HEALTH", f"{service}: failed to become healthy") all_ok = False return all_ok