Build script interaction with TeamCity using Service Messages

teamcity.py

'''Build script interaction with TeamCity using Service Messages
​
https://confluence.jetbrains.com/display/TCD9/Build+Script+Interaction+with+TeamCity
'''
​
import time
import re
import os
from contextlib import contextmanager
​
​
@contextmanager
def test(test_name, enabled=False):
    test_obj = TeamCityTestWriter(test_name, enabled or is_teamcity_build())
    test_obj.start()
    yield test_obj
    test_obj.finish()  # This will yield an error
​
​
class TeamCityServiceMessageWriter(object):
    def __init__(self, enabled=False):
        self.enabled = enabled or is_teamcity_build()
​
​
    def tc_escape(self, text):
        '''Copied this sweet code from Stack Overflow
        http://stackoverflow.com/a/6117124
        '''
        replacements = {
            "'": "|'",
            "\n": "|n",
            "\r": "|r",
            "|": "||",
            "[": "|[",
            "]": "|]",
        }
        replacements = dict((re.escape(k), v) for k, v in replacements.iteritems())
        pattern = re.compile("|".join(replacements.keys()))
        text = pattern.sub(lambda m: replacements[re.escape(m.group(0))], text)
        return text
​
​
    def print_tc_message(self, event, **kwargs):
        if not self.enabled:
            return
        argstr = ' '.join({k+"='"+self.tc_escape(v)+"'" for k, v in kwargs.items()})
        print("##teamcity["+event+" "+argstr+"]")
​
​
class TeamCityTestWriter(TeamCityServiceMessageWriter):
    '''Write service messages that TeamCity interprets and displays as unit tests
​
    TeamCity requires that start/finish/fail messages are output no more than once per test.
    '''
    def __init__(self, name, enabled=False):
        '''You might want to disable the output during debugging
        '''
        self.name = name
        self.active = False
        super(TeamCityTestWriter, self).__init__(enabled)
​
​
    def start(self):
        if self.active:
            return
        self.print_tc_message('testStarted', name=self.name, captureStandardOutput='false')
        self.active = True
        self.start_time = time.time()
​
​
    def ignore(self, reason):
        self.print_tc_message('testIgnored', name=self.name, message=reason)
​
​
    def finish(self):
        if not self.active:
            return
        self.end_time = time.time()
        duration_ms = (self.end_time - self.start_time) * 1000.0
        self.print_tc_message('testFinished', name=self.name, duration=str(int(duration_ms)))
        self.active = False
​
​
    def fail(self, reason='', details=''):
        if not self.active:
            return
        self.print_tc_message('testFailed', name=self.name, message=reason, details=details)
​
​
def is_teamcity_build():
    '''Decide whether we are running inside a TeamCity Build right now.
    '''
​
    if not hasattr(is_teamcity_build, 'cache'):
        is_teamcity_build.cache = False
        teamcity_environment_vars = [
            'TEAMCITY_PROJECT_NAME',  # eg. Configuration Management
            'TEAMCITY_PROCESS_PARENT_FLOW_ID',  # normally blank
            'TEAMCITY_BUILD_PROPERTIES_FILE',  # eg. /mnt/buildAgent-2/temp/buildTmp/teamcity.build8181951451428436503.properties
            'TEAMCITY_BUILDCONF_NAME',  # eg. test
            'TEAMCITY_GIT_PATH',  # /usr/bin/git
            'TEAMCITY_CAPTURE_ENV',  # some Java stuff lives in here
            'TEAMCITY_VERSION',  # eg. 9.0.2 (build 32195)
            'TEAMCITY_PROCESS_FLOW_ID',  # eg. 16614881948997065
        ]
        for varname in teamcity_environment_vars:
            if os.environ.get(varname) is not None:
                is_teamcity_build.cache = True
                break
    return is_teamcity_build.cache
