Source code for cup.jenkinslib.internal.promotion_build

#!/usr/bin/env python
# -*- coding: utf-8 -*
# Copyright: [CUP] - See LICENSE for details.
# Authors: liushuxian(liushuxian)
"""
This module provides PromotionBuild object.
"""

import contextlib
import logging

from cup.jenkinslib.internal import base
from cup.jenkinslib.internal import exception
from cup.jenkinslib.internal import utils


class PromotionBuild(base.JenkinsBase):
    """Represents a promotion build."""
    def __init__(self, url, build_number, promotion, depth=None):
        """initialize PromotionBuild object

        Args:
            url: url of promotion build.
            build_number: build number.
            promotion: Promotion object.
            depth: set param 'depth' for jenkins api,
                   the more deeper it is, the more data you can get back.
        """
        self.build_number = build_number
        self.promotion = promotion
        self.depth = depth
        super(PromotionBuild, self).__init__(url, static=promotion.is_static)

    def get_jenkins_obj(self):
        """get object of current jenkins."""
        return self.promotion.job.jenkins

    def _poll(self, tree=None):
        """poll out api info.

        Add param 'depth', so we can get more information of upstream and downstream.
        """
        url = self.python_api_url(self.url)
        params = {} if self.depth is None else {'depth': self.depth}
        return self.get_data(url, params=params, tree=tree)

    def __str__(self):
        return self._data['fullDisplayName']

    @property
    def name(self):
        """get build name."""
        return str(self)

    @property
    def number(self):
        """get build number."""
        return self._data["number"]

    @property
    def job_build_number(self):
        """get build number of target job build."""
        return self._data["target"]["number"]

    @property
    def job_build(self):
        """get Build object of target job build."""
        return self.promotion.job[self.job_build_number]

    @property
    def description(self):
        """get description of build if exists, otherwise None.

        Returns:
            description info of this build, encoding in utf-8.
            return None if not exists.
        """
        return self._data["description"]

    @description.setter
    def description(self, description=""):
        """submit description to this build.

        Args:
            description: description content, should encode in utf-8.
        """
        url = "%s/submitDescription" % self.url
        data = {"description": description}
        self.get_jenkins_obj().requester.post_and_confirm_status(url, data=data)
        # update info
        self._data["description"] = self.poll(tree="description")["description"]

    @property
    def result(self):
        """get build result."""
        return self._data["result"]

    @property
    def revision(self):
        """get latest revision, only support svn."""
        vcs = self._data['changeSet']['kind'] or "svn"
        if vcs != "svn":
            raise exception.UnknownVCS(vcs)

        max_revision = 0
        for info in self._data["changeSet"]["revisions"]:
            max_revision = max(info["revision"], max_revision)
        return max_revision

    @property
    def duration_ms(self):
        """get build duration(ms)."""
        return self._data["duration"]

    @property
    def duration(self):
        """get build duration(s)."""
        return int(self.duration_ms / 1000)

    @property
    def is_success(self):
        """check build is success or not, return False if is running."""
        return not self.is_running and self._data["result"] == utils.STATUS_SUCCESS

    @property
    def is_running(self):
        """check build is running or not."""
        data = self.poll(tree='building')
        return data.get("building", False)

    def block(self):
        """block until this build stop running.

        Do not use this method too often,
        because it will request jenkins server every second.
        """
        if self.is_static:
            raise exception.ImappropriateMethodInStaticMode("block")
        while self.is_running:
            time.sleep(1)

        # update info
        self.poll()

    def block_until_complete(self, delay=15, timeout=None):
        """block until this build stop running.

        Args:
            delay: check status every `delay` seconds, default is 15s.
            timeout: wait `timeout` seconds at most, default is forever.

        Returns:
            True if stopped, False if still running.
        """
        if self.is_static:
            raise exception.ImappropriateMethodInStaticMode("block_until_complete")

        stime = time.time()
        while self.is_running:
            waited = time.time() - stime
            logging.info("waited %is for %s to complete" % (waited, str(self)))
            if timeout is not None and timeout < waited:
                raise exception.RunTimeout("wait %is > %is, timeout!" % (waited, timeout))

            time.sleep(delay)

        # update info
        self.poll()
        return True

    @property
    def user_name(self):
        """get name of user who invoke this build if exists, otherwise None."""
        try:
            return self.actions["causes"][0]["userName"]
        except KeyError:
            return None

    @property
    def cause_description(self):
        """get short description of causes."""
        try:
            return self.actions["causes"][0]["shortDescription"]
        except KeyError:
            return ""

    @property
    def is_started_by_timer(self):
        """check if is started by timer."""
        return self.cause_description.startswith("Started by timer")

    @property
    def is_started_by_user(self):
        """check if is started by user."""
        return self.cause_description.startswith("Started by user")

    @property
    def is_started_by_upstream_project(self):
        """check if is started by upstream project."""
        return self.cause_description.startswith("Started by upstream project")

    @property
    def is_started_by_scm_change(self):
        """check if is started by an SCM change."""
        return self.cause_description.startswith("Started by an SCM change")

    @property
    def actions(self):
        """get actions of build."""
        all_actions = {}
        for dct_action in self._data["actions"]:
            if dct_action:
                all_actions.update(dct_action)
        return all_actions

    @property
    def causes(self):
        """get causes of build, empty causes will be ignored."""
        all_causes = []
        for dct_action in self._data["actions"]:
            if dct_action is None:
                continue
            causes = dct_action.get("causes")
            if causes:
                all_causes.extend(causes)
        return all_causes

    @property
    def timestamp(self):
        """get build timestamp."""
        return self._data["timestamp"] / 1000.0

    @property
    def console(self):
        """get text console of build."""
        url = "%s/logText/progressiveText?start=0" % self.url
        with contextlib.closing(self.get_jenkins_obj().requester.get(url)) as fd:
            return fd.content

    def stop(self):
        """Stops the build execution if it's running."""
        if self.is_running:
            url = "%s/stop" % self.url
            self.get_jenkins_obj().requester.post_and_confirm_status(url, data='')

            # update info
            self.poll()
            return True
        return False