osgos_postinstall/deb_dist/postinstall-1.0.1/build/lib/tests/docker_config.py

import logging
import subprocess
import os
import re

REGISTRY_URL = "dr.gosrv.ru"
REGISTRY_LOGIN = "dr_admin"
REGISTRY_PASSWORD = "Qweasdzxc123C@"
DOCKER_COMPOSE_FILE_PATH = "/usr/local/share/postinstall/portainer/"


class DockerConfig:
    def check_installed(self, command: str) -> bool:
        try:
            return (
                subprocess.run(
                    ["which", command], capture_output=True, text=True
                ).returncode
                == 0
            )
        except subprocess.SubprocessError as e:
            logging.error(f"Error checking if {command} is installed: {e}")
            return False

    def install_docker(self):
        try:
            if not self.check_installed("docker"):
                logging.info("Установка Docker...")
                subprocess.run(["apt-get", "update"], check=True)
                subprocess.run(["apt-get", "install", "-y", "docker.io"], check=True)
                logging.info("Docker успешно установился")

            if not self.check_installed("docker-compose"):
                logging.info("Установка Docker Compose...")
                subprocess.run(
                    ["apt-get", "install", "-y", "docker-compose"], check=True
                )
                logging.info("Docker Compose успешно установился")
        except subprocess.CalledProcessError as e:
            logging.error(f"Ошибка при установке Docker: {e}")
            raise

    def check_docker_login(self, registry_url):
        try:
            result = subprocess.run(
                ["docker", "login", registry_url, "--password-stdin"],
                input=b"\n",
                capture_output=True,
                text=True,
                check=True,
            )
            return "login succeeded" in result.stdout.lower()
        except subprocess.CalledProcessError as e:
            logging.error(f"Error checking Docker login: {e}")
            return False

    def login_to_registry(self):
        try:
            if not self.check_docker_login(REGISTRY_URL):
                logging.info("Вход в Docker registry...")
                subprocess.run(
                    ["docker", "login", REGISTRY_URL],
                    input=f"{REGISTRY_LOGIN}\n{REGISTRY_PASSWORD}\n".encode(),
                    check=True,
                )
                logging.info("Успешный вход в Docker registry")
        except subprocess.CalledProcessError as e:
            logging.error(f"Ошибка при входе в Docker registry: {e}")
            raise

    def install_portainer(self, port, login, password, ip):
        try:
            if not self.check_docker_login(REGISTRY_URL):
                self.login_to_registry()

            os.makedirs(os.path.dirname(DOCKER_COMPOSE_FILE_PATH), exist_ok=True)
            docker_compose_content = f"""
version: '3'

services:
  portainer:
    image: dr.gosrv.ru/portainer
    container_name: portainer
    restart: unless-stopped
    security_opt:
      - no-new-privileges:true
    volumes:
      - /etc/localtime:/etc/localtime:ro
      - /var/run/docker.sock:/var/run/docker.sock:ro
      - portainer_data:/data
      - ./portainer_password:{DOCKER_COMPOSE_FILE_PATH}/portainer_password
    ports:
      - "{ip}:{port}:9000"
    environment:
      - ADMIN_USER={login}
      - ADMIN_PASSWORD_FILE={DOCKER_COMPOSE_FILE_PATH}/portainer_password

volumes:
  portainer_data:
            """

            with open(DOCKER_COMPOSE_FILE_PATH + "docker-compose.yml", "w") as f:
                f.write(docker_compose_content)

            subprocess.run(
                ["htpasswd", "-nb", login, password],
                stdout=open(DOCKER_COMPOSE_FILE_PATH + "portainer_password", "w"),
                check=True,
            )

            subprocess.run(
                [
                    "docker-compose",
                    "-f",
                    DOCKER_COMPOSE_FILE_PATH + "docker-compose.yml",
                    "up",
                    "-d",
                ],
                check=True,
            )
            logging.info("Portainer успешно установлен")
        except subprocess.CalledProcessError as e:
            logging.error(f"Ошибка при установке Portainer: {e}")
            raise
        except IOError as e:
            logging.error(f"Ошибка при создании файла: {e}")
            raise

    def validate_input(self, value, pattern, error_message):
        if not re.match(pattern, value):
            return False
        return True