unki/integration/docker_compose.py

import subprocess
import shlex
from utils import *
from docker import docker, check_state, print_errors


class Compose:
    ALL_CONTAINERS = [
        'u_agent_1',
        'u_agent_2',
        'u_server',
        'u_db',
        'tests_runner',
    ]

    def __init__(self):
        self.container_tpl = 'integration_%s_1'
        self.cmd_container = self.container_tpl % 'tests_runner'
        self.ALL_CONTAINERS = [self.container_tpl % c for c in self.ALL_CONTAINERS]

    def _call(self, *args):
        subprocess.check_call([
            'docker-compose',
            '--no-ansi',
        ] + list(args)
        )

    def up(self):
        log('Instanciating cluster')
        self._call('up', '-d')
        log('Ok')

    def down(self):
        log('Shutting down cluster')
        self._call('down')
        log('Ok')

    def stop(self):
        log('Stopping cluster')
        self._call('stop')
        log('Ok')

    def run(self, cmd):
        container = self.cmd_container
        if isinstance(cmd, str):
            cmd = shlex.split(cmd)
        log(f'Running command "{cmd}" in container {container}')
        docker([
            'exec',
            '-i',
            container
        ] + cmd)
        log('Ok')

    def is_alive(self):
        log('Check if all containers are alive')

        errors = check_state(self.ALL_CONTAINERS)
        log('Check done')

        if errors:
            print_errors(errors)
            raise TestsError('Error during `is_alive` check')
        else:
            log('All containers are alive')