db_sync_tool.utility.system.link_configuration_with_hosts() - Code Metrics - Inspection of "[Task] Further error prevention" - jackd248/db-sync-tool - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 8d66d0...ffbe9f )

by Konrad

created 2020-11-26 10:18 UTC

link_configuration_with_hosts() C

↳ Parent: db_sync_tool.utility.system

Complexity

Conditions

Size

Total Lines	40
Code Lines	28

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	28
dl	0
loc	40
rs	5.4
c	0
b	0
f	0
cc	11
nop	0

How to fix Complexity

#!/usr/bin/env python3

# -*- coding: utf-8 -*-

import sys
import json
import os
import getpass
from db_sync_tool.utility import log, parser, mode, helper, output

#
# GLOBALS
#

config = {}
option = {
    'verbose': False,
    'mute': False,
    'use_origin_ssh_key': False,
    'use_target_ssh_key': False,
    'keep_dump': False,
    'dump_name': '',
    'import': '',
    'link_hosts': '',
    'default_origin_dump_dir': True,
    'default_target_dump_dir': True,
    'check_dump': True,
    'is_same_client': False,
    'config_file_path': None,
    'ssh_password': {
        'origin': None,
        'target': None
    }
}

#
# DEFAULTS
#

default_local_sync_path = '/tmp/'



#
# FUNCTIONS
#

def check_target_configuration():
    """
    Checking target database configuration
    :return:
    """
    parser.get_database_configuration(mode.Client.TARGET)


def get_configuration(host_config):
    """
    Checking configuration information by file or dictionary
    :param host_config: Dictionary
    :return:
    """
    if not option['config_file_path'] is None:
        if os.path.isfile(option['config_file_path']):
            with open(option['config_file_path'], 'r') as read_file:
                config['host'] = json.load(read_file)
                output.message(
                    output.Subject.LOCAL,
                    'Loading host configuration',
                    True
                )

                check_options()
        else:
            sys.exit(
                output.message(
                    output.Subject.ERROR,
                    f'Local configuration not found: {option["config_file_path"]}',
                    False
                )
            )

    if host_config:
        config['host'] = host_config

    log.get_logger().info('Starting db_sync_tool')
    if option['config_file_path']:
        output.message(
            output.Subject.INFO,
            'Configuration: ' + option['config_file_path'],
            True,
            True
        )


def check_options():
    """
    Checking configuration provided file
    :return:
    """
    if 'dump_dir' in config['host']['origin']:
        option['default_origin_dump_dir'] = False

    if 'dump_dir' in config['host']['target']:
        option['default_target_dump_dir'] = False

    if 'check_dump' in config['host']:
        option['check_dump'] = config['host']['check_dump']

    link_configuration_with_hosts()
    mode.check_sync_mode()
    check_authorization(mode.Client.ORIGIN)
    check_authorization(mode.Client.TARGET)


def check_authorization(client):
    """
    Checking arguments and fill options array
    :param client: String
    :return:
    """
    # only need authorization if client is remote
    if mode.is_remote(client):
        # Workaround
        if (mode.get_sync_mode() == mode.SyncMode.DUMP_REMOTE and client == mode.Client.TARGET) or (

                mode.get_sync_mode() == mode.SyncMode.DUMP_LOCAL and client == mode.Client.ORIGIN) or (

                mode.get_sync_mode() == mode.SyncMode.IMPORT_REMOTE and client == mode.Client.ORIGIN):

            return

        # ssh key authorization
        if 'ssh_key' in config['host'][client]:

            _ssh_key = config['host'][client]['ssh_key']
            if os.path.isfile(_ssh_key):
                option[f'use_{client}_ssh_key'] = True
            else:
                sys.exit(
                    output.message(
                        output.Subject.ERROR,
                        f'SSH {client} private key not found: {_ssh_key}',
                        False
                    )
                )
        elif 'password' in config['host'][client]:
            config['host'][client]['password'] = config['host'][client]['password']
        else:
            # user input authorization
            config['host'][client]['password'] = get_password_by_user(client)

        if mode.get_sync_mode() == mode.SyncMode.DUMP_REMOTE and client == mode.Client.ORIGIN and 'password' in config['host'][mode.Client.ORIGIN]:

            config['host'][mode.Client.TARGET]['password'] = config['host'][mode.Client.ORIGIN]['password']



