db_sync_tool.remote.client

Complexity

Total Complexity

10

Size/Duplication

Total Lines	102
Duplicated Lines	0 %

Importance

Changes

0

4 Functions

Rating	Name	Duplication	Size	Complexity
A	load_ssh_client_target()	0	8	1
A	close_ssh_clients()	0	12	3
B	load_ssh_client()	0	55	5
A	load_ssh_client_origin()	0	8	1

#!/usr/bin/env python3

Missing module docstring

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

import sys
import paramiko
from db_sync_tool.utility import mode, system, helper, output
from db_sync_tool.remote import utility

ssh_client_origin = None

Constant name "ssh_client_origin" doesn't conform to UPPER_CASE naming style ('([^\\W\\da-z][^\\Wa-z]*|__.*__)$' pattern)

ssh_client_target = None

Constant name "ssh_client_target" doesn't conform to UPPER_CASE naming style ('([^\\W\\da-z][^\\Wa-z]*|__.*__)$' pattern)

def load_ssh_client_origin():
    """
    Loading the origin ssh client
    :return:
    """
    global ssh_client_origin

Usage of the global statement should be avoided.

Constant name "ssh_client_origin" doesn't conform to UPPER_CASE naming style ('([^\\W\\da-z][^\\Wa-z]*|__.*__)$' pattern)

    ssh_client_origin = load_ssh_client(mode.Client.ORIGIN)
    utility.run_before_script(mode.Client.ORIGIN)


def load_ssh_client_target():
    """
    Loading the target ssh client
    :return:
    """
    global ssh_client_target

Usage of the global statement should be avoided.

Constant name "ssh_client_target" doesn't conform to UPPER_CASE naming style ('([^\\W\\da-z][^\\Wa-z]*|__.*__)$' pattern)

    ssh_client_target = load_ssh_client(mode.Client.TARGET)
    utility.run_before_script(mode.Client.TARGET)


def load_ssh_client(ssh):
    """
    Initializing the given ssh client
    :param ssh: String
    :return:
    """
    _host_name = helper.get_ssh_host_name(ssh, True)
    _ssh_client = paramiko.SSHClient()
    _ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

    _ssh_port = _ssh_port = system.config[ssh]['port'] if 'port' in system.config[ssh] else 22
    _ssh_key = None
    _ssh_password = None

    # Check authentication
    if 'ssh_key' in system.config[ssh]:
        _authentication_method = f'{output.CliFormat.BLACK} - (authentication: key){output.CliFormat.ENDC}'

This line is too long as per the coding-style (107/100).

        _ssh_key = system.config[ssh]['ssh_key']
    elif 'password' in system.config[ssh]:
        _authentication_method = f'{output.CliFormat.BLACK} - (authentication: password){output.CliFormat.ENDC}'

This line is too long as per the coding-style (112/100).

        _ssh_password = system.config[ssh]['password']
    else:
        sys.exit(
            output.message(
                output.Subject.ERROR,
                f'Missing SSH authentication. Neither ssh key nor ssh password given.',

Using an f-string that does not have any interpolated variables

                False
            )
        )

    # Try to connect to remote client via paramiko
    try:
        _ssh_client.connect(hostname=system.config[ssh]['host'],
                            username=system.config[ssh]['user'],
                            key_filename=_ssh_key,
                            password=_ssh_password,
                            port=_ssh_port,
                            compress=True)

    except paramiko.ssh_exception.AuthenticationException:
        sys.exit(
            output.message(
                output.Subject.ERROR,
                f'SSH authentication for {_host_name} failed',
                False
            )
        )

    output.message(
        output.host_to_subject(ssh),
        f'Initialize remote SSH connection {_host_name}{_authentication_method}',
        True
    )

    return _ssh_client


def close_ssh_clients():
    """
    Closing ssh client sessions
    :return:
    """
    utility.run_after_script(mode.Client.ORIGIN)
    if not ssh_client_origin is None:
        ssh_client_origin.close()

    utility.run_after_script(mode.Client.TARGET)
    if not ssh_client_target is None:
        ssh_client_target.close()

Final newline missing