db_sync_tool.utility.mode.SyncMode.is_proxy() - Code Metrics - Inspection of "[Task] Sync modes "Sync local" and "Sync remote"" - jackd248/db-sync-tool - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 95757f...1637f4 )

by Konrad

created 2021-03-07 12:03 UTC

db_sync_tool.utility.mode.SyncMode.is_proxy() A

↳ Parent: db_sync_tool.utility.mode

Complexity

Conditions

Size

Total Lines	7
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	3
dl	0
loc	7
rs	10
c	0
b	0
f	0
cc	1
nop	0

#!/usr/bin/env python3

# -*- coding: future_fstrings -*-

import subprocess
import sys

from db_sync_tool.utility import system, output, helper
from db_sync_tool.remote import system as remote_system


#
# GLOBALS
#

class Client:

    ORIGIN = 'origin'
    TARGET = 'target'
    LOCAL = 'local'


class SyncMode:
    """
    Sync Mode
    """

    DUMP_LOCAL = 'DUMP_LOCAL'
    DUMP_REMOTE = 'DUMP_REMOTE'
    IMPORT_LOCAL = 'IMPORT_LOCAL'
    IMPORT_REMOTE = 'IMPORT_REMOTE'
    RECEIVER = 'RECEIVER'
    SENDER = 'SENDER'
    PROXY = 'PROXY'
    SYNC_REMOTE = 'SYNC_REMOTE'
    SYNC_LOCAL = 'SYNC_LOCAL'

    @staticmethod
    def is_dump_local():
        """

        :return: boolean
        """
        return SyncMode.is_full_local() and SyncMode.is_same_host() and not SyncMode.is_sync_local()

    @staticmethod
    def is_dump_remote():
        """

        :return: boolean
        """
        return SyncMode.is_full_remote() and SyncMode.is_same_host() and not SyncMode.is_sync_remote()


    @staticmethod
    def is_receiver():
        """

        :return: boolean
        """
        return 'host' in system.config[Client.ORIGIN] and not SyncMode.is_proxy() and not SyncMode.is_sync_remote()


    @staticmethod
    def is_sender():
        """

        :return: boolean
        """
        return 'host' in system.config[Client.TARGET] and not SyncMode.is_proxy() and not SyncMode.is_sync_remote()


    @staticmethod
    def is_proxy():
        """

        :return: boolean
        """
        return SyncMode.is_full_remote()

    @staticmethod
    def is_import_local():
        """

        :return: boolean
        """
        return system.config['import'] != '' and SyncMode.is_full_local()

    @staticmethod
    def is_import_remote():
        """

        :return: boolean
        """
        return system.config['import'] != '' and 'host' in system.config[Client.TARGET]

    @staticmethod
    def is_sync_local():
        """

        :return: boolean
        """
        return SyncMode.is_full_local() and SyncMode.is_same_host()  and SyncMode.is_same_sync()

    @staticmethod
    def is_sync_remote():
        """

        :return: boolean
        """
        return SyncMode.is_full_remote() and SyncMode.is_same_host() and SyncMode.is_same_sync()

    @staticmethod
    def is_same_sync():

        return ((SyncMode.is_available_configuration('path') and not SyncMode.is_same_configuration('path')) or

               (SyncMode.is_available_configuration('db') and not SyncMode.is_same_configuration('db')))


    @staticmethod
    def is_full_remote():
        """

        :return: boolean
        """
        return SyncMode.is_available_configuration('host')

    @staticmethod
    def is_full_local():
        """

        :return: boolean
        """
        return SyncMode.is_unavailable_configuration('host')

    @staticmethod
    def is_same_host():
        """

        :return: boolean
        """
        return SyncMode.is_same_configuration('host') and SyncMode.is_same_configuration('port')

    @staticmethod
    def is_available_configuration(key):
        """

        :return: boolean
        """
        return key in system.config[Client.ORIGIN] and key in system.config[Client.TARGET]

    @staticmethod
    def is_unavailable_configuration(key):
        """

        :return: boolean
        """
        return key not in system.config[Client.ORIGIN] and key not in system.config[Client.TARGET]

    @staticmethod
    def is_same_configuration(key):
        """

        :return: boolean
        """
        return (SyncMode.is_available_configuration(key) and

               system.config[Client.ORIGIN][key] == system.config[Client.TARGET][key]) or \

               SyncMode.is_unavailable_configuration(key)