def get_password_by_user(client):
    """
    Getting password by user input
    :param client: String
    :return: String password
    """
    _password = getpass.getpass(
        output.message(
            output.Subject.INFO,
            'SSH password ' + helper.get_ssh_host_name(client, True) + ': ',
            False
        )
    )

    while _password.strip() == '':
        output.message(
            output.Subject.WARNING,
            'Password seems to be empty. Please enter a valid password.',
            True
        )

        _password = getpass.getpass(
            output.message(
                output.Subject.INFO,
                'SSH password ' + helper.get_ssh_host_name(client, True) + ': ',
                False
            )
        )

    return _password


def check_args_options(args):
    """
    Checking arguments and fill options array
    :param args:
    :return:
    """
    global option

    global default_local_sync_path


    if not args.file is None:
        option['config_file_path'] = args.file

    if not args.verbose is None:
        option['verbose'] = args.verbose

    if not args.mute is None:
        option['mute'] = args.mute

    if not args.importfile is None:
        option['import'] = args.importfile

    if not args.dumpname is None:
        option['dump_name'] = args.dumpname

    if not args.hosts is None:
        option['link_hosts'] = args.hosts

    if not args.keepdump is None:
        default_local_sync_path = args.keepdump

        # Adding trailing slash if necessary
        if default_local_sync_path[-1] != '/':
            default_local_sync_path += '/'

        option['keep_dump'] = True
        output.message(
            output.Subject.INFO,
            '"Keep dump" option chosen',
            True
        )


def link_configuration_with_hosts():
    """
    Merging the hosts definition with the given configuration file
    :return:
    """
    if ('link' in config['host']['origin'] or 'link' in config['host']['target']) and option['link_hosts'] == '':

        # Try to find default hosts.json file in same directory
        sys.exit(
            output.message(
                output.Subject.ERROR,
                f'Missing hosts file for linking hosts with configuration. '

                f'Use the "-o" / "--hosts" argument to define the filepath for the hosts file, when using a link parameter within the configuration.',

                False
            )
        )

    if option['link_hosts'] != '':
        if os.path.isfile(option['link_hosts']):
            with open(option['link_hosts'], 'r') as read_file:
                _hosts = json.load(read_file)
                output.message(
                    output.Subject.INFO,
                    'Linking configuration with hosts',
                    True
                )
                if 'link' in config['host']['origin']:
                    _host_name = str(config['host']['origin']['link']).replace('@','')

                    if _host_name in _hosts:
                        config['host']['origin'] = {**config['host']['origin'], **_hosts[_host_name]}


                if 'link' in config['host']['target']:
                    _host_name = str(config['host']['target']['link']).replace('@','')

                    if _host_name in _hosts:
                        config['host']['target'] = {**config['host']['target'], **_hosts[_host_name]}

        else:
            sys.exit(
                output.message(
                    output.Subject.ERROR,
                    f'Local host file not found: {option["link_hosts"]}',
                    False
                )
            )






1			#!/usr/bin/env python3
			0 ignored issues – show introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report Missing module docstring Loading history...
