db_sync_tool.utility.system.check_options() - Code Metrics - Inspection of "[Release] Version 2.5.3" - jackd248/db-sync-tool - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 48167f...cc4cd6 )

by Konrad

created 2021-08-31 09:30 UTC

db_sync_tool.utility.system.check_options() A

↳ Parent: db_sync_tool.utility.system

Complexity

Conditions

Size

Total Lines	17
Code Lines	10

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	10
dl	0
loc	17
rs	9.9
c	0
b	0
f	0
cc	4
nop	0

#!/usr/bin/env python3
# -*- coding: future_fstrings -*-

"""
System module
"""

import sys
import json
import os
import getpass
import yaml
from db_sync_tool.utility import log, parser, mode, helper, output, validation
from db_sync_tool.remote import utility as remote_utility

#
# GLOBALS
#

config = {
    'verbose': False,
    'mute': False,
    'dry_run': 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,
    'clear_database': False,
    'force_password': False,
    'ssh_agent': False,
    'ssh_password': {
        'origin': None,
        'target': None
    },
    'link_target': None,
    'link_origin': None,
}

#
# DEFAULTS
#

default_local_sync_path = '/tmp/db_sync_tool/'



#
# 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:
    """
    global config


    if config['config_file_path'] is None and host_config == {}:
        sys.exit(
            output.message(
                output.Subject.ERROR,
                f'Configuration is missing, use a separate file or provide host parameter',

                False
            )
        )

    if host_config:
        config.update(host_config)

    _config_file_path = config['config_file_path']
    if not _config_file_path is None:
        if os.path.isfile(_config_file_path):
            with open(_config_file_path, 'r') as read_file:
                if _config_file_path.endswith('.json'):
                    config.update(json.load(read_file))
                elif _config_file_path.endswith('.yaml') or _config_file_path.endswith('.yml'):
                    config.update(yaml.safe_load(read_file))
                else:
                    sys.exit(
                        output.message(
                            output.Subject.ERROR,
                            f'Unsupported configuration file type [json,yml,yaml]: '
                            f'{config["config_file_path"]}',
                            False
                        )
                    )
                output.message(
                    output.Subject.LOCAL,
                    f'Loading host configuration '
                    f'{output.CliFormat.BLACK}{_config_file_path}{output.CliFormat.ENDC}',
                    True
                )
        else:
            sys.exit(
                output.message(
                    output.Subject.ERROR,
                    f'Local configuration not found: {config["config_file_path"]}',
                    False
                )
            )

    validation.check(config)
    check_options()
    helper.run_script(script='before')
    log.get_logger().info('Starting db_sync_tool')


def check_options():
    """
    Checking configuration provided file
    :return:
    """
    global config

    if 'dump_dir' in config['origin']:
        config['default_origin_dump_dir'] = False

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

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

    link_configuration_with_hosts()
    mode.check_sync_mode()


def check_authorizations():
    """
    Checking authorization for clients
    :return:
    """
    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 no authorization is needed
        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 config['force_password']:

            config[client]['password'] = get_password_by_user(client)
        elif 'ssh_key' in config[client]:
            _ssh_key = config[client]['ssh_key']
            if not os.path.isfile(_ssh_key):
                sys.exit(
                    output.message(
                        output.Subject.ERROR,
                        f'SSH {client} private key not found: {_ssh_key}',
                        False
                    )
                )
        elif 'password' in config[client]:
            config[client]['password'] = config[client]['password']
        elif remote_utility.check_keys_from_ssh_agent():
            config['ssh_agent'] = True
        else:
            # user input authorization
            config[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[mode.Client.ORIGIN]:
            config[mode.Client.TARGET]['password'] = config[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(config_file=None,

                       verbose=False,
                       yes=False,
                       mute=False,
                       dry_run=False,
                       import_file=None,
                       dump_name=None,
                       keep_dump=None,
                       host_file=None,
                       clear=False,
                       force_password=False):
    """
    Checking arguments and fill options array
    :param config_file:
    :param verbose:
    :param yes:
    :param mute:
    :param dry_run:
    :param import_file:
    :param dump_name:
    :param keep_dump:
    :param host_file:
    :param clear:
    :param force_password:
    :return:
    """
    global config

    global default_local_sync_path


    if not config_file is None:
        config['config_file_path'] = config_file

    if not verbose is None:
        config['verbose'] = verbose

    if not yes is None:
        config['yes'] = yes

    if not mute is None:
        config['mute'] = mute

    if not dry_run is None:
        config['dry_run'] = dry_run

        if dry_run:
            output.message(
                output.Subject.INFO,
                'Test mode: DRY RUN',
                True
            )

    if not import_file is None:
        config['import'] = import_file

    if not dump_name is None:
        config['dump_name'] = dump_name

    if not host_file is None:
        config['link_hosts'] = host_file

    if not clear is None:
        config['clear_database'] = clear

    if not force_password is None:
        config['force_password'] = force_password

    if not keep_dump is None:
        default_local_sync_path = keep_dump

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

        config['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
    @ToDo Simplify function
    :return:
    """
    if ('link' in config['origin'] or 'link' in config['target']) and config['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, '
                f'when using a link parameter within the configuration.',
                False
            )
        )

    if config['link_hosts'] != '':
        if os.path.isfile(config['link_hosts']):
            with open(config['link_hosts'], 'r') as read_file:
                _hosts = json.load(read_file)
                output.message(
                    output.Subject.INFO,
                    f'Linking configuration with hosts {output.CliFormat.BLACK}{config["link_hosts"]}{output.CliFormat.ENDC}',

                    True
                )
                if not config['config_file_path'] is None:
                    if 'link' in config['origin']:
                        _host_name = str(config['origin']['link']).replace('@', '')
                        if _host_name in _hosts:
                            config['origin'] = {**config['origin'], **_hosts[_host_name]}

                    if 'link' in config['target']:
                        _host_name = str(config['target']['link']).replace('@', '')
                        if _host_name in _hosts:
                            config['target'] = {**config['target'], **_hosts[_host_name]}
                else:
                    if 'link_target' in config and 'link_origin' in config:
                        if config['link_target'] in _hosts and config['link_origin'] in _hosts:
                            config['target'] = _hosts[config['link_target']]
                            config['origin'] = _hosts[config['link_origin']]
                        else:
                            sys.exit(
                                output.message(
                                    output.Subject.ERROR,
                                    f'Misconfiguration of link hosts {config["link_origin"]}, '
                                    f'{config["link_target"]} in {config["link_hosts"]}',
                                    False
                                )
                            )
                    else:
                        sys.exit(
                            output.message(
                                output.Subject.ERROR,
                                f'Missing link hosts for {config["link_hosts"]}',
                                False
                            )
                        )
        else:
            sys.exit(
                output.message(
                    output.Subject.ERROR,
                    f'Local host file not found: {config["link_hosts"]}',
                    False
                )
            )