# Default sync mode
sync_mode = SyncMode.RECEIVER



#
# FUNCTIONS
#
def get_sync_mode():
    """
    Returning the sync mode
    :return: String sync_mode
    """
    return sync_mode


def check_sync_mode():
    """
    Checking the sync_mode based on the given configuration
    :return: String subject
    """
    global sync_mode

    _description = ''

    _modes = {
        SyncMode.RECEIVER: '(REMOTE ➔ LOCAL)',
        SyncMode.SENDER: '(LOCAL ➔ REMOTE)',
        SyncMode.PROXY: '(REMOTE ➔ LOCAL ➔ REMOTE)',
        SyncMode.DUMP_LOCAL: '(LOCAL, ONLY EXPORT)',
        SyncMode.DUMP_REMOTE: '(REMOTE, ONLY EXPORT)',
        SyncMode.IMPORT_LOCAL: '(REMOTE, ONLY IMPORT)',
        SyncMode.IMPORT_REMOTE: '(LOCAL, ONLY IMPORT)',
        SyncMode.SYNC_LOCAL: '(LOCAL ➔ LOCAL)',
        SyncMode.SYNC_REMOTE: '(REMOTE ➔ REMOTE)'
    }

    for _mode, _desc in _modes.items():
        if getattr(SyncMode, 'is_' + _mode.lower())():
            sync_mode = _mode
            _description = _desc

    if is_import():
        output.message(
            output.Subject.INFO,
            f'Import file {output.CliFormat.BLACK}{system.config["import"]}{output.CliFormat.ENDC}',
            True
        )

    system.config['is_same_client'] = SyncMode.is_same_host()

    output.message(
        output.Subject.INFO,
        f'Sync mode: {sync_mode} {output.CliFormat.BLACK}{_description}{output.CliFormat.ENDC}',
        True
    )


def is_remote(client):

    """
    Check if given client is remote client
    :param client: String
    :return: Boolean
    """
    if client == Client.ORIGIN:

        return is_origin_remote()
    elif client == Client.TARGET:
        return is_target_remote()
    elif client == Client.LOCAL:
        return False


def is_target_remote():
    """
    Check if target is remote client
    :return: Boolean
    """
    return sync_mode == SyncMode.SENDER or sync_mode == SyncMode.PROXY or sync_mode == SyncMode.DUMP_REMOTE or \

           sync_mode == SyncMode.IMPORT_REMOTE or sync_mode == SyncMode.SYNC_REMOTE


def is_origin_remote():
    """
    Check if origin is remote client
    :return: Boolean
    """
    return sync_mode == SyncMode.RECEIVER or sync_mode == SyncMode.PROXY or sync_mode == SyncMode.DUMP_REMOTE or \

           sync_mode == SyncMode.IMPORT_REMOTE or sync_mode == SyncMode.SYNC_REMOTE


def is_import():
    """
    Check if sync mode is import
    :return: Boolean
    """
    return sync_mode == SyncMode.IMPORT_LOCAL or sync_mode == SyncMode.IMPORT_REMOTE



def is_dump():
    """
    Check if sync mode is import
    :return: Boolean
    """
    return sync_mode == SyncMode.DUMP_LOCAL or sync_mode == SyncMode.DUMP_REMOTE



def run_command(command, client, force_output=False, allow_fail=False, skip_dry_run=False):

    """
    Run command depending on the given client
    :param command: String
    :param client: String
    :param force_output: Boolean
    :param allow_fail: Boolean
    :param skip_dry_run: Boolean
    :return:
    """
    if system.config['verbose']:
        output.message(
            output.host_to_subject(client),
            output.CliFormat.BLACK + command + output.CliFormat.ENDC,
            debug=True
        )

    if system.config['dry_run'] and skip_dry_run:
        return

    if is_remote(client):
        if force_output:

            return ''.join(remote_system.run_ssh_command_by_client(client, command).readlines()).strip()

        else:
            return remote_system.run_ssh_command_by_client(client, command)
    else:
        res = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
        # Wait for the process end and print error in case of failure
        out, err = res.communicate()

        if res.wait() != 0 and err.decode() != '' and not allow_fail:
            helper.run_script(script='error')
            sys.exit(output.message(output.Subject.ERROR, err.decode(), False))

        if force_output:
            return out.decode().strip()


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: future_fstrings --
3
4			import subprocess
5			import sys
6
7			from db_sync_tool.utility import system, output, helper
8			from db_sync_tool.remote import system as remote_system
9
10
11			#
12			# GLOBALS
13			#
14
15			class Client:
			0 ignored issues – show introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report Missing class docstring Loading history...
