diff --git a/JenkinsHelper.py b/JenkinsHelper.py new file mode 100755 index 0000000..52f5ceb --- /dev/null +++ b/JenkinsHelper.py @@ -0,0 +1,310 @@ +#!/usr/bin/env python +"""Jenkins Helper functions +It contains jenkins helper +Run JenkinsHelper --help or JenkinsHelper --help for +detail help.""" + +from __future__ import ( + absolute_import, division, print_function, unicode_literals) +import ast +import logging +import os +import os.path +import re +import sys + +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) + +from ZanataFunctions import UrlHelper +from ZanataArgParser import ZanataArgParser # pylint: disable=E0401 + + +class JenkinsServer(object): + """JenkinsServer can connect to a Jenkins server""" + 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 + + @classmethod + def add_parser(cls, arg_parser=None, env_sub_commands=None): + # type: (ZanataArgParser) -> ZanataArgParser + """Add JenkinsServer parameters to a parser""" + if not arg_parser: + arg_parser = ZanataArgParser(description=__doc__) + + # Add env + arg_parser.add_env( + 'JENKINS_URL', dest='server_url', required=True, + sub_commands=env_sub_commands) + arg_parser.add_env( + 'ZANATA_JENKINS_USER', dest='user', required=True, + sub_commands=env_sub_commands) + arg_parser.add_env( + 'ZANATA_JENKINS_TOKEN', dest='token', required=True, + sub_commands=env_sub_commands) + return arg_parser + + @classmethod + def init_from_parsed_args(cls, args): + """New an instance from parsed args""" + return cls(args.server_url, args.user, args.token) + + +class JenkinsJob(object): + """JenkinsJob object can access a Jenkins Job""" + + @staticmethod + def dict_get_elem_by_path(dic, path): + # type (dict, str) -> object + """Return the elem in python dictionary given path + for example: you can use a/b to retrieve answer from following + dict: + { 'a': { 'b': 'answer' }}""" + obj = dic + for key in path.split('/'): + if obj[key]: + obj = obj[key] + else: + return None + return obj + + @staticmethod + def print_key_value(key, value): + # type (str, str) -> None + """Pretty print the key and value""" + return "%30s : %s" % (key, value) + + def get_elem(self, path): + # type: (str) -> object + """Get element from the job object""" + return JenkinsJob.dict_get_elem_by_path(self.content, path) + + def __repr__(self): + # type: () -> str + result = "\n".join([ + JenkinsJob.print_key_value(tup[0], tup[1]) for tup in [ + ['job_name', self.job_name], + ['folder', self.folder], + ['branch', self.branch]]]) + if self.content: + result += "\n\n%s" % "\n".join([ + JenkinsJob.print_key_value( + key, self.get_elem(key)) for key in [ + 'displayName', + 'fullName', + 'lastBuild/number', + 'lastCompletedBuild/number', + 'lastFailedBuild/number', + 'lastSuccessfulBuild/number']]) + return result + + def __init__(self, server, job_name, folder='', branch=''): + # type (JenkinsServer, str, str, str) -> None + self.server = server + self.job_name = job_name + self.folder = folder + self.branch = branch + self.content = None + job_path = "job/%s" % self.job_name + if folder: + job_path = "job/%s/%s" % (folder, job_path) + if branch: + job_path += "/job/%s" % branch + self.url = "%s%s" % (self.server.server_url, job_path) + + @classmethod + def add_parser( + cls, arg_parser=None, + only_options=False, env_sub_commands=None): + # type: (ZanataArgParser, bool) -> ZanataArgParser + """Add JenkinsJob parameters to parser + arg_parser: existing parser to be appended to + only_options: Add only options and JenkinsServer env""" + if not arg_parser or not arg_parser.has_env('JENKINS_URL'): + arg_parser = JenkinsServer.add_parser( + arg_parser, env_sub_commands) + arg_parser.add_common_argument( + '-b', '--branch', type=str, + help='branch or PR name') + arg_parser.add_common_argument( + '-F', '--folder', type=str, + help='GitHub Organization Folder') + if not only_options: + arg_parser.add_common_argument( + 'job_name', type=str, help='job name') + + # Add sub commands + arg_parser.add_sub_command( + 'get-job', None, + help='Show job objects') + arg_parser.add_sub_command( + 'get-last-successful-build', None, + help=cls.get_last_successful_build.__doc__) + arg_parser.add_sub_command( + 'get-last-successful-artifacts', + { + '-p --artifact-path-patterns': { + 'type': str, 'default': '.*', + 'help': 'comma split artifact path regex pattern'}}, # noqa: E501, #pylint: disable=line-too-long + help='Get matching last-successful artifacts. Default: .*') + arg_parser.add_sub_command( + 'download-last-successful-artifacts', + { + '-p --artifact-path-patterns': { + 'type': str, 'default': '.*', + 'help': 'comma split artifact path regex' + }, + '-d --download-dir': { + 'type': str, 'default': '.', + 'help': 'Download directory'} + }, + help='Get matching last-successful artifacts. Default: .*') + return arg_parser + + @classmethod + def init_from_parsed_args(cls, args): + """New an instance from parsed args""" + server = JenkinsServer.init_from_parsed_args(args) + kwargs = {'job_name': args.job_name} + for k in ['folder', 'branch']: + if hasattr(args, k): + kwargs[k] = getattr(args, k) + return cls(server, **kwargs) + + def load(self): + # type: () -> None + """Load the build object from Jenkins server""" + logging.debug("Loading job from %s/api/python", self.url) + self.content = ast.literal_eval(UrlHelper.read( + "%s/api/python" % self.url)) + + def get_last_successful_build(self): + # type: () -> JenkinsJobBuild + """Get last successful build""" + if not self.content: + self.load() + + if not self.content: + raise AssertionError("Failed to load job from %s" % self.url) + return JenkinsJobBuild( + self, + int(self.get_elem('lastSuccessfulBuild/number')), + self.get_elem('lastSuccessfulBuild/url')) + + def get_last_successful_artifacts( + self, artifact_path_patterns=None): + # type: (List[str]) -> List[str] + """Get last successful artifacts that matches patterns""" + build = self.get_last_successful_build() + return build.list_artifacts_related_paths(artifact_path_patterns) + + def download_last_successful_artifacts( + self, artifact_path_patterns=None, download_dir='.'): + # type: (List[str])-> List[str] + """Download last successful artifacts that matches patterns. + Returns related path of artifacts + + Note the directory structure will be flattern.""" + if not artifact_path_patterns: + artifact_path_patterns = ['.*'] + build = self.get_last_successful_build() + artifact_path_list = build.list_artifacts_related_paths( + artifact_path_patterns) + for artifact_path in artifact_path_list: + UrlHelper.download_file( + build.url + 'artifact/' + artifact_path, + download_dir=download_dir) + return artifact_path_list + + +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 get_elem(self, path): + # type: (str) -> object + """Get element from the build object""" + return JenkinsJob.dict_get_elem_by_path(self.content, path) + + def load(self): + """Load the build object from Jenkins server""" + logging.debug("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_patterns=None): + # type: (str) -> List[str] + """Return a List of relativePaths of artifacts + that matches the path patterns""" + if not artifact_path_patterns: + artifact_path_patterns = ['.*'] + if not self.content: + self.load() + if not self.content: + raise AssertionError("Failed to load build from %s" % self.url) + result = [] + for artifact in self.content['artifacts']: + for pattern in artifact_path_patterns: + if re.search(pattern, artifact['relativePath']): + result.append(artifact['relativePath']) + break # Only append once + return result + + 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 run_sub_command(args): + # type (ZanataArgParser.Namespace) -> None + """Run the sub command""" + job = JenkinsJob.init_from_parsed_args(args) + job.load() + if args.sub_command == 'get-job': + print(job) + elif args.sub_command == 'get-last-successful-build': + build = job.get_last_successful_build() + build.load() + print(build) + elif args.sub_command == 'get-last-successful-artifacts': + print('\n'.join( + job.get_last_successful_artifacts( + args.artifact_path_patterns.split(',')) + )) + elif args.sub_command == 'download-last-successful-artifacts': + artifact_path_list = job.download_last_successful_artifacts( + args.artifact_path_patterns.split(','), + args.download_dir) + print("Downloaded files %s" % '\n'.join(artifact_path_list)) + + +if __name__ == '__main__': + run_sub_command(JenkinsJob.add_parser().parse_all()) diff --git a/ZanataServer.py b/ZanataServer.py new file mode 100755 index 0000000..c64a8e3 --- /dev/null +++ b/ZanataServer.py @@ -0,0 +1,136 @@ +#!/usr/bin/env python +"""Zanata Server Helper functions""" + +from __future__ import ( + absolute_import, division, print_function, unicode_literals) +import logging +import os +import os.path +import sys + + +from ZanataArgParser import ZanataArgParser # pylint: disable=E0401 +from ZanataFunctions import SshHost +from JenkinsHelper import JenkinsServer, JenkinsJob + +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 ZanataServer(SshHost): + """zanata.Server Class""" + def __init__( + self, host, ssh_user=None, identity=None, + deploy_dir='/var/opt/rh/eap7/lib/wildfly/standalone/deployments'): + # type (str, str, str) -> None + """New an instance + src_url: Source URL of zanata WAR file + local_war_path: The local path to the WAR file""" + super(ZanataServer, self).__init__(host, ssh_user, identity) + self.deploy_dir = deploy_dir + self.jboss_user = 'jboss' + self.jboss_group = 'jboss' + + @classmethod + def add_parser(cls, arg_parser=None): + # type (ZanataArgParser) -> ZanataArgParser + """Add Zanata Server parameters to a parser""" + if not arg_parser: + arg_parser = ZanataArgParser(description=__doc__) + arg_parser = super(ZanataServer, cls).add_parser(arg_parser) + + arg_parser.add_sub_command( + 'deploy-war-file', + { + 'war_file': { + 'type': str, + 'help': 'WAR file'} + }, + help=cls.deploy_war_file.__doc__) + + # Include JenkinsJob parser for following sub command + if not arg_parser.has_common_argument(dest='branch'): + arg_parser = JenkinsJob.add_parser( + arg_parser, True, ['deploy-from-jenkins-last-successful']) + arg_parser.add_sub_command( + 'deploy-from-jenkins-last-successful', + None, + help=cls.deploy_from_jenkins_last_successful.__doc__) + return arg_parser + + @staticmethod + def download_last_successful_war( + jenkins_server, branch, + folder, download_dir='.'): + # type (JenkinsServer, str, str) -> str + """Dowload last successful war from jenkins""" + logging.debug( + "download_last_successful_war_from_jenkins(%s, %s, %s)", + jenkins_server.server_url, branch, folder) + jenkins_job = JenkinsJob( + jenkins_server, 'zanata-platform', folder=folder, + branch=branch) + jenkins_job.load() + war_file_list = jenkins_job.download_last_successful_artifacts( + [r'zanata-war/.*/zanata.*\.war'], + download_dir) + return download_dir + '/' + os.path.basename(war_file_list[0]) + + def deploy_war_file( + self, war_file, rm_old=True, + scp_dest_dir='/usr/local/share/applications'): + # type (str, bool, str) -> None + """scp WAR file to server""" + dest_war = "%s/zanata.war" % scp_dest_dir + tmp_dest_war = dest_war + '.tmp' + + self.scp_to_host( + war_file, tmp_dest_war, sudo=True, rm_old=rm_old) + self.run_chown(self.jboss_user, self.jboss_group, tmp_dest_war) + + # Link war file to dest + self.run_check_call("systemctl stop eap7-standalone", True) + # mv to dest_war after eap7 is stopped + self.run_check_call("mv -f %s %s" % (tmp_dest_war, dest_war), True) + deploy_war_file = "%s/zanata.war" % self.deploy_dir + # -n is required in case symlink zanata.war is a directory + self.run_check_call( + "ln -fsn %s %s" % (dest_war, deploy_war_file), True) + self.run_chown(self.jboss_user, self.jboss_group, deploy_war_file) + self.run_check_call("systemctl start eap7-standalone", True) + + logging.info("Done") + + def deploy_from_jenkins_last_successful( + self, jenkins_server, + branch='master', + folder='github-zanata-org', rm_old=True): + # type () -> None + """Download from last successful WAR file from jenkins, then deploy""" + downloaded_war = ZanataServer.download_last_successful_war( + jenkins_server, branch, folder) + self.deploy_war_file(downloaded_war, rm_old) + + +def run_sub_command(args): + # type (dict) -> None + """Run the sub command""" + z_server = ZanataServer.init_from_parsed_args(args) + + if args.sub_command == 'deploy-war-file': + z_server.deploy_war_file(args.war_file) + elif args.sub_command == 'deploy-from-jenkins-last-successful': + jenkins_server = JenkinsServer.init_from_parsed_args(args) + kwargs = {} + for key in ['branch', 'folder']: + if hasattr(args, key) and getattr(args, key): + kwargs[key] = getattr(args, key) + z_server.deploy_from_jenkins_last_successful( + jenkins_server, **kwargs) + + +if __name__ == '__main__': + run_sub_command(ZanataServer.add_parser().parse_all())