1			#!/usr/bin/env python3
2			# -- coding: future_fstrings --
3
4			"""
5			System module
6			"""
7
8			import sys
9			import json
10			import os
11			import getpass
12			import yaml
13			from db_sync_tool.utility import log, parser, mode, helper, output, validation
14			from db_sync_tool.remote import utility as remote_utility
15
16			#
17			# GLOBALS
18			#
19
20			config = {
21			'verbose': False,
22			'mute': False,
23			'dry_run': False,
24			'keep_dump': False,
25			'dump_name': '',
26			'import': '',
27			'link_hosts': '',
28			'default_origin_dump_dir': True,
29			'default_target_dump_dir': True,
30			'check_dump': True,
31			'is_same_client': False,
32			'config_file_path': None,
33			'clear_database': False,
34			'force_password': False,
35			'ssh_agent': False,
36			'ssh_password': {
37			'origin': None,
38			'target': None
39			},
40			'link_target': None,
41			'link_origin': None,
42			}
43
44			#
45			# DEFAULTS
46			#
47
48			default_local_sync_path = '/tmp/db_sync_tool/'
			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...
49
50
51			#
52			# FUNCTIONS
53			#
54
55			def check_target_configuration():
56			"""
57			Checking target database configuration
58			:return:
59			"""
60			parser.get_database_configuration(mode.Client.TARGET)
61
62
63			def get_configuration(host_config):
64			"""
65			Checking configuration information by file or dictionary
66			:param host_config: Dictionary
67			:return:
68			"""
69			global config
			0 ignored issues – show Coding Style introduced 2020-12-03 19:28 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-12-03 19:28 UTC by Report Bug Copy Issue Report Constant name "config" 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...
