stakkr.actions._print_status_body() - Code Metrics - Inspection of "Create .coveragerc" - stakkr-org/stakkr - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 61e6a6...28b40d )

by Emmanuel

created 2019-06-22 07:09 UTC

stakkr.actions._print_status_body() A

↳ Parent: stakkr.actions

Complexity

Conditions

Size

Total Lines	11
Code Lines	9

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	12

Importance

Changes

Metric	Value
cc	3
eloc	9
nop	1
dl	0
loc	11
ccs	0
cts	6
cp	0
crap	12
rs	9.95
c	0
b	0
f	0

# coding: utf-8
"""Stakkr main controller. Used by the CLI to do all its actions."""

import os
from platform import system as os_name
import subprocess
import sys
import click
from clint.textui import colored, puts, columns
from stakkr import command, docker_actions as docker
from stakkr.configreader import Config
from stakkr.proxy import Proxy


class StakkrActions:
    """Main class that does actions asked in the cli."""

    def __init__(self, ctx: dict):
        """Set all require properties."""
        self.context = ctx

        # Set some general variables
        self.config = None
        self.project_name = None
        self.project_dir = None
        self.cwd_relative = None

    def console(self, container: str, user: str, tty: bool):
        """Enter a container. Stakkr will try to guess the right shell."""
        self.init_project()

        docker.check_cts_are_running(self.project_name)

        tty = 't' if tty is True else ''
        ct_name = docker.get_ct_name(container)
        cmd = ['docker', 'exec', '-u', user, '-i' + tty]
        cmd += [docker.get_ct_name(container), docker.guess_shell(ct_name)]
        subprocess.call(cmd)

        command.verbose(self.context['VERBOSE'], 'Command : "' + ' '.join(cmd) + '"')

    def get_services_urls(self):
        """Once started, displays a message with a list of running containers."""
        self.init_project()

        cts = docker.get_running_containers(self.project_name)[1]

        text = ''
        for _, ct_info in cts.items():
            service_config = self.config['services'][ct_info['compose_name']]
            if ({'service_name', 'service_url'} <= set(service_config)) is False:
                continue

            url = self.get_url(service_config['service_url'], ct_info['compose_name'])
            name = colored.yellow(service_config['service_name'])

            text += '  - For {}'.format(name).ljust(55, ' ') + ' : ' + url + '\n'

            if 'service_extra_ports' in service_config:
                ports = ', '.join(map(str, service_config['service_extra_ports']))
                text += ' '*4 + '(In your containers use the host '
                text += '"{}" and port(s) {})\n'.format(ct_info['compose_name'], ports)

        return text

    def exec_cmd(self, container: str, user: str, args: tuple, tty: bool):
        """Run a command from outside to any container. Wrapped into /bin/sh."""
        self.init_project()

        docker.check_cts_are_running(self.project_name)

        # Protect args to avoid strange behavior in exec
        args = ['"{}"'.format(arg) for arg in args]

        tty = 't' if tty is True else ''
        ct_name = docker.get_ct_name(container)
        cmd = ['docker', 'exec', '-u', user, '-i' + tty, ct_name, 'sh', '-c']
        cmd += ["""test -d "/var/{0}" && cd "/var/{0}" ; exec {1}""".format(self.cwd_relative, ' '.join(args))]
        command.verbose(self.context['VERBOSE'], 'Command : "' + ' '.join(cmd) + '"')
        subprocess.call(cmd, stdin=sys.stdin)

    def get_config(self):
        """Read and validate config from config file"""
        config = Config(self.context['CONFIG'])
        main_config = config.read()
        if main_config is False:
            config.display_errors()
            sys.exit(1)

        return main_config

    def init_project(self):
        """
        Initializing the project by reading config and
        setting some properties of the object
        """
        if self.config is not None:
            return

        self.config = self.get_config()
        self.project_name = self.config['project_name']
        self.project_dir = self.config['project_dir']
        sys.path.append(self.project_dir)

        self.cwd_relative = self._get_relative_dir()
        os.chdir(self.project_dir)

    def start(self, container: str, pull: bool, recreate: bool, proxy: bool):
        """If not started, start the containers defined in config."""
        self.init_project()
        verb = self.context['VERBOSE']
        debug = self.context['DEBUG']

        self._is_up(container)

        if pull is True:
            command.launch_cmd_displays_output(self._get_compose_base_cmd() + ['pull'], verb, debug, True)

        recreate_param = '--force-recreate' if recreate is True else '--no-recreate'
        cmd = self._get_compose_base_cmd() + ['up', '-d', recreate_param, '--remove-orphans']
        cmd += _get_single_container_option(container)

        command.verbose(self.context['VERBOSE'], 'Command: ' + ' '.join(cmd))
        command.launch_cmd_displays_output(cmd, verb, debug, True)

        running_cts, cts = docker.get_running_containers(self.project_name)
        if not running_cts:
            raise SystemError("Couldn't start the containers, run the start with '-v' and '-d'")

        self._run_iptables_rules(cts)
        if proxy is True:
            conf = self.config['proxy']
            Proxy(conf.get('http_port'), conf.get('https_port')).start(
                docker.get_network_name(self.project_name))

    def status(self):
        """Return a nice table with the list of started containers."""
        self.init_project()

        try:
            docker.check_cts_are_running(self.project_name)
        except SystemError:
            puts(colored.yellow('[INFO]') + ' stakkr is currently stopped')
            sys.exit(0)

        _, cts = docker.get_running_containers(self.project_name)

        _print_status_headers()
        _print_status_body(cts)

    def stop(self, container: str, proxy: bool):
        """If started, stop the containers defined in config. Else throw an error."""
        self.init_project()
        verb = self.context['VERBOSE']
        debug = self.context['DEBUG']

        docker.check_cts_are_running(self.project_name)

        cmd = self._get_compose_base_cmd() + ['stop'] + _get_single_container_option(container)
        command.launch_cmd_displays_output(cmd, verb, debug, True)

        running_cts, _ = docker.get_running_containers(self.project_name)
        if running_cts and container is None:
            raise SystemError("Couldn't stop services ...")

        if proxy is True:
            Proxy().stop()

    def _get_compose_base_cmd(self):
        if self.context['CONFIG'] is None:
            return ['stakkr-compose']

        return ['stakkr-compose', '-c', self.context['CONFIG']]

    def _get_relative_dir(self):
        if os.getcwd().startswith(self.project_dir):
            return os.getcwd()[len(self.project_dir):].lstrip('/')

        return ''

    def _is_up(self, container: str):
        try:
            docker.check_cts_are_running(self.project_name)
        except SystemError:
            return

        if container is None:
            puts(colored.yellow('[INFO]') + ' stakkr is already started ...')
            sys.exit(0)

        # If single container : check if that specific one is running
        ct_name = docker.get_ct_item(container, 'name')
        if docker.container_running(ct_name):
            puts(colored.yellow('[INFO]') + ' service {} is already started ...'.format(container))
            sys.exit(0)

    def _run_iptables_rules(self, cts: dict):
        """For some containers we need to add iptables rules added from the config."""
        for _, ct_info in cts.items():
            container = ct_info['compose_name']
            ct_config = self.config['services'][container]
            if 'blocked_ports' not in ct_config:
                continue

            blocked_ports = ct_config['blocked_ports']
            error, msg = docker.block_ct_ports(container, blocked_ports, self.project_name)
            if error is True:
                click.secho(msg, fg='red')
                continue

            command.verbose(self.context['VERBOSE'], msg)

    def get_url(self, service_url: str, service: str):
        """Build URL to be displayed."""
        proxy_conf = self.config['proxy']
        # By default our URL is the IP
        url = docker.get_ct_item(service, 'ip')
        # If proxy enabled, display nice urls
        if bool(proxy_conf['enabled']):
            http_port = int(proxy_conf['http_port'])
            url = docker.get_ct_item(service, 'traefik_host').lower()
            url += '' if http_port == 80 else ':{}'.format(http_port)
        elif os_name() in ['Windows', 'Darwin']:
            puts(colored.yellow('[WARNING]') + ' Under Win and Mac, you need the proxy enabled')

        return service_url.format(url)


