system-config/tools/test-jenkins-api.py

#!/usr/bin/env python
#
# Test all of the Jenkins API features used by the
# OpenStack Infrastructure project
#
# Copyright (C) 2013 OpenStack Foundation
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
#
# See the License for the specific language governing permissions and
# limitations under the License.

import urllib2
import urlparse
from pprint import pprint
import time
from uuid import uuid4
import ConfigParser
import os
import re

import sys
sys.path.insert(0, '../../zuul/zuul')
from launcher.jenkins import ExtendedJenkins

sys.path.insert(0, '../../devstack-gate/')
from myjenkins import Jenkins as DGJenkins

JOB_NAME = 'test-job'
NODE_NAME = 'test-node'


class JenkinsTest(object):
    def __init__(self):
        self.config = ConfigParser.ConfigParser()
        if len(sys.argv) < 2:
            print "Usage: %s zuul.conf" % sys.argv[0]
            sys.exit(1)
        fp = sys.argv[1]
        if os.path.exists(os.path.expanduser(fp)):
            self.config.read(os.path.expanduser(fp))

        server = self.config.get('jenkins', 'server')
        user = self.config.get('jenkins', 'user')
        apikey = self.config.get('jenkins', 'apikey')

        self.jenkins = ExtendedJenkins(server, user, apikey)
        self.dgjenkins = DGJenkins(server, user, apikey)

    def nodeExists(self):
        return self.dgjenkins.node_exists(NODE_NAME)

    def createNode(self):
        assert not self.nodeExists()
        priv_key = '/var/lib/jenkins/.ssh/id_rsa'
        self.dgjenkins.create_node(
            NODE_NAME, numExecutors=1,
            nodeDescription='Test node',
            remoteFS='/home/jenkins',
            labels='testnode',
            exclusive=True,
            launcher='hudson.plugins.sshslaves.SSHLauncher',
            launcher_params={'port': 22,
                             'username': 'jenkins',
                             'privatekey': priv_key,
                             'host': 'nowhere.example.com'})
        assert self.nodeExists()

    def reconfigNode(self):
        LABEL_RE = re.compile(r'<label>.*</label>')
        config = self.dgjenkins.get_node_config(NODE_NAME)
        assert '<label>testnode</label>' in config
        config = LABEL_RE.sub('<label>devstack-used</label>', config)
        self.dgjenkins.reconfig_node(NODE_NAME, config)
        config = self.dgjenkins.get_node_config(NODE_NAME)
        assert '<label>devstack-used</label>' in config

    def deleteNode(self):
        assert self.nodeExists()
        self.dgjenkins.delete_node(NODE_NAME)
        assert not self.nodeExists()

    def findBuildInQueue(self, build):
        for item in self.jenkins.get_queue_info():
            if 'actions' not in item:
                continue
            for action in item['actions']:
                if 'parameters' not in action:
                    continue
                parameters = action['parameters']
                for param in parameters:
                    # UUID is deprecated in favor of ZUUL_UUID
                    if ((param['name'] in ['ZUUL_UUID', 'UUID'])
                        and build == param['value']):
                        return item
        return False

    def addJob(self, quiet_period):
        assert not self.jobExists()
        xml = open('jenkins-job.xml').read()
        xml = xml % quiet_period

        self.jenkins.create_job(JOB_NAME, xml)
        assert self.jobExists()

    def reconfigJob(self, quiet_period):
        assert self.jobExists()
        xml = open('jenkins-job.xml').read()
        xml = xml % quiet_period

        self.jenkins.reconfig_job(JOB_NAME, xml)
        xml2 = self.jenkins.get_job_config(JOB_NAME)
        s = '<quietPeriod>%s</quietPeriod>' % quiet_period
        assert s in xml2

    def jobExists(self):
        return self.jenkins.job_exists(JOB_NAME)

    def deleteJob(self):
        assert self.jobExists()
        self.jenkins.delete_job(JOB_NAME)
        assert not self.jobExists()

    def getJobs(self):
        pprint(self.jenkins.get_jobs())

    def testCancelQueue(self):
        uuid = str(uuid4().hex)
        self.jenkins.build_job(JOB_NAME, parameters=dict(UUID=uuid))

        item = self.findBuildInQueue(uuid)
        assert item
        self.jenkins.cancel_queue(item['id'])
        assert not self.findBuildInQueue(uuid)

    def testCancelBuild(self):
        uuid = str(uuid4().hex)
        self.jenkins.build_job(JOB_NAME, parameters=dict(UUID=uuid))

        assert self.findBuildInQueue(uuid)
        for x in range(60):
            if not self.findBuildInQueue(uuid):
                break
        assert not self.findBuildInQueue(uuid)
        time.sleep(1)

        buildno = self.jenkins.get_job_info(JOB_NAME)['lastBuild']['number']
        info = self.jenkins.get_build_info(JOB_NAME, buildno)
        assert info['building']
        self.jenkins.stop_build(JOB_NAME, buildno)
        time.sleep(1)
        info = self.jenkins.get_build_info(JOB_NAME, buildno)
        assert not info['building']

        console_url = urlparse.urljoin(info['url'], 'consoleFull')
        self.jenkins.jenkins_open(urllib2.Request(console_url))

        self.jenkins.set_build_description(JOB_NAME, buildno,
                                           "test description")

        info = self.jenkins.get_build_info(JOB_NAME, buildno)
        assert info['description'] == 'test description'

j = JenkinsTest()
if j.nodeExists():
    j.deleteNode()
j.createNode()
j.reconfigNode()
j.deleteNode()
if j.jobExists():
    j.deleteJob()
j.addJob(5)
j.reconfigJob(10)
j.testCancelQueue()
j.testCancelBuild()
j.deleteJob()