​
​
class BuildProperty:
    AGENT_HOME_DIR = 'agent.home.dir'
    AGENT_NAME = 'agent.name'
    AGENT_OWNPORT = 'agent.ownPort'
    AGENT_WORK_DIR = 'agent.work.dir'
    BUILD_NUMBER = 'build.number'
    JAVA_IO_TMPDIR = 'java.io.tmpdir'
    TEAMCITY_AGENT_CPUBENCHMARK = 'teamcity.agent.cpuBenchmark'
    TEAMCITY_AGENT_DOTNET_AGENT_URL = 'teamcity.agent.dotnet.agent_url'
    TEAMCITY_AGENT_DOTNET_BUILD_ID = 'teamcity.agent.dotnet.build_id'
    TEAMCITY_AGENT_ENSURE_FREE_SPACE = 'teamcity.agent.ensure.free.space'
    TEAMCITY_AUTH_PASSWORD = 'teamcity.auth.password'
    TEAMCITY_AUTH_USERID = 'teamcity.auth.userId'
    TEAMCITY_BUILD_CHANGEDFILES_FILE = 'teamcity.build.changedFiles.file'
    TEAMCITY_BUILD_CHECKOUTDIR = 'teamcity.build.checkoutDir'
    TEAMCITY_BUILD_ID = 'teamcity.build.id'
    TEAMCITY_BUILD_PROPERTIES_FILE = 'teamcity.build.properties.file'
    TEAMCITY_BUILD_TEMPDIR = 'teamcity.build.tempDir'
    TEAMCITY_BUILD_WORKINGDIR = 'teamcity.build.workingDir'
    TEAMCITY_BUILDCONFNAME = 'teamcity.buildConfName'
    TEAMCITY_BUILDTYPE_ID = 'teamcity.buildType.id'
    TEAMCITY_CONFIGURATION_PROPERTIES_FILE = 'teamcity.configuration.properties.file'
    TEAMCITY_PROJECTNAME = 'teamcity.projectName'
    TEAMCITY_RUNNER_PROPERTIES_FILE = 'teamcity.runner.properties.file'
    TEAMCITY_TESTS_RECENTLYFAILEDTESTS_FILE = 'teamcity.tests.recentlyFailedTests.file'
    TEAMCITY_VERSION = 'teamcity.version'
​
​
def build_property(property_key):
    if not hasattr(build_property, 'cache'):
        if not is_teamcity_build():
            raise Exception('Cannot get build properties when not running in TeamCity')
        props = {}
        with open(os.environ.get('TEAMCITY_BUILD_PROPERTIES_FILE'), 'r') as propfile:
            for line in propfile.readlines():
                if line[0:1] == '#':
                    continue
                (key, sep, value) = line.partition('=')
                props[key] = value.rstrip()
        build_property.cache = props
    return build_property.cache[property_key]
​
​
class ConfigurationViewTabs:
    OVERVIEW='buildTypeStatusDiv'
    HISTORY='buildTypeHistoryList'
    CHANGE_LOG='buildTypeChangeLog'
    ISSUE_LOG='buildTypeIssueLog'
    STATISTICS='buildTypeStatistics'
    COMPATIBLE_AGENTS='compatibilityList'
    PENDING_CHANGES='pendingChangesDiv'
    SETTINGS='buildTypeSettings'
    WEBHOOKS='webHooks'
​
​
def link_to_this_configuration(tab=ConfigurationViewTabs.OVERVIEW):
    '''Generate link to the currently running TeamCity build
​
    TeamCity structures things thusly:
    - Project
      - Configuration
        - Build
        - Build
    '''
​
    teamcity_url = os.environ.get('TEAMCITY_URL')
    if teamcity_url is None:
        raise Exception('TEAMCITY_URL environment variable is not defined. TeamCity does not provide this by default; you MUST populate env.TEAMCITY_URL in your build configuration.')
    arguments = {
        'buildId': build_property(BuildProperty.TEAMCITY_BUILD_ID),
        'buildTypeId': build_property(BuildProperty.TEAMCITY_BUILDTYPE_ID),
        'tab': tab,
    }
    query_string = '&'.join([k+'='+str(v) for k,v in arguments.iteritems()])
    return teamcity_url + '/viewLog.html?' + query_string
​
​
class BuildViewTabs:
    OVERVIEW='buildResultsDiv'
    CHANGES='buildChangesDiv'
    BUILD_LOG='buildLog'
    PARAMETERS='buildParameters'
    ARTIFACTS='artifacts'
​
​
def link_to_this_build(tab=BuildViewTabs.BUILD_LOG):
    '''Generate link to the currently running TeamCity build
​
    TeamCity structures things thusly:
    - Project
      - Configuration
        - Build
        - Build
    '''
​
    teamcity_url = os.environ.get('TEAMCITY_URL')
    if teamcity_url is None:
        raise Exception('TEAMCITY_URL environment variable is not defined. TeamCity does not provide this by default; you MUST populate env.TEAMCITY_URL in your build configuration.')
    arguments = {
        'buildId': build_property(BuildProperty.TEAMCITY_BUILD_ID),
        'buildTypeId': build_property(BuildProperty.TEAMCITY_BUILDTYPE_ID),
        'tab': tab,
    }
    query_string = '&'.join([k+'='+str(v) for k,v in arguments.iteritems()])
    return teamcity_url + '/viewLog.html?' + query_string