16			ORIGIN = 'origin'
17			TARGET = 'target'
18			LOCAL = 'local'
19
20
21			class SyncMode:
22			"""
23			Sync Mode
24			"""
25
26			DUMP_LOCAL = 'DUMP_LOCAL'
27			DUMP_REMOTE = 'DUMP_REMOTE'
28			IMPORT_LOCAL = 'IMPORT_LOCAL'
29			IMPORT_REMOTE = 'IMPORT_REMOTE'
30			RECEIVER = 'RECEIVER'
31			SENDER = 'SENDER'
32			PROXY = 'PROXY'
33			SYNC_REMOTE = 'SYNC_REMOTE'
34			SYNC_LOCAL = 'SYNC_LOCAL'
35
36			@staticmethod
37			def is_dump_local():
38			"""
39
40			:return: boolean
41			"""
42			return SyncMode.is_full_local() and SyncMode.is_same_host() and not SyncMode.is_sync_local()
43
44			@staticmethod
45			def is_dump_remote():
46			"""
47
48			:return: boolean
49			"""
50			return SyncMode.is_full_remote() and SyncMode.is_same_host() and not SyncMode.is_sync_remote()
			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 (102/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
51
52			@staticmethod
53			def is_receiver():
54			"""
55
56			:return: boolean
57			"""
58			return 'host' in system.config[Client.ORIGIN] and not SyncMode.is_proxy() and not SyncMode.is_sync_remote()
			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 (115/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
59
60			@staticmethod
61			def is_sender():
62			"""
63
64			:return: boolean
65			"""
66			return 'host' in system.config[Client.TARGET] and not SyncMode.is_proxy() and not SyncMode.is_sync_remote()
			0 ignored issues – show Coding Style introduced 2021-02-19 18:09 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (115/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
67
68			@staticmethod
69			def is_proxy():
70			"""
71
72			:return: boolean
73			"""
74			return SyncMode.is_full_remote()
75
76			@staticmethod
77			def is_import_local():
78			"""
79
80			:return: boolean
81			"""
82			return system.config['import'] != '' and SyncMode.is_full_local()
83
84			@staticmethod
85			def is_import_remote():
86			"""
87
88			:return: boolean
89			"""
90			return system.config['import'] != '' and 'host' in system.config[Client.TARGET]
91
92			@staticmethod
93			def is_sync_local():
94			"""
95
96			:return: boolean
97			"""
98			return SyncMode.is_full_local() and SyncMode.is_same_host() and SyncMode.is_same_sync()
99
100			@staticmethod
101			def is_sync_remote():
102			"""
103
104			:return: boolean
105			"""
106			return SyncMode.is_full_remote() and SyncMode.is_same_host() and SyncMode.is_same_sync()
107
108			@staticmethod
109			def is_same_sync():
			0 ignored issues – show introduced 2021-03-07 12:04 UTC by Report Bug Copy Issue Report Missing function or method docstring Loading history...
110			return ((SyncMode.is_available_configuration('path') and not SyncMode.is_same_configuration('path')) or
			0 ignored issues – show Coding Style introduced 2021-02-19 18:09 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (111/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
111			(SyncMode.is_available_configuration('db') and not SyncMode.is_same_configuration('db')))
			0 ignored issues – show Coding Style introduced 2020-11-27 11:55 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (104/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history... Coding Style introduced 2021-03-07 12:04 UTC by Report Bug Copy Issue Report Wrong continued indentation (add 1 space). Loading history...
112
113			@staticmethod
114			def is_full_remote():
115			"""
116
117			:return: boolean
118			"""
119			return SyncMode.is_available_configuration('host')
120
121			@staticmethod
122			def is_full_local():
123			"""
124
125			:return: boolean
126			"""
127			return SyncMode.is_unavailable_configuration('host')
128
129			@staticmethod
130			def is_same_host():
131			"""
132
133			:return: boolean
134			"""
135			return SyncMode.is_same_configuration('host') and SyncMode.is_same_configuration('port')
136
137			@staticmethod
138			def is_available_configuration(key):
139			"""
140
141			:return: boolean
142			"""
143			return key in system.config[Client.ORIGIN] and key in system.config[Client.TARGET]
144
145			@staticmethod
146			def is_unavailable_configuration(key):
147			"""
148
149			:return: boolean
150			"""
151			return key not in system.config[Client.ORIGIN] and key not in system.config[Client.TARGET]
152
153			@staticmethod
154			def is_same_configuration(key):
155			"""
156
157			:return: boolean
158			"""
159			return (SyncMode.is_available_configuration(key) and
			0 ignored issues – show Unused Code introduced 2021-03-07 12:04 UTC by Report Bug Copy Issue Report Consider using ternary (system.config[Client.ORIGIN][key] == system.config[Client.TARGET][key] if SyncMode.is_available_configuration(key) else SyncMode.is_unavailable_configuration(key)) Loading history...
160			system.config[Client.ORIGIN][key] == system.config[Client.TARGET][key]) or \
			0 ignored issues – show Coding Style introduced 2021-03-07 12:04 UTC by Report Bug Copy Issue Report Wrong continued indentation (add 1 space). Loading history...
161			SyncMode.is_unavailable_configuration(key)
162
163
164			# Default sync mode
165			sync_mode = SyncMode.RECEIVER
			0 ignored issues – show Coding Style Naming introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report Constant name "sync_mode" 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...
166
167
168			#
169			# FUNCTIONS
170			#
171			def get_sync_mode():
172			"""
173			Returning the sync mode
174			:return: String sync_mode
175			"""
176			return sync_mode
177
178
179			def check_sync_mode():
180			"""
181			Checking the sync_mode based on the given configuration
182			:return: String subject
183			"""
184			global sync_mode
			0 ignored issues – show Coding Style Naming introduced 2020-12-28 13:48 UTC by Report Bug Copy Issue Report Constant name "sync_mode" 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... Coding Style introduced 2020-12-28 13:48 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...
185			_description = ''
186
187			_modes = {
188			SyncMode.RECEIVER: '(REMOTE ➔ LOCAL)',
189			SyncMode.SENDER: '(LOCAL ➔ REMOTE)',
190			SyncMode.PROXY: '(REMOTE ➔ LOCAL ➔ REMOTE)',
191			SyncMode.DUMP_LOCAL: '(LOCAL, ONLY EXPORT)',
192			SyncMode.DUMP_REMOTE: '(REMOTE, ONLY EXPORT)',
193			SyncMode.IMPORT_LOCAL: '(REMOTE, ONLY IMPORT)',
194			SyncMode.IMPORT_REMOTE: '(LOCAL, ONLY IMPORT)',
195			SyncMode.SYNC_LOCAL: '(LOCAL ➔ LOCAL)',
196			SyncMode.SYNC_REMOTE: '(REMOTE ➔ REMOTE)'
197			}
198
199			for _mode, _desc in _modes.items():
200			if getattr(SyncMode, 'is_' + _mode.lower())():
201			sync_mode = _mode
202			_description = _desc
203
204			if is_import():
205			output.message(
206			output.Subject.INFO,
207			f'Import file {output.CliFormat.BLACK}{system.config["import"]}{output.CliFormat.ENDC}',
208			True
209			)
210
211			system.config['is_same_client'] = SyncMode.is_same_host()
212
213			output.message(
214			output.Subject.INFO,
215			f'Sync mode: {sync_mode} {output.CliFormat.BLACK}{_description}{output.CliFormat.ENDC}',
216			True
217			)
218
219
220			def is_remote(client):
			0 ignored issues – show Unused Code introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report Either all return statements in a function should return an expression, or none of them should. Loading history...
221			"""
222			Check if given client is remote client
223			:param client: String
224			:return: Boolean
225			"""
226			if client == Client.ORIGIN:
			0 ignored issues – show unused-code introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report Unnecessary "elif" after "return" Loading history...
227			return is_origin_remote()
228			elif client == Client.TARGET:
229			return is_target_remote()
230			elif client == Client.LOCAL:
231			return False
232
233
234			def is_target_remote():
235			"""
236			Check if target is remote client
237			:return: Boolean
238			"""
239			return sync_mode == SyncMode.SENDER or sync_mode == SyncMode.PROXY or sync_mode == SyncMode.DUMP_REMOTE or \
			0 ignored issues – show Coding Style introduced 2021-03-07 12:04 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (112/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history... Unused Code introduced 2021-03-07 12:04 UTC by Report Bug Copy Issue Report Consider merging these comparisons with "in" to 'sync_mode in (SyncMode.SENDER, SyncMode.PROXY, SyncMode.DUMP_REMOTE, SyncMode.IMPORT_REMOTE, SyncMode.SYNC_REMOTE)' Loading history...
240			sync_mode == SyncMode.IMPORT_REMOTE or sync_mode == SyncMode.SYNC_REMOTE
241
242
243			def is_origin_remote():
244			"""
245			Check if origin is remote client
246			:return: Boolean
247			"""
248			return sync_mode == SyncMode.RECEIVER or sync_mode == SyncMode.PROXY or sync_mode == SyncMode.DUMP_REMOTE or \
			0 ignored issues – show Coding Style introduced 2021-03-07 12:04 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (114/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history... Unused Code introduced 2021-03-07 12:04 UTC by Report Bug Copy Issue Report Consider merging these comparisons with "in" to 'sync_mode in (SyncMode.RECEIVER, SyncMode.PROXY, SyncMode.DUMP_REMOTE, SyncMode.IMPORT_REMOTE, SyncMode.SYNC_REMOTE)' Loading history...
249			sync_mode == SyncMode.IMPORT_REMOTE or sync_mode == SyncMode.SYNC_REMOTE
250
251
252			def is_import():
253			"""
254			Check if sync mode is import
255			:return: Boolean
256			"""
257			return sync_mode == SyncMode.IMPORT_LOCAL or sync_mode == SyncMode.IMPORT_REMOTE
			0 ignored issues – show Unused Code introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report Consider merging these comparisons with "in" to 'sync_mode in (SyncMode.IMPORT_LOCAL, SyncMode.IMPORT_REMOTE)' Loading history...
258
259
260			def is_dump():
261			"""
262			Check if sync mode is import
263			:return: Boolean
264			"""
265			return sync_mode == SyncMode.DUMP_LOCAL or sync_mode == SyncMode.DUMP_REMOTE
			0 ignored issues – show Unused Code introduced 2021-03-07 12:04 UTC by Report Bug Copy Issue Report Consider merging these comparisons with "in" to 'sync_mode in (SyncMode.DUMP_LOCAL, SyncMode.DUMP_REMOTE)' Loading history...
266
267
268			def run_command(command, client, force_output=False, allow_fail=False, skip_dry_run=False):
			0 ignored issues – show Unused Code introduced 2020-12-28 16:47 UTC by Report Bug Copy Issue Report Either all return statements in a function should return an expression, or none of them should. Loading history...
269			"""
270			Run command depending on the given client
271			:param command: String
272			:param client: String
273			:param force_output: Boolean
274			:param allow_fail: Boolean
275			:param skip_dry_run: Boolean
276			:return:
277			"""
278			if system.config['verbose']:
279			output.message(
280			output.host_to_subject(client),
281			output.CliFormat.BLACK + command + output.CliFormat.ENDC,
282			debug=True
283			)
284
285			if system.config['dry_run'] and skip_dry_run:
286			return
287
288			if is_remote(client):
289			if force_output:
			0 ignored issues – show unused-code introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report Unnecessary "else" after "return" Loading history...
290			return ''.join(remote_system.run_ssh_command_by_client(client, command).readlines()).strip()
			0 ignored issues – show Coding Style introduced 2021-03-07 12:04 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (104/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
291			else:
292			return remote_system.run_ssh_command_by_client(client, command)
293			else:
294			res = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
295			# Wait for the process end and print error in case of failure
296			out, err = res.communicate()
297
298			if res.wait() != 0 and err.decode() != '' and not allow_fail:
299			helper.run_script(script='error')
300			sys.exit(output.message(output.Subject.ERROR, err.decode(), False))
301
302			if force_output:
303			return out.decode().strip()
304

jackd248 / db-sync-tool

Push — master ( 95757f...1637f4 )

db_sync_tool.utility.mode.SyncMode.is_proxy() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like