# encoding: utf-8
import json
import xml.etree.ElementTree as ET
from functools import partial
from pathlib import PurePosixPath
from urllib.parse import unquote_plus
from .credential import Credentials
from .item import Item, append_slash, snake
from .mix import (ConfigurationMixIn, DeletionMixIn, DescriptionMixIn,
EnableMixIn)
from .queue import QueueItem
from .view import Views
[docs]
class Job(Item, ConfigurationMixIn, DescriptionMixIn, DeletionMixIn):
[docs]
def move(self, path):
path = path.strip('/')
params = {'destination': f'/{path}',
'json': json.dumps({'destination': f'/{path}'})}
resp = self.handle_req('POST', 'move/move',
data=params, allow_redirects=False)
self.url = resp.headers['Location']
return resp
[docs]
def rename(self, name):
resp = self.handle_req('POST', 'confirmRename',
params={'newName': name},
allow_redirects=False)
self.url = append_slash(resp.headers['Location'])
return resp
[docs]
def duplicate(self, path, recursive=False):
self.jenkins.create_job(path, self.configure(), recursive=recursive)
@property
def parent(self):
path = PurePosixPath(self.full_name)
if path.parent.name == '':
return self.jenkins
return self.jenkins.get_job(str(path.parent))
@property
def name(self):
return self.full_name.split('/')[-1]
@property
def full_name(self):
return unquote_plus(self.jenkins._url2name(self.url))
@property
def full_display_name(self):
return unquote_plus(self.full_name.replace('/', ' ยป '))
[docs]
class Folder(Job):
[docs]
def create(self, name, xml):
return self.handle_req('POST', 'createItem', params={'name': name},
headers=self.headers, data=xml)
[docs]
def get(self, name):
for item in self.api_json(tree='jobs[name,url]')['jobs']:
if name == item['name']:
return self._new_instance_by_item(__name__, item)
return None
[docs]
def iter(self, depth=0):
query = 'jobs[url]'
query_format = 'jobs[url,%s]'
for _ in range(int(depth)):
query = query_format % query
def _resolve(item):
yield self._new_instance_by_item(__name__, item)
jobs = item.get('jobs')
if jobs:
for job in jobs:
yield from _resolve(job)
for item in self.api_json(tree=query)['jobs']:
yield from _resolve(item)
[docs]
def copy(self, src, dest):
params = {'name': dest, 'mode': 'copy', 'from': src}
return self.handle_req('POST', 'createItem', params=params,
allow_redirects=False)
[docs]
def reload(self):
return self.handle_req('POST', 'reload')
@property
def views(self):
return Views(self)
@property
def credentials(self):
return Credentials(self.jenkins, f'{self.url}credentials/store/folder/')
def __iter__(self):
yield from self.iter()
def __call__(self, depth):
yield from self.iter(depth)
def __getitem__(self, name):
return self.get(name)
[docs]
class WorkflowMultiBranchProject(Folder, EnableMixIn):
[docs]
def scan(self, delay=0):
return self.handle_req('POST', 'build', params={'delay': delay})
[docs]
def get_scan_log(self, stream=False):
with self.handle_req('GET', 'indexing/consoleText',
stream=stream) as resp:
yield from resp.iter_lines()
@property
def buildable(self):
return ET.XML(self.configure()).find('disabled').text == 'false'
[docs]
class OrganizationFolder(WorkflowMultiBranchProject):
pass
[docs]
class Project(Job, EnableMixIn):
def __init__(self, jenkins, url):
super().__init__(jenkins, url)
def _get_build_by_key(key):
item = self.api_json(tree=f'{key}[url]')[key]
if item is None:
return None
return self._new_instance_by_item('api4jenkins.build', item)
for key in ['firstBuild', 'lastBuild', 'lastCompletedBuild',
'lastFailedBuild', 'lastStableBuild', 'lastUnstableBuild',
'lastSuccessfulBuild', 'lastUnsuccessfulBuild']:
setattr(self, snake(f'get_{key}'),
partial(_get_build_by_key, key))
[docs]
def build(self, **params):
reserved = ['token', 'delay']
if not params or all(k in reserved for k in params):
entry = 'build'
else:
entry = 'buildWithParameters'
resp = self.handle_req('POST', entry, params=params)
return QueueItem(self.jenkins, resp.headers['Location'])
[docs]
def get_build(self, number):
for item in self.api_json(tree='builds[number,displayName,url]')['builds']:
if number == item['number'] or number == item['displayName']:
return self._new_instance_by_item('api4jenkins.build', item)
return None
[docs]
def iter_builds(self):
yield from self
[docs]
def iter_all_builds(self):
for item in self.api_json(tree='allBuilds[number,url]')['allBuilds']:
yield self._new_instance_by_item('api4jenkins.build', item)
[docs]
def set_next_build_number(self, number):
self.handle_req('POST', 'nextbuildnumber/submit',
params={'nextBuildNumber': number})
[docs]
def get_parameters(self):
params = []
for p in self.api_json()['property']:
if 'parameterDefinitions' in p:
params = p['parameterDefinitions']
return params
@property
def building(self):
builds = self.api_json(tree='builds[building]')['builds']
return any(b['building'] for b in builds)
def __iter__(self):
for item in self.api_json(tree='builds[number,url]')['builds']:
yield self._new_instance_by_item('api4jenkins.build', item)
def __getitem__(self, number):
return self.get_build(number)
[docs]
def filter_builds_by_result(self, *, result):
"""filter build by build results, avaliable results are:
'SUCCESS', 'UNSTABLE', 'FAILURE', 'NOT_BUILT', 'ABORTED'
see: https://javadoc.jenkins-ci.org/hudson/model/Result.html
"""
expect = ['SUCCESS', 'UNSTABLE', 'FAILURE', 'NOT_BUILT', 'ABORTED']
if result not in expect:
raise ValueError(f'Expect one of {expect}')
yield from filter(lambda build: build.result == result, self)
[docs]
class WorkflowJob(Project):
pass
[docs]
class MatrixProject(Project):
pass
[docs]
class FreeStyleProject(Project):
pass
[docs]
class MavenModuleSet(Project):
pass
[docs]
class ExternalJob(Project):
pass
[docs]
class MultiJobProject(Project):
pass
[docs]
class IvyModuleSet(Project):
pass
[docs]
class BitbucketSCMNavigator(Project):
pass
[docs]
class GitHubSCMNavigator(Project):
pass
[docs]
class PipelineMultiBranchDefaultsProject(Project):
pass