From 5d846c13f70a030fb970facacc1adc3fb5ed30be Mon Sep 17 00:00:00 2001 From: Ding-Yi Chen Date: Mon, 16 Apr 2018 17:58:37 +1000 Subject: [PATCH] refactor --- JenkinsHelper.py | 243 +++++++++++++++++++++++------------------------ ZanataWar.py | 47 +++++---- 2 files changed, 147 insertions(+), 143 deletions(-) diff --git a/JenkinsHelper.py b/JenkinsHelper.py index 37b8b6d..2818430 100755 --- a/JenkinsHelper.py +++ b/JenkinsHelper.py @@ -10,27 +10,74 @@ import os import os.path import re +import sys # 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 -ZANATA_JENKINS = { - 'server': os.environ.get('JENKINS_URL'), - 'user': os.environ.get('ZANATA_JENKINS_USER'), - 'token': os.environ.get('ZANATA_JENKINS_TOKEN'), - } +try: + # We need to import 'List' and 'Any' for mypy to work + from typing import List, Any # noqa: F401 # pylint: disable=unused-import +except ImportError: + sys.stderr.write("python typing module is not installed" + os.linesep) -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 JenkinsServer(object): + """Jenkins Helper functions""" + def __init__(self, server_url, user, token): + # type: (str, str,str) -> None + self.url_helper = UrlHelper( + server_url, user, token) + self.server_url = server_url + self.user = user + self.token = token + + def __getitem__(self, key): + # type: (str) -> str + return self[key] + + @staticmethod + def init_default(): + # type: () -> None + """Init JenkinsServer connection with default environment.""" + zanata_jenkins = { + 'server_url': os.environ.get('JENKINS_URL'), + 'user': os.environ.get('ZANATA_JENKINS_USER'), + 'token': os.environ.get('ZANATA_JENKINS_TOKEN'), + } + + if not zanata_jenkins['server_url']: + 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'") + + return JenkinsServer( + zanata_jenkins['server_url'], + zanata_jenkins['user'], + zanata_jenkins['token'], + ) + + @staticmethod + def create_parent_parser(): + # type () -> argparse.ArgumentParser + """Create a parser as parent of Jenkins job argument parser + e.g. -F , -b and """ + job_parent_parser = argparse.ArgumentParser(add_help=False) + job_parent_parser.add_argument( + '-F', '--folder', type=str, default='', + help='folder name') + job_parent_parser.add_argument( + '-b', '--branch', type=str, default='', + help='branch or PR name') + job_parent_parser.add_argument('job', type=str, help='job name') + return job_parent_parser + + +class JenkinsJob(object): + """Jenkins Job Objects""" -class JenkinsJobBuild(object): - """Build object for Jenkins job""" @staticmethod def dict_get_elem_by_path(dic, path): # type (dict, str) -> object @@ -52,80 +99,21 @@ def print_key_value(key, value): """Pretty print the key and value""" return "%30s : %s" % (key, value) - def __init__(self, parent_job, build_number, build_url): - # type (object, int, str) -> None - self.parent_job = parent_job - self.number = build_number - self.url = build_url - self.content = None - - def __getitem__(self, key): - # type: (str) -> str - return self[key] - - def get_elem(self, path): - # type: (str) -> object - """Get element from the build object""" - return JenkinsJobBuild.dict_get_elem_by_path(self.content, path) - - def load(self): - """Load the build object from Jenkins server""" - logging.info("Loading build from %sapi/python", self.url) - self.content = ast.literal_eval(UrlHelper.read( - "%s/api/python" % self.url)) - - def list_artifacts_related_paths(self, artifact_path_pattern='.*'): - # type: (str) -> List[str] - """Return a List of relativePaths of artifacts - that matches the path pattern""" - if not self.content: - self.load() - if not self.content: - raise AssertionError("Failed to load build from %s" % self.url) - return map( - lambda y: y['relativePath'], - filter( - lambda x: re.search( - artifact_path_pattern, - x['relativePath']), - self.content['artifacts'])) - - def __repr__(self): - # type: () -> str - result = "\n".join([ - JenkinsJobBuild.print_key_value( - tup[0], str(tup[1])) for tup in [ - ['number', self.number], - ['url', self.url]]]) - - if self.content: - result += "\n\n%s" % "\n".join([ - JenkinsJobBuild.print_key_value( - key, self.get_elem(key)) for key in [ - 'nextBuild/number', - 'previousBuild/number']]) - result += "\n\nArtifacts:\n%s" % "\n ".join( - self.list_artifacts_related_paths()) - return result - - -class JenkinsJob(object): - """Jenkins Job Objects""" def get_elem(self, path): # type: (str) -> object """Get element from the job object""" - return JenkinsJobBuild.dict_get_elem_by_path(self.content, path) + return JenkinsJob.dict_get_elem_by_path(self.content, path) def __repr__(self): # type: () -> str result = "\n".join([ - JenkinsJobBuild.print_key_value(tup[0], tup[1]) for tup in [ + JenkinsJob.print_key_value(tup[0], tup[1]) for tup in [ ['name', self.name], ['folder', self.folder], ['branch', self.branch]]]) if self.content: result += "\n\n%s" % "\n".join([ - JenkinsJobBuild.print_key_value( + JenkinsJob.print_key_value( key, self.get_elem(key)) for key in [ 'displayName', 'fullName', @@ -133,10 +121,11 @@ def __repr__(self): 'lastCompletedBuild/number', 'lastFailedBuild/number', 'lastSuccessfulBuild/number']]) - return result + return result - def __init__(self, name, folder, branch): - # type (str, str, str) -> None + def __init__(self, server, name, folder, branch): + # type (JenkinsServer, str, str, str) -> None + self.server = server self.name = name self.folder = folder self.branch = branch @@ -146,7 +135,7 @@ def __init__(self, name, folder, branch): job_path = "job/%s/%s" % (folder, job_path) if branch: job_path += "/job/%s" % branch - self.url = "%s%s" % (ZANATA_JENKINS['server'], job_path) + self.url = "%s%s" % (self.server.server_url, job_path) def __getitem__(self, key): # type: (str) -> str @@ -173,66 +162,76 @@ def get_last_successful_build(self): self.get_elem('lastSuccessfulBuild/url')) -class JenkinsServer(object): - """Jenkins Helper functions""" - def __init__(self, server, user, token): - # type: (str, str,str) -> None - self.url_helper = UrlHelper( - server, user, token) - self.server = server +class JenkinsJobBuild(object): + """Build object for Jenkins job""" + + def __init__(self, parent_job, build_number, build_url): + # type (object, int, str) -> None + self.parent_job = parent_job + self.number = build_number + self.url = build_url + self.content = None def __getitem__(self, key): # type: (str) -> str return self[key] - @staticmethod - def init_default(): - # type: () -> None - """Init JenkinsServer connection with default environment.""" - JenkinsServer( - ZANATA_JENKINS['server'], - ZANATA_JENKINS['user'], - ZANATA_JENKINS['token'], - ) + def get_elem(self, path): + # type: (str) -> object + """Get element from the build object""" + return JenkinsJob.dict_get_elem_by_path(self.content, path) - @staticmethod - def load_job(job_name, folder, branch): - # type (str, str, str) -> JenkinsJob - """Load the job object - Once the JenkinsServer object created,""" - job = JenkinsJob(job_name, folder, branch) - job.load() - return job + def load(self): + """Load the build object from Jenkins server""" + logging.info("Loading build from %sapi/python", self.url) + self.content = ast.literal_eval(UrlHelper.read( + "%s/api/python" % self.url)) - @staticmethod - def create_parent_parser(): - # type () -> argparse.ArgumentParser - """Create a parser as parent of Jenkins job argument parser - e.g. -F , -b and """ - job_parent_parser = argparse.ArgumentParser(add_help=False) - job_parent_parser.add_argument( - '-F', '--folder', type=str, default='', - help='folder name') - job_parent_parser.add_argument( - '-b', '--branch', type=str, default='', - help='branch or PR name') - job_parent_parser.add_argument('job', type=str, help='job name') - return job_parent_parser + def list_artifacts_related_paths(self, artifact_path_pattern='.*'): + # type: (str) -> List[str] + """Return a List of relativePaths of artifacts + that matches the path pattern""" + if not self.content: + self.load() + if not self.content: + raise AssertionError("Failed to load build from %s" % self.url) + return [ + artifact['relativePath'] + for artifact in self.content['artifacts'] + if re.search(artifact_path_pattern, artifact['relativePath'])] + + def __repr__(self): + # type: () -> str + result = "\n".join([ + JenkinsJob.print_key_value( + tup[0], str(tup[1])) for tup in [ + ['number', self.number], + ['url', self.url]]]) + + if self.content: + result += "\n\n%s" % "\n".join([ + JenkinsJob.print_key_value( + key, self.get_elem(key)) for key in [ + 'nextBuild/number', + 'previousBuild/number']]) + result += "\n\nArtifacts:\n%s" % "\n ".join( + self.list_artifacts_related_paths()) + return result def show_job(): # type () -> None """Show the job information""" - JenkinsServer.init_default() - job = JenkinsServer.load_job(args.job, args.folder, args.branch) + server = JenkinsServer.init_default() + job = JenkinsJob(server, args.job, args.folder, args.branch) print(job) def show_last_successful_build(): # type () -> None """Show the last successful build for a Jenkins job""" - JenkinsServer.init_default() - job = JenkinsServer.load_job(args.job, args.folder, args.branch) + server = JenkinsServer.init_default() + job = JenkinsJob(server, args.job, args.folder, args.branch) build = job.get_last_successful_build() build.load() diff --git a/ZanataWar.py b/ZanataWar.py index d9601bd..c4a5620 100755 --- a/ZanataWar.py +++ b/ZanataWar.py @@ -1,24 +1,27 @@ #!/usr/bin/env python """Zanata WAR Helper functions""" -#from typing import List, Any # noqa: F401 # pylint: disable=unused-import -# We need to import 'List' and 'Any' for mypy to work - import argparse import os +import sys 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 +import JenkinsHelper + +try: + # We need to import 'List' and 'Any' for mypy to work + from typing import List, Any # noqa: F401 # pylint: disable=unused-import +except ImportError: + sys.stderr.write("python typing module is not installed" + os.linesep) class ZanataWar(object): """Class that manipulate zanata.war""" def __init__(self, download_url=None, local_path=None): - # type (JenkinsJobBuild, str) -> None + # type (str, str) -> None self.download_url = download_url self.local_path = local_path @@ -37,10 +40,11 @@ def create_parent_parser(): return parent_parser @staticmethod - def get_last_successful_build_url(job_name, folder, branch): - # type (str, str, str) -> str + def get_last_successful_build_url(server, job_name, folder, branch): + # type (JenkinsHelper.JenkinsServer, str, str, str) -> str """Get the WAR file download url for last successful build""" - job = JenkinsServer.load_job(job_name, folder, branch) + job = JenkinsHelper.JenkinsJob(server, job_name, folder, branch) + job.load() build = job.get_last_successful_build() build.load() war_paths = build.list_artifacts_related_paths( @@ -90,17 +94,17 @@ def scp_to_server( # pylint: disable=too-many-arguments ssh_host.scp_to_remote(local_path, target_path, True, rm_old) if chmod: - ssh_host.run_command( + ssh_host.run_check_call( "chown jboss:jboss %s" % target_path, True) def show_download_link(): # type () -> None """Show the download link of last successful build""" - JenkinsServer.init_default() + server = JenkinsHelper.JenkinsServer.init_default() war = ZanataWar(ZanataWar.get_last_successful_build_url( - args.job, args.folder, args.branch)) + server, args.job, args.folder, args.branch)) print(war.download_url) @@ -108,10 +112,10 @@ def download_from_jenkins(): # type () -> None """Handling download-from-jenkins command (download last successful build)""" - JenkinsServer.init_default() + server = JenkinsHelper.JenkinsServer.init_default() war = ZanataWar(ZanataWar.get_last_successful_build_url( - args.job, args.folder, args.branch)) + server, args.job, args.folder, args.branch)) war.download( None if not args.local_path else os.path.basename(args.local_path), None if not args.local_path else os.path.dirname(args.local_path)) @@ -120,11 +124,11 @@ def download_from_jenkins(): def scp_to_server(): # type () -> None """Handling scp-to-server command""" - JenkinsServer.init_default() + server = JenkinsHelper.JenkinsServer.init_default() war = ZanataWar( ZanataWar.get_last_successful_build_url( - args.job, args.folder, args.branch), + server, args.job, args.folder, args.branch), args.local_path) war.scp_to_server(args.host, args.dest_path, args.identity_file) @@ -140,12 +144,12 @@ def deploy_local_war(war=None): war = ZanataWar(local_path=args.local_path) ssh_host = SshHost(args.host, args.identity_file) - ssh_host.run_command("systemctl stop eap7-standalone", True) + ssh_host.run_check_call("systemctl stop eap7-standalone", True) war.scp_to_server( args.host, args.dest_path, args.identity_file, True, True) - ssh_host.run_command("systemctl start eap7-standalone", True) + ssh_host.run_check_call("systemctl start eap7-standalone", True) def deploy(): @@ -153,12 +157,13 @@ def deploy(): """Download the last successfully built WAR, then deploy the WAR file to zanata server. This assumes login as root""" - JenkinsServer.init_default() + server = JenkinsHelper.JenkinsServer.init_default() war = ZanataWar( ZanataWar.get_last_successful_build_url( - args.job, args.folder, args.branch), + server, args.job, args.folder, args.branch), args.local_path) + war.download( None if not args.local_path else os.path.basename(args.local_path), None if not args.local_path else os.path.dirname(args.local_path)) @@ -170,7 +175,7 @@ def parse(): # type () -> None """Parse options and arguments""" parser = argparse.ArgumentParser(description='WAR file functions') - job_parent_parser = JenkinsServer.create_parent_parser() + job_parent_parser = JenkinsHelper.JenkinsServer.create_parent_parser() war_parent_parser = ZanataWar.create_parent_parser() ssh_parent_parser = SshHost.create_parent_parser()