Skip to content

Commit

Permalink
refactor
Browse files Browse the repository at this point in the history
  • Loading branch information
@definite
definite committed Apr 30, 2018
1 parent 849f8a1 commit 5d846c1
Show file tree
Hide file tree
Showing 2 changed files with 147 additions and 143 deletions.
243 changes: 121 additions & 122 deletions JenkinsHelper.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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 <folder>, -b <branch> and <job>"""
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
Expand All @@ -52,91 +99,33 @@ 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',
'lastBuild/number',
'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
Expand All @@ -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
Expand All @@ -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 <folder>, -b <branch> and <job>"""
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()
Expand Down
Loading

0 comments on commit 5d846c1

Please sign in to comment.