From ef04885aeb9f08b6a71df3387f514f14a81f843a Mon Sep 17 00:00:00 2001 From: Ding-Yi Chen Date: Thu, 12 Apr 2018 15:45:12 +0800 Subject: [PATCH] Fix codefactor --- JenkinsHelper.py | 20 +++++++++++++------- ZanataFunctions.py | 33 +++++++++++++++++++-------------- ZanataWar.py | 26 +++++++++++++++----------- 3 files changed, 47 insertions(+), 32 deletions(-) diff --git a/JenkinsHelper.py b/JenkinsHelper.py index 47ecf50..2fb2734 100755 --- a/JenkinsHelper.py +++ b/JenkinsHelper.py @@ -13,16 +13,20 @@ from typing import List, Any # noqa: F401 # pylint: disable=unused-import # We need to import 'List' and 'Any' for mypy to work -from ZanataFunctions import UrlHelper, logging_init +from .ZanataFunctions import UrlHelper, logging_init ZANATA_JENKINS = { 'server': os.environ.get('JENKINS_URL'), 'user': os.environ.get('ZANATA_JENKINS_USER'), 'token': os.environ.get('ZANATA_JENKINS_TOKEN'), } -assert ZANATA_JENKINS['server'], "Missing environment 'JENKINS_URL'" -assert ZANATA_JENKINS['user'], "Missing environment 'ZANATA_JENKINS_USER'" -assert ZANATA_JENKINS['token'], "Missing environment 'ZANATA_JENKINS_TOKEN'" + +if not ZANATA_JENKINS['server']: + raise AssertionError("Missing environment 'JENKINS_URL'") +if not ZANATA_JENKINS['user']: + raise AssertionError("Missing environment 'ZANATA_JENKINS_USER'") +if not ZANATA_JENKINS['token']: + raise AssertionError("Missing environment 'ZANATA_JENKINS_TOKEN'") class JenkinsJobBuild(object): @@ -76,7 +80,8 @@ def list_artifacts_related_paths(self, artifact_path_pattern='.*'): that matches the path pattern""" if not self.content: self.load() - assert self.content, "Failed to load build from %s" % self.url + if not self.content: + raise AssertionError("Failed to load build from %s" % self.url) return map( lambda y: y['relativePath'], filter( @@ -160,7 +165,8 @@ def get_last_successful_build(self): if not self.content: self.load() - assert self.content, "Failed to load job from %s" % self.url + if not self.content: + raise AssertionError("Failed to load job from %s" % self.url) return JenkinsJobBuild( self, int(self.get_elem('lastSuccessfulBuild/number')), @@ -263,5 +269,5 @@ def parse(): if __name__ == '__main__': logging_init() - args = parse() + args = parse() # pylint: disable=invalid-name args.func() diff --git a/ZanataFunctions.py b/ZanataFunctions.py index d885154..493150a 100755 --- a/ZanataFunctions.py +++ b/ZanataFunctions.py @@ -7,11 +7,12 @@ import os import subprocess # nosec import sys -import urllib2 # noqa: F401 # pylint: disable=unused-import -import urlparse # noqa: F401 # pylint: disable=unused-import +import urllib2 # noqa: F401 # pylint: disable=import-error,unused-import +import urlparse # noqa: F401 # pylint: disable=import-error,unused-import SCRIPT_DIR = os.path.dirname(os.path.realpath(__file__)) ZANATA_ENV_FILE = os.path.join(SCRIPT_DIR, 'zanata-env.sh') +BASH_CMD = '/bin/bash' def logging_init(level=logging.INFO): @@ -26,8 +27,8 @@ def logging_init(level=logging.INFO): def read_env(filename): # type (str) -> dict """Read environment variables by sourcing a bash file""" - proc = subprocess.Popen( - ['bash', '-c', + proc = subprocess.Popen( # nosec + [BASH_CMD, '-c', "source %s && set -o posix && set" % (filename)], stdout=subprocess.PIPE) return {kv[0]: kv[1] for kv in map( @@ -40,17 +41,21 @@ def read_env(filename): class HTTPBasicAuthHandler(urllib2.HTTPBasicAuthHandler): """Handle Basic Authentication""" - def http_error_401(self, req, error_body, code, msg, headers): # noqa: E501 # pylint: disable=unused-argument, too-many-arguments + @staticmethod + def http_error_401(req, error_body, code, msg, headers): # noqa: E501 # pylint: disable=unused-argument, too-many-arguments """retry with basic auth when facing a 401""" host = req.get_host() realm = None - return self.retry_http_basic_auth(host, req, realm) + return urllib2.HTTPBasicAuthHandler.retry_http_basic_auth( + host, req, realm) - def http_error_403(self, req, error_body, code, msg, headers): # noqa: E501 # pylint: disable=unused-argument,too-many-arguments + @staticmethod + def http_error_403(req, error_body, code, msg, headers): # noqa: E501 # pylint: disable=unused-argument,too-many-arguments """retry with basic auth when facing a 403""" host = req.get_host() realm = None - return self.retry_http_basic_auth(host, req, realm) + return urllib2.HTTPBasicAuthHandler.retry_http_basic_auth( + host, req, realm) class SshHost(object): @@ -94,7 +99,7 @@ def run_command(self, command, sudo=False): ('sudo ' if sudo else '') + command] logging.info(' '.join(cmd_list)) - subprocess.check_call(cmd_list) + subprocess.check_call(cmd_list) # nosec def scp_to_remote( self, source_path, dest_path, @@ -112,7 +117,7 @@ def scp_to_remote( logging.info(' '.join(cmd_list)) - subprocess.check_call(cmd_list) + subprocess.check_call(cmd_list) # nosec class UrlHelper(object): @@ -157,14 +162,14 @@ def download_file(url, dest_file='', dest_dir='.'): raise logging.info("Downloading to %s from %s", target_path, url) - response = urllib2.urlopen(url) + response = urllib2.urlopen(url) # nosec chunk_count = 0 with open(target_path, 'wb') as out_file: while True: - buffer = response.read(chunk) - if not buffer: + buf = response.read(chunk) + if not buf: break - out_file.write(buffer) + out_file.write(buf) chunk_count += 1 if chunk_count % 100 == 0: sys.stderr.write('#') diff --git a/ZanataWar.py b/ZanataWar.py index fb73c8e..11be16b 100755 --- a/ZanataWar.py +++ b/ZanataWar.py @@ -6,12 +6,13 @@ import argparse import os -import urlparse -from ZanataFunctions import logging_init -from ZanataFunctions import SshHost -from ZanataFunctions import UrlHelper -from JenkinsHelper import JenkinsJobBuild # noqa: E501,F401 # pylint: disable=unused-import -from JenkinsHelper import JenkinsServer +import urlparse # pylint: disable=import-error +# python3-pylint does not do well on importing python2 module +from .ZanataFunctions import logging_init +from .ZanataFunctions import SshHost +from .ZanataFunctions import UrlHelper +from .JenkinsHelper import JenkinsJobBuild # noqa: E501,F401 # pylint: disable=unused-import +from .JenkinsHelper import JenkinsServer class ZanataWar(object): @@ -52,7 +53,8 @@ def download( dest_dir=None): # type (str, str) -> None """Download WAR file from .download_url""" - assert self.download_url, "war download_url is not set." + if not self.download_url: + raise AssertionError("war download_url is not set.") target_file = dest_file if not target_file: url_dict = urlparse.urlparse(self.download_url) @@ -62,7 +64,7 @@ def download( UrlHelper.download_file(self.download_url, target_file, target_dir) self.local_path = os.path.join(target_dir, target_file) - def scp_to_server( + def scp_to_server( # pylint: disable=too-many-arguments self, dest_host, dest_path=None, identity_file=None, @@ -72,8 +74,10 @@ def scp_to_server( # type (str, str, str, bool, bool, str) -> None """SCP to Zanata server""" local_path = source_path if source_path else self.local_path - assert local_path, "source_path is missing" - assert os.path.isfile(local_path), local_path + " does not exist" + if not local_path: + raise AssertionError("source_path is missing") + if not os.path.isfile(local_path): + raise AssertionError(local_path + " does not exist") if dest_path: target_path = dest_path @@ -190,5 +194,5 @@ def parse(): # Set logging logging_init() - args = parse() + args = parse() # pylint: disable=invalid-name args.func()