70
71			if config['config_file_path'] is None and host_config == {}:
72			sys.exit(
73			output.message(
74			output.Subject.ERROR,
75			f'Configuration is missing, use a separate file or provide host parameter',
			0 ignored issues – show introduced 2020-12-06 17:30 UTC by Report Bug Copy Issue Report Using an f-string that does not have any interpolated variables Loading history...
76			False
77			)
78			)
79
80			if host_config:
81			config.update(host_config)
82
83			_config_file_path = config['config_file_path']
84			if not _config_file_path is None:
85			if os.path.isfile(_config_file_path):
86			with open(_config_file_path, 'r') as read_file:
87			if _config_file_path.endswith('.json'):
88			config.update(json.load(read_file))
89			elif _config_file_path.endswith('.yaml') or _config_file_path.endswith('.yml'):
90			config.update(yaml.safe_load(read_file))
91			else:
92			sys.exit(
93			output.message(
94			output.Subject.ERROR,
95			f'Unsupported configuration file type [json,yml,yaml]: '
96			f'{config["config_file_path"]}',
97			False
98			)
99			)
100			output.message(
101			output.Subject.LOCAL,
102			f'Loading host configuration '
103			f'{output.CliFormat.BLACK}{_config_file_path}{output.CliFormat.ENDC}',
104			True
105			)
106			else:
107			sys.exit(
108			output.message(
109			output.Subject.ERROR,
110			f'Local configuration not found: {config["config_file_path"]}',
111			False
112			)
113			)
114
115			validation.check(config)
116			check_options()
117			helper.run_script(script='before')
118			log.get_logger().info('Starting db_sync_tool')
119
120
121			def check_options():
122			"""
123			Checking configuration provided file
124			:return:
125			"""
126			global config
			0 ignored issues – show Coding Style introduced 2020-12-03 19:28 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-12-03 19:28 UTC by Report Bug Copy Issue Report Constant name "config" 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...
127			if 'dump_dir' in config['origin']:
128			config['default_origin_dump_dir'] = False
129
130			if 'dump_dir' in config['target']:
131			config['default_target_dump_dir'] = False
132
133			if 'check_dump' in config:
134			config['check_dump'] = config['check_dump']
135
136			link_configuration_with_hosts()
137			mode.check_sync_mode()
138
139
140			def check_authorizations():
141			"""
142			Checking authorization for clients
143			:return:
144			"""
145			check_authorization(mode.Client.ORIGIN)
146			check_authorization(mode.Client.TARGET)
147
148
149			def check_authorization(client):
150			"""
151			Checking arguments and fill options array
152			:param client: String
153			:return:
154			"""
155			# only need authorization if client is remote
156			if mode.is_remote(client):
157			# Workaround if no authorization is needed
158			if (mode.get_sync_mode() == mode.SyncMode.DUMP_REMOTE and
			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...
159			client == mode.Client.TARGET) or \
			0 ignored issues – show Coding Style introduced 2021-04-13 18:13 UTC by Report Bug Copy Issue Report Wrong continued indentation before block (add 4 spaces). Loading history...
160			(mode.get_sync_mode() == mode.SyncMode.DUMP_LOCAL and
161			client == mode.Client.ORIGIN) or \
162			(mode.get_sync_mode() == mode.SyncMode.IMPORT_REMOTE and
163			client == mode.Client.ORIGIN):
164			return
165
166			# ssh key authorization
167			if config['force_password']:
			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...
168			config[client]['password'] = get_password_by_user(client)
169			elif 'ssh_key' in config[client]:
170			_ssh_key = config[client]['ssh_key']
171			if not os.path.isfile(_ssh_key):
172			sys.exit(
173			output.message(
174			output.Subject.ERROR,
175			f'SSH {client} private key not found: {_ssh_key}',
176			False
177			)
178			)
179			elif 'password' in config[client]:
180			config[client]['password'] = config[client]['password']
181			elif remote_utility.check_keys_from_ssh_agent():
182			config['ssh_agent'] = True
183			else:
184			# user input authorization
185			config[client]['password'] = get_password_by_user(client)
186
187			if mode.get_sync_mode() == mode.SyncMode.DUMP_REMOTE and \
188			client == mode.Client.ORIGIN and 'password' in \
189			config[mode.Client.ORIGIN]:
190			config[mode.Client.TARGET]['password'] = config[mode.Client.ORIGIN]['password']
191
192
193			def get_password_by_user(client):
194			"""
195			Getting password by user input
196			:param client: String
197			:return: String password
198			"""
199			_password = getpass.getpass(
200			output.message(
201			output.Subject.INFO,
202			'SSH password ' + helper.get_ssh_host_name(client, True) + ': ',
203			False
204			)
205			)
206
207			while _password.strip() == '':
208			output.message(
209			output.Subject.WARNING,
210			'Password seems to be empty. Please enter a valid password.',
211			True
212			)
213
214			_password = getpass.getpass(
215			output.message(
216			output.Subject.INFO,
217			'SSH password ' + helper.get_ssh_host_name(client, True) + ': ',
218			False
219			)
220			)
221
222			return _password
223
224
225			def check_args_options(config_file=None,
			0 ignored issues – show best-practice introduced 2021-08-31 09:32 UTC by Report Bug Copy Issue Report Too many arguments (11/5) Loading history...
226			verbose=False,
227			yes=False,
228			mute=False,
229			dry_run=False,
230			import_file=None,
231			dump_name=None,
232			keep_dump=None,
233			host_file=None,
234			clear=False,
235			force_password=False):
236			"""
237			Checking arguments and fill options array
238			:param config_file:
239			:param verbose:
240			:param yes:
241			:param mute:
242			:param dry_run:
243			:param import_file:
244			:param dump_name:
245			:param keep_dump:
246			:param host_file:
247			:param clear:
248			:param force_password:
249			:return:
250			"""
251			global config
			0 ignored issues – show Coding Style introduced 2020-12-03 19:28 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-12-03 19:28 UTC by Report Bug Copy Issue Report Constant name "config" 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...
