Source code for api4jenkins.node

# encoding: utf-8

import json

from .exceptions import ItemNotFoundError
from .item import Item, new_item
from .mix import ConfigurationMixIn, DeletionMixIn, RunScriptMixIn

[docs]class Nodes(Item): ''' classdocs '''
[docs] def create(self, name, **kwargs): node_setting = { 'nodeDescription': '', 'numExecutors': 1, 'remoteFS': '/home/jenkins', 'labelString': '', 'mode': 'NORMAL', 'retentionStrategy': { 'stapler-class': 'hudson.slaves.RetentionStrategy$Always' }, 'nodeProperties': {'stapler-class-bag': 'true'}, 'launcher': {'stapler-class': 'hudson.slaves.JNLPLauncher'} } node_setting.update(kwargs) params = { 'name': name, 'type': 'hudson.slaves.DumbSlave$DescriptorImpl', 'json': json.dumps(node_setting) } self.handle_req('POST', 'doCreateItem', data=params)
[docs] def get(self, name): for item in self.api_json(tree='computer[displayName]')['computer']: if name == item['displayName']: item['url'] = f"{self.url}{item['displayName']}/" return self._new_instance_by_item(__name__, item) return None
[docs] def iter_builds(self): builds = {} # iterate 'executors', 'oneOffExecutors' in order, # cause freestylebuild is in executors, and _class of workflowbuild in # executors is PlaceholderExecutable tree = ('computer[executors[currentExecutable[url]],' 'oneOffExecutors[currentExecutable[url]]]') for computer in self.api_json(tree, 2)['computer']: _parse_builds(computer, builds) yield from _new_items(self.jenkins, builds)
def __iter__(self): for item in self.api_json(tree='computer[displayName]')['computer']: item['url'] = f"{self.url}{item['displayName']}/" yield self._new_instance_by_item(__name__, item)
# following two functions should be used in this module only def _new_items(jenkins, builds): for url, class_name in builds.items(): item = {'url': url, '_class': class_name} yield new_item(jenkins, '', item) def _parse_builds(data, builds): for kind in ['executors', 'oneOffExecutors']: for executor in data.get(kind): # in case of issue: # execable = executor['currentExecutable'] if not execable: continue if execable['_class'].endswith('PlaceholderExecutable'): execable['_class'] = 'org.jenkinsci.plugins.workflow.job.WorkflowRun' builds[execable['url']] = execable['_class']
[docs]class Node(Item, ConfigurationMixIn, DeletionMixIn, RunScriptMixIn):
[docs] def enable(self): if self.offline: self.handle_req('POST', 'toggleOffline', params={'offlineMessage': ''})
[docs] def disable(self, msg=''): if not self.offline: self.handle_req('POST', 'toggleOffline', params={'offlineMessage': msg})
[docs] def iter_builds(self): builds = {} # iterate 'executors', 'oneOffExecutors' in order, # cause freestylebuild is in executors tree = ('executors[currentExecutable[url]],' 'oneOffExecutors[currentExecutable[url]]') _parse_builds(self.api_json(tree, 2), builds) yield from _new_items(self.jenkins, builds)
def __iter__(self): yield from self.iter_builds()
[docs]class MasterComputer(Node): def __init__(self, jenkins, url): # rename built-in node: name = 'master' if url.endswith('/master/') else 'built-in' super().__init__(jenkins, f'{jenkins.url}computer/({name})/')
[docs]class SlaveComputer(Node): pass
[docs]class KubernetesComputer(Node):
[docs] def exists(self): try: self.handle_req('GET', '') return True except ItemNotFoundError: return False
[docs]class DockerComputer(Node): pass