2			# -- coding: utf-8 --
3
4			import sys
5			import json
6			import os
7			import getpass
8			from db_sync_tool.utility import log, parser, mode, helper, output
9
10			#
11			# GLOBALS
12			#
13
14			config = {}
15			option = {
16			'verbose': False,
17			'mute': False,
18			'use_origin_ssh_key': False,
19			'use_target_ssh_key': False,
20			'keep_dump': False,
21			'dump_name': '',
22			'import': '',
23			'link_hosts': '',
24			'default_origin_dump_dir': True,
25			'default_target_dump_dir': True,
26			'check_dump': True,
27			'is_same_client': False,
28			'config_file_path': None,
29			'ssh_password': {
30			'origin': None,
31			'target': None
32			}
33			}
34
35			#
36			# DEFAULTS
37			#
38
39			default_local_sync_path = '/tmp/'
			0 ignored issues – show Coding Style Naming introduced 2020-11-26 10:20 UTC by Report Bug Copy Issue Report Constant name "default_local_sync_path" doesn't conform to UPPER_CASE naming style ('([^\\W\\da-z][^\\Wa-z]\|__.__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
40
41
42			#
43			# FUNCTIONS
44			#
45
46			def check_target_configuration():
47			"""
48			Checking target database configuration
49			:return:
50			"""
51			parser.get_database_configuration(mode.Client.TARGET)
52
53
54			def get_configuration(host_config):
55			"""
56			Checking configuration information by file or dictionary
57			:param host_config: Dictionary
58			:return:
59			"""
60			if not option['config_file_path'] is None:
61			if os.path.isfile(option['config_file_path']):
62			with open(option['config_file_path'], 'r') as read_file:
63			config['host'] = json.load(read_file)
64			output.message(
65			output.Subject.LOCAL,
66			'Loading host configuration',
67			True
68			)
69
70			check_options()
71			else:
72			sys.exit(
73			output.message(
74			output.Subject.ERROR,
75			f'Local configuration not found: {option["config_file_path"]}',
76			False
77			)
78			)
79
80			if host_config:
81			config['host'] = host_config
82
83			log.get_logger().info('Starting db_sync_tool')
84			if option['config_file_path']:
85			output.message(
86			output.Subject.INFO,
87			'Configuration: ' + option['config_file_path'],
88			True,
89			True
90			)
91
92
93			def check_options():
94			"""
95			Checking configuration provided file
96			:return:
97			"""
98			if 'dump_dir' in config['host']['origin']:
99			option['default_origin_dump_dir'] = False
100
101			if 'dump_dir' in config['host']['target']:
102			option['default_target_dump_dir'] = False
103
104			if 'check_dump' in config['host']:
105			option['check_dump'] = config['host']['check_dump']
106
107			link_configuration_with_hosts()
108			mode.check_sync_mode()
109			check_authorization(mode.Client.ORIGIN)
110			check_authorization(mode.Client.TARGET)
111
112
113			def check_authorization(client):
114			"""
115			Checking arguments and fill options array
116			:param client: String
117			:return:
118			"""
119			# only need authorization if client is remote
120			if mode.is_remote(client):
121			# Workaround
122			if (mode.get_sync_mode() == mode.SyncMode.DUMP_REMOTE and client == mode.Client.TARGET) or (
			0 ignored issues – show best-practice introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report Too many boolean expressions in if statement (6/5) Loading history...
123			mode.get_sync_mode() == mode.SyncMode.DUMP_LOCAL and client == mode.Client.ORIGIN) or (
			0 ignored issues – show Coding Style introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (103/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
124			mode.get_sync_mode() == mode.SyncMode.IMPORT_REMOTE and client == mode.Client.ORIGIN):
			0 ignored issues – show Coding Style introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report Wrong hanging indentation (add 4 spaces). Loading history... Coding Style introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (102/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
125			return
126
127			# ssh key authorization
128			if 'ssh_key' in config['host'][client]:
			0 ignored issues – show Comprehensibility Best Practice introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report The variable `config` does not seem to be defined. Loading history...
129			_ssh_key = config['host'][client]['ssh_key']
130			if os.path.isfile(_ssh_key):
131			option[f'use_{client}_ssh_key'] = True
132			else:
133			sys.exit(
134			output.message(
135			output.Subject.ERROR,
136			f'SSH {client} private key not found: {_ssh_key}',
137			False
138			)
139			)
140			elif 'password' in config['host'][client]:
141			config['host'][client]['password'] = config['host'][client]['password']
142			else:
143			# user input authorization
144			config['host'][client]['password'] = get_password_by_user(client)
145
146			if mode.get_sync_mode() == mode.SyncMode.DUMP_REMOTE and client == mode.Client.ORIGIN and 'password' in config['host'][mode.Client.ORIGIN]:
			0 ignored issues – show Coding Style introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (147/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
147			config['host'][mode.Client.TARGET]['password'] = config['host'][mode.Client.ORIGIN]['password']
			0 ignored issues – show Coding Style introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (107/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
148
149
150			def get_password_by_user(client):
151			"""
152			Getting password by user input
153			:param client: String
154			:return: String password
155			"""
156			_password = getpass.getpass(
157			output.message(
158			output.Subject.INFO,
159			'SSH password ' + helper.get_ssh_host_name(client, True) + ': ',
160			False
161			)
162			)
163
164			while _password.strip() == '':
165			output.message(
166			output.Subject.WARNING,
167			'Password seems to be empty. Please enter a valid password.',
168			True
169			)
170
171			_password = getpass.getpass(
172			output.message(
173			output.Subject.INFO,
174			'SSH password ' + helper.get_ssh_host_name(client, True) + ': ',
175			False
176			)
177			)
178
179			return _password
180
181
182			def check_args_options(args):
183			"""
184			Checking arguments and fill options array
185			:param args:
186			:return:
187			"""
188			global option
			0 ignored issues – show Coding Style introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report Usage of the `global` statement should be avoided. Usage of `global` can make code hard to read and test, its usage is generally not recommended unless you are dealing with legacy code. Loading history... Coding Style Naming introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report Constant name "option" doesn't conform to UPPER_CASE naming style ('([^\\W\\da-z][^\\Wa-z]\|__.__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
189			global default_local_sync_path
			0 ignored issues – show Coding Style introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report Usage of the `global` statement should be avoided. Usage of `global` can make code hard to read and test, its usage is generally not recommended unless you are dealing with legacy code. Loading history... Coding Style Naming introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report Constant name "default_local_sync_path" doesn't conform to UPPER_CASE naming style ('([^\\W\\da-z][^\\Wa-z]\|__.__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
190
191			if not args.file is None:
192			option['config_file_path'] = args.file
193
194			if not args.verbose is None:
195			option['verbose'] = args.verbose
196
197			if not args.mute is None:
198			option['mute'] = args.mute
199
200			if not args.importfile is None:
201			option['import'] = args.importfile
202
203			if not args.dumpname is None:
204			option['dump_name'] = args.dumpname
205
206			if not args.hosts is None:
207			option['link_hosts'] = args.hosts
208
209			if not args.keepdump is None:
210			default_local_sync_path = args.keepdump
211
212			# Adding trailing slash if necessary
213			if default_local_sync_path[-1] != '/':
214			default_local_sync_path += '/'
215
216			option['keep_dump'] = True
217			output.message(
218			output.Subject.INFO,
219			'"Keep dump" option chosen',
220			True
221			)
222
223
224			def link_configuration_with_hosts():
225			"""
226			Merging the hosts definition with the given configuration file
227			:return:
228			"""
229			if ('link' in config['host']['origin'] or 'link' in config['host']['target']) and option['link_hosts'] == '':
			0 ignored issues – show Comprehensibility Best Practice introduced 2020-11-26 10:20 UTC by Report Bug Copy Issue Report The variable `config` does not seem to be defined. Loading history... Coding Style introduced 2020-11-26 10:20 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (113/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
230			# Try to find default hosts.json file in same directory
231			sys.exit(
232			output.message(
233			output.Subject.ERROR,
234			f'Missing hosts file for linking hosts with configuration. '
			0 ignored issues – show introduced 2020-11-26 10:20 UTC by Report Bug Copy Issue Report Using an f-string that does not have any interpolated variables Loading history...
235			f'Use the "-o" / "--hosts" argument to define the filepath for the hosts file, when using a link parameter within the configuration.',
			0 ignored issues – show Coding Style introduced 2020-11-26 10:20 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (150/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
236			False
237			)
238			)
239
240			if option['link_hosts'] != '':
241			if os.path.isfile(option['link_hosts']):
242			with open(option['link_hosts'], 'r') as read_file:
243			_hosts = json.load(read_file)
244			output.message(
245			output.Subject.INFO,
246			'Linking configuration with hosts',
247			True
248			)
249			if 'link' in config['host']['origin']:
250			_host_name = str(config['host']['origin']['link']).replace('@','')
			0 ignored issues – show Coding Style introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report Exactly one space required after comma Loading history...
251			if _host_name in _hosts:
252			config['host']['origin'] = {config['host']['origin'], _hosts[_host_name]}
			0 ignored issues – show Coding Style introduced 2020-11-26 10:20 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (101/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
253
254			if 'link' in config['host']['target']:
255			_host_name = str(config['host']['target']['link']).replace('@','')
			0 ignored issues – show Coding Style introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report Exactly one space required after comma Loading history...
256			if _host_name in _hosts:
257			config['host']['target'] = {config['host']['target'], _hosts[_host_name]}
			0 ignored issues – show Coding Style introduced 2020-11-26 10:20 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (101/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
258			else:
259			sys.exit(
260			output.message(
261			output.Subject.ERROR,
262			f'Local host file not found: {option["link_hosts"]}',
263			False
264			)
265			)
266
267
268
			0 ignored issues – show coding-style introduced 2020-11-26 10:20 UTC by Report Bug Copy Issue Report Trailing newlines Loading history...
269

jackd248 / db-sync-tool

Push — master ( 8d66d0...ffbe9f )

link_configuration_with_hosts() C

Complexity

Size

Duplication

Importance

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like