def _get_single_container_option(container: str):
    if container is None:
        return []

    return [container]


def _print_status_headers():
    """Display messages for stakkr status (header)"""
    puts(columns(
        [(colored.green('Container')), 16], [colored.green('IP'), 15],
        [(colored.green('Url')), 32], [(colored.green('Image')), 32],
        [(colored.green('Docker ID')), 15], [(colored.green('Docker Name')), 25]
    ))

    puts(columns(
        ['-'*16, 16], ['-'*15, 15],
        ['-'*32, 32], ['-'*32, 32],
        ['-'*15, 15], ['-'*25, 25]
    ))


def _print_status_body(cts: dict):
    """Display messages for stakkr status (body)"""
    for container in sorted(cts.keys()):
        ct_data = cts[container]
        if ct_data['ip'] == '':
            continue

        puts(columns(
            [ct_data['compose_name'], 16], [ct_data['ip'], 15],
            [ct_data['traefik_host'], 32], [ct_data['image'], 32],
            [ct_data['id'][:12], 15], [ct_data['name'], 25]
        ))


1			# coding: utf-8
2			"""Stakkr main controller. Used by the CLI to do all its actions."""
3
4			import os
5			from platform import system as os_name
6			import subprocess
7			import sys
8			import click
9			from clint.textui import colored, puts, columns
10			from stakkr import command, docker_actions as docker
11			from stakkr.configreader import Config
12			from stakkr.proxy import Proxy
13
14
15			class StakkrActions:
16			"""Main class that does actions asked in the cli."""
17
18			def __init__(self, ctx: dict):
19			"""Set all require properties."""
20			self.context = ctx
21
22			# Set some general variables
23			self.config = None
24			self.project_name = None
25			self.project_dir = None
26			self.cwd_relative = None
27
28			def console(self, container: str, user: str, tty: bool):
29			"""Enter a container. Stakkr will try to guess the right shell."""
30			self.init_project()
31
32			docker.check_cts_are_running(self.project_name)
33
34			tty = 't' if tty is True else ''
35			ct_name = docker.get_ct_name(container)
36			cmd = ['docker', 'exec', '-u', user, '-i' + tty]
37			cmd += [docker.get_ct_name(container), docker.guess_shell(ct_name)]
38			subprocess.call(cmd)
39
40			command.verbose(self.context['VERBOSE'], 'Command : "' + ' '.join(cmd) + '"')
41
42			def get_services_urls(self):
43			"""Once started, displays a message with a list of running containers."""
44			self.init_project()
45
46			cts = docker.get_running_containers(self.project_name)[1]
47
48			text = ''
49			for _, ct_info in cts.items():
50			service_config = self.config['services'][ct_info['compose_name']]
51			if ({'service_name', 'service_url'} <= set(service_config)) is False:
52			continue
53
54			url = self.get_url(service_config['service_url'], ct_info['compose_name'])
55			name = colored.yellow(service_config['service_name'])
56
57			text += ' - For {}'.format(name).ljust(55, ' ') + ' : ' + url + '\n'
58
59			if 'service_extra_ports' in service_config:
60			ports = ', '.join(map(str, service_config['service_extra_ports']))
61			text += ' '*4 + '(In your containers use the host '
62			text += '"{}" and port(s) {})\n'.format(ct_info['compose_name'], ports)
63
64			return text
65
66			def exec_cmd(self, container: str, user: str, args: tuple, tty: bool):
67			"""Run a command from outside to any container. Wrapped into /bin/sh."""
68			self.init_project()
69
70			docker.check_cts_are_running(self.project_name)
71
72			# Protect args to avoid strange behavior in exec
73			args = ['"{}"'.format(arg) for arg in args]
74
75			tty = 't' if tty is True else ''
76			ct_name = docker.get_ct_name(container)
77			cmd = ['docker', 'exec', '-u', user, '-i' + tty, ct_name, 'sh', '-c']
78			cmd += ["""test -d "/var/{0}" && cd "/var/{0}" ; exec {1}""".format(self.cwd_relative, ' '.join(args))]
79			command.verbose(self.context['VERBOSE'], 'Command : "' + ' '.join(cmd) + '"')
80			subprocess.call(cmd, stdin=sys.stdin)
81
82			def get_config(self):
83			"""Read and validate config from config file"""
84			config = Config(self.context['CONFIG'])
85			main_config = config.read()
86			if main_config is False:
87			config.display_errors()
88			sys.exit(1)
89
90			return main_config
91
92			def init_project(self):
93			"""
94			Initializing the project by reading config and
95			setting some properties of the object
96			"""
97			if self.config is not None:
98			return
99
100			self.config = self.get_config()
101			self.project_name = self.config['project_name']
102			self.project_dir = self.config['project_dir']
103			sys.path.append(self.project_dir)
104
105			self.cwd_relative = self._get_relative_dir()
106			os.chdir(self.project_dir)
107
108			def start(self, container: str, pull: bool, recreate: bool, proxy: bool):
109			"""If not started, start the containers defined in config."""
110			self.init_project()
111			verb = self.context['VERBOSE']
112			debug = self.context['DEBUG']
113
114			self._is_up(container)
115
116			if pull is True:
117			command.launch_cmd_displays_output(self._get_compose_base_cmd() + ['pull'], verb, debug, True)
118
119			recreate_param = '--force-recreate' if recreate is True else '--no-recreate'
120			cmd = self._get_compose_base_cmd() + ['up', '-d', recreate_param, '--remove-orphans']
121			cmd += _get_single_container_option(container)
122
123			command.verbose(self.context['VERBOSE'], 'Command: ' + ' '.join(cmd))
124			command.launch_cmd_displays_output(cmd, verb, debug, True)
125
126			running_cts, cts = docker.get_running_containers(self.project_name)
127			if not running_cts:
128			raise SystemError("Couldn't start the containers, run the start with '-v' and '-d'")
129
130			self._run_iptables_rules(cts)
131			if proxy is True:
132			conf = self.config['proxy']
133			Proxy(conf.get('http_port'), conf.get('https_port')).start(
134			docker.get_network_name(self.project_name))
135
136			def status(self):
137			"""Return a nice table with the list of started containers."""
138			self.init_project()
139
140			try:
141			docker.check_cts_are_running(self.project_name)
142			except SystemError:
143			puts(colored.yellow('[INFO]') + ' stakkr is currently stopped')
144			sys.exit(0)
145
146			_, cts = docker.get_running_containers(self.project_name)
147
148			_print_status_headers()
149			_print_status_body(cts)
150
151			def stop(self, container: str, proxy: bool):
152			"""If started, stop the containers defined in config. Else throw an error."""
153			self.init_project()
154			verb = self.context['VERBOSE']
155			debug = self.context['DEBUG']
156
157			docker.check_cts_are_running(self.project_name)
158
159			cmd = self._get_compose_base_cmd() + ['stop'] + _get_single_container_option(container)
160			command.launch_cmd_displays_output(cmd, verb, debug, True)
161
162			running_cts, _ = docker.get_running_containers(self.project_name)
163			if running_cts and container is None:
164			raise SystemError("Couldn't stop services ...")
165
166			if proxy is True:
167			Proxy().stop()
168
169			def _get_compose_base_cmd(self):
170			if self.context['CONFIG'] is None:
171			return ['stakkr-compose']
172
173			return ['stakkr-compose', '-c', self.context['CONFIG']]
174
175			def _get_relative_dir(self):
176			if os.getcwd().startswith(self.project_dir):
177			return os.getcwd()[len(self.project_dir):].lstrip('/')
178
179			return ''
180
181			def _is_up(self, container: str):
182			try:
183			docker.check_cts_are_running(self.project_name)
184			except SystemError:
185			return
186
187			if container is None:
188			puts(colored.yellow('[INFO]') + ' stakkr is already started ...')
189			sys.exit(0)
190
191			# If single container : check if that specific one is running
192			ct_name = docker.get_ct_item(container, 'name')
193			if docker.container_running(ct_name):
194			puts(colored.yellow('[INFO]') + ' service {} is already started ...'.format(container))
195			sys.exit(0)
196
197			def _run_iptables_rules(self, cts: dict):
198			"""For some containers we need to add iptables rules added from the config."""
199			for _, ct_info in cts.items():
200			container = ct_info['compose_name']
201			ct_config = self.config['services'][container]
202			if 'blocked_ports' not in ct_config:
203			continue
204
205			blocked_ports = ct_config['blocked_ports']
206			error, msg = docker.block_ct_ports(container, blocked_ports, self.project_name)
207			if error is True:
208			click.secho(msg, fg='red')
209			continue
210
211			command.verbose(self.context['VERBOSE'], msg)
212
213			def get_url(self, service_url: str, service: str):
214			"""Build URL to be displayed."""
215			proxy_conf = self.config['proxy']
216			# By default our URL is the IP
217			url = docker.get_ct_item(service, 'ip')
218			# If proxy enabled, display nice urls
219			if bool(proxy_conf['enabled']):
220			http_port = int(proxy_conf['http_port'])
221			url = docker.get_ct_item(service, 'traefik_host').lower()
222			url += '' if http_port == 80 else ':{}'.format(http_port)
223			elif os_name() in ['Windows', 'Darwin']:
224			puts(colored.yellow('[WARNING]') + ' Under Win and Mac, you need the proxy enabled')
225
226			return service_url.format(url)
227
228
229			def _get_single_container_option(container: str):
230			if container is None:
231			return []
232
233			return [container]
234
235
236			def _print_status_headers():
237			"""Display messages for stakkr status (header)"""
238			puts(columns(
239			[(colored.green('Container')), 16], [colored.green('IP'), 15],
240			[(colored.green('Url')), 32], [(colored.green('Image')), 32],
241			[(colored.green('Docker ID')), 15], [(colored.green('Docker Name')), 25]
242			))
243
244			puts(columns(
245			['-'16, 16], ['-'15, 15],
246			['-'32, 32], ['-'32, 32],
247			['-'15, 15], ['-'25, 25]
248			))
249
250
251			def _print_status_body(cts: dict):
252			"""Display messages for stakkr status (body)"""
253			for container in sorted(cts.keys()):
254			ct_data = cts[container]
255			if ct_data['ip'] == '':
256			continue
257
258			puts(columns(
259			[ct_data['compose_name'], 16], [ct_data['ip'], 15],
260			[ct_data['traefik_host'], 32], [ct_data['image'], 32],
261			[ct_data['id'][:12], 15], [ct_data['name'], 25]
262			))
263

stakkr-org / stakkr

Push — master ( 61e6a6...28b40d )

stakkr.actions._print_status_body() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like