252			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...
253
254			if not config_file is None:
255			config['config_file_path'] = config_file
256
257			if not verbose is None:
258			config['verbose'] = verbose
259
260			if not yes is None:
261			config['yes'] = yes
262
263			if not mute is None:
264			config['mute'] = mute
265
266			if not dry_run is None:
267			config['dry_run'] = dry_run
268
269			if dry_run:
270			output.message(
271			output.Subject.INFO,
272			'Test mode: DRY RUN',
273			True
274			)
275
276			if not import_file is None:
277			config['import'] = import_file
278
279			if not dump_name is None:
280			config['dump_name'] = dump_name
281
282			if not host_file is None:
283			config['link_hosts'] = host_file
284
285			if not clear is None:
286			config['clear_database'] = clear
287
288			if not force_password is None:
289			config['force_password'] = force_password
290
291			if not keep_dump is None:
292			default_local_sync_path = keep_dump
293
294			# Adding trailing slash if necessary
295			if default_local_sync_path[-1] != '/':
296			default_local_sync_path += '/'
297
298			config['keep_dump'] = True
299			output.message(
300			output.Subject.INFO,
301			'"Keep dump" option chosen',
302			True
303			)
304
305
306			def link_configuration_with_hosts():
307			"""
308			Merging the hosts definition with the given configuration file
309			@ToDo Simplify function
310			:return:
311			"""
312			if ('link' in config['origin'] or 'link' in config['target']) and config['link_hosts'] == '':
			0 ignored issues – show Comprehensibility Best Practice introduced 2020-11-27 11:55 UTC by Report Bug Copy Issue Report The variable `config` does not seem to be defined. Loading history...
313			# Try to find default hosts.json file in same directory
314			sys.exit(
315			output.message(
316			output.Subject.ERROR,
317			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...
318			f'Use the "-o" / "--hosts" argument to define the filepath for the hosts file, '
319			f'when using a link parameter within the configuration.',
320			False
321			)
322			)
323
324			if config['link_hosts'] != '':
325			if os.path.isfile(config['link_hosts']):
326			with open(config['link_hosts'], 'r') as read_file:
327			_hosts = json.load(read_file)
328			output.message(
329			output.Subject.INFO,
330			f'Linking configuration with hosts {output.CliFormat.BLACK}{config["link_hosts"]}{output.CliFormat.ENDC}',
			0 ignored issues – show Coding Style introduced 2020-12-03 19:28 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (126/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
331			True
332			)
333			if not config['config_file_path'] is None:
334			if 'link' in config['origin']:
335			_host_name = str(config['origin']['link']).replace('@', '')
336			if _host_name in _hosts:
337			config['origin'] = {config['origin'], _hosts[_host_name]}
338
339			if 'link' in config['target']:
340			_host_name = str(config['target']['link']).replace('@', '')
341			if _host_name in _hosts:
342			config['target'] = {config['target'], _hosts[_host_name]}
343			else:
344			if 'link_target' in config and 'link_origin' in config:
345			if config['link_target'] in _hosts and config['link_origin'] in _hosts:
346			config['target'] = _hosts[config['link_target']]
347			config['origin'] = _hosts[config['link_origin']]
348			else:
349			sys.exit(
350			output.message(
351			output.Subject.ERROR,
352			f'Misconfiguration of link hosts {config["link_origin"]}, '
353			f'{config["link_target"]} in {config["link_hosts"]}',
354			False
355			)
356			)
357			else:
358			sys.exit(
359			output.message(
360			output.Subject.ERROR,
361			f'Missing link hosts for {config["link_hosts"]}',
362			False
363			)
364			)
365			else:
366			sys.exit(
367			output.message(
368			output.Subject.ERROR,
369			f'Local host file not found: {config["link_hosts"]}',
370			False
371			)
372			)
373

jackd248 / db-sync-tool

Push — master ( 48167f...cc4cd6 )

db_sync_tool.utility.system.check_options() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like