buildtimetrend.travis.TravisData.set_build_matrix() - Code Metrics - Inspection of "increase build number" - buildtimetrend/python-lib - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( fec650...1a3e44 )

by Dieter

created 2016-01-24 13:25 UTC

set_build_matrix() C

↳ Parent: buildtimetrend.travis.TravisData

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	8
dl	0
loc	63
rs	5.6104

How to fix Long Method

# vim: set expandtab sw=4 ts=4:
"""
Functions and classes to retrieve and parse Travis CI build data.

Copyright (C) 2014-2016 Dieter Adriaenssens <[email protected]>

This file is part of buildtimetrend/python-lib
<https://github.com/buildtimetrend/python-lib/>

This program is free software: you can redistribute it and/or modify
it under the terms of the GNU Affero General Public License as published by
the Free Software Foundation, either version 3 of the License, or
any later version.

This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Affero General Public License for more details.

You should have received a copy of the GNU Affero General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
"""
from builtins import str
from builtins import object
import re
import json
from buildtimetrend import logger
from buildtimetrend.tools import check_file
from buildtimetrend.tools import check_dict
from buildtimetrend.tools import is_string
from buildtimetrend.buildjob import BuildJob
from buildtimetrend.collection import Collection
from buildtimetrend.travis.connector import TravisOrgConnector
from buildtimetrend.travis.connector import TravisConnector
from buildtimetrend.travis.substage import TravisSubstage
try:
    # For Python 3.0 and later
    from urllib.error import HTTPError, URLError
except ImportError:
    # Fall back to Python 2's urllib2
    from urllib2 import HTTPError, URLError


# strings to parse timestamps in Travis CI log file
TRAVIS_LOG_PARSE_TIMING_STRINGS = [
    r'travis_time:end:(?P<end_hash>.*):start=(?P<start_timestamp>\d+),'
    r'finish=(?P<finish_timestamp>\d+),duration=(?P<duration>\d+)\x0d\x1b',
    r'travis_fold:end:(?P<end_stage>\w+)\.(?P<end_substage>\d+)\x0d\x1b',
    r'travis_fold:start:(?P<start_stage>\w+)\.(?P<start_substage>\d+)\x0d\x1b',
    r'travis_time:start:(?P<start_hash>.*)\x0d\x1b\[0K',
    r'\$\ (?P<command>.*)\r',
]
TRAVIS_LOG_PARSE_WORKER_STRING = r'Using worker:\ (?P<hostname>.*):(?P<os>.*)'


class TravisData(object):

    """Gather data from Travis CI using the API."""

    def __init__(self, repo, build_id, connector=None):
        """
        Retrieve Travis CI build data using the API.

        Parameters:
        - repo : github repository slug (fe. buildtimetrend/python-lib)
        - build_id : Travis CI build id (fe. 158)
        - connector : Travis Connector instance
        """
        self.builds_data = {}
        self.build_jobs = {}
        self.current_build_data = {}
        self.current_job = BuildJob()
        self.travis_substage = None
        self.repo = repo
        self.build_id = str(build_id)
        # set TravisConnector if it is defined
        if isinstance(connector, TravisConnector):
            self.connector = connector
        # use Travis Org connector by default
        else:
            self.connector = TravisOrgConnector()

    def get_build_data(self):
        """
        Retrieve Travis CI build data.

        Returns true if retrieving data was succesful, false on error.
        """
        request = 'repos/{repo}/builds?number={build_id}'.format(
            repo=self.repo, build_id=self.build_id
        )
        try:
            self.builds_data = self.connector.json_request(request)
        except (HTTPError, URLError) as msg:
            logger.error("Error getting build data from Travis CI: %s", msg)
            return False

        # log builds_data
        logger.debug(
            "Build #%s data : %s",
            str(self.build_id),
            json.dumps(self.builds_data, sort_keys=True, indent=2)
        )

        return True

    def get_substage_name(self, command):
        """
        Resolve Travis CI substage name that corresponds to a cli command.

        Parameters:
        - command : cli command
        """
        if not is_string(command):
            return ""

        if len(self.current_build_data) > 0 and \
                "config" in self.current_build_data:
            build_config = self.current_build_data["config"]
        else:
            logger.warning(
                "Travis CI build config is not set"
            )
            return ""

        # check if build_config collection is empty
        if build_config:
            for stage_name, commands in build_config.items():
                if type(commands) is list and command in commands:
                    substage_number = commands.index(command) + 1
                    substage_name = "{stage}.{substage:d}".format(
                        stage=stage_name, substage=substage_number
                    )
                    logger.debug(
                        "Substage %s corresponds to '%s'",
                        substage_name, command
                    )
                    return substage_name

        return ""

    def process_build_jobs(self):
        """
        Retrieve Travis CI build job data.

        Method is a generator, iterate result to get each processed build job.
        """
        if len(self.builds_data) > 0 and "builds" in self.builds_data:
            for build in self.builds_data['builds']:
                self.current_build_data = build

                if "job_ids" in build:
                    for job_id in build['job_ids']:
                        yield self.process_build_job(job_id)

            # reset current_build_data after builds are processed
            self.current_build_data = {}

    def process_build_job(self, job_id):
        """
        Retrieve Travis CI build job data.

        Parameters:
        - job_id : ID of the job to process
        """
        if job_id is None:
            return None

        # retrieve job data from Travis CI
        job_data = self.get_job_data(job_id)
        # process build/job data
        self.process_job_data(job_data)
        # parse Travis CI job log file
        self.parse_job_log(job_id)

        # store build job
        self.build_jobs[str(job_id)] = self.current_job
        # create new build job instance
        self.current_job = BuildJob()

        # return processed build job
        return self.build_jobs[str(job_id)]

    def get_job_data(self, job_id):
        """
        Retrieve Travis CI job data.

        Parameters:
        - job_id : ID of the job to process
        """
        request = 'jobs/{:s}'.format(str(job_id))
        job_data = self.connector.json_request(request)

        # log job_data
        logger.debug(
            "Job #%s data : %s",
            str(job_id),
            json.dumps(job_data, sort_keys=True, indent=2)
        )

        return job_data

    def process_job_data(self, job_data):
        """
        Process Job/build data.

        Set build/job properties :
        - Build/job ID
        - build result : passed, failed, errored
        - git repo
        - git branch
        - CI platform : Travis
        - build matrix (language, language version, compiler, ...)
        - build_trigger : push, pull_request
        - pull_request (is_pull_request, title, number)

        Parameters:
        - job_data : dictionary with Travis CI job data
        """
        self.current_job.add_property(
            "build",
            # buildnumber is part before "." of job number
            job_data['job']['number'].split(".")[0]
        )
        self.current_job.add_property("job", job_data['job']['number'])
        self.current_job.add_property("branch", job_data['commit']['branch'])
        self.current_job.add_property(
            "repo",
            job_data['job']['repository_slug']
        )
        self.current_job.add_property("ci_platform", 'travis')
        self.current_job.add_property("result", job_data['job']['state'])

        self.set_build_matrix(job_data)

        self.process_pull_request_data()

        self.current_job.set_started_at(job_data['job']['started_at'])
        self.current_job.set_finished_at(job_data['job']['finished_at'])

        # calculate job duration from start and finished timestamps
        # if no timing tags are available
        if not self.has_timing_tags():
            self.current_job.add_property("duration", self.get_job_duration())

    def set_build_matrix(self, job_data):
        """
        Retrieve build matrix data from job data and store in properties.

        Properties :
        - language
        - language version (if applicable)
        - compiler (if applicable)
        - operating system
        - environment parameters

        Parameters:
        - job_data : dictionary with Travis CI job data
        """
        # check if job config data exists
        if 'job' not in job_data or 'config' not in job_data['job']:
            logger.warning("Job config data doesn't exist")
            return

        build_matrix = Collection()
        job_config = job_data['job']['config']

        if 'language' in job_config:
            language = job_config['language']
            build_matrix.add_item('language', language)

            # set language version
            # ('d', 'dart', 'go', 'perl', 'php', 'python', 'rust')
            if language in job_config:
                if language == 'android':
                    build_matrix.add_item(
                        "language_components",
                        " ".join(job_config[language]["components"])
                    )
                else:
                    build_matrix.add_item(
                        'language_version',
                        str(job_config[language])
                    )

        # language specific build matrix parameters
        parameters = {
            'ghc': 'ghc',  # Haskell
            'jdk': 'jdk',  # Java, Android, Groovy, Ruby, Scala
            'lein': 'lein',  # Clojure
            'mono': 'mono',  # C#, F#, Visual Basic
            'node_js': 'node_js',  # Javascript
            'otp_release': 'otp_release',  # Erlang
            'rvm': 'rvm',  # Ruby, Objective-C
            'gemfile': 'gemfile',  # Ruby, Objective-C
            'xcode_sdk': 'xcode_sdk',  # Objective-C
            'xcode_scheme': 'xcode_scheme',  # Objective-C
            'compiler': 'compiler',  # C, C++
            'os': 'os',
            'env': 'parameters'
        }
        for parameter, name in parameters.items():
            if parameter in job_config:
                build_matrix.add_item(name, str(job_config[parameter]))

        self.current_job.add_property(
            "build_matrix",
            build_matrix.get_items_with_summary()
        )

    def process_pull_request_data(self):
        """Retrieve pull request data from Travis CI API."""
        # check if collection is empty
        if self.current_build_data:
            if "event_type" in self.current_build_data:
                # build trigger (push or pull_request)
                self.current_job.add_property(
                    "build_trigger",
                    self.current_build_data["event_type"]
                )

            # pull_request
            pull_request_data = {}
            if "pull_request" in self.current_build_data:
                pull_request_data["is_pull_request"] = \
                    self.current_build_data["pull_request"]
            else:
                pull_request_data["is_pull_request"] = False

            if "pull_request_title" in self.current_build_data:
                pull_request_data["title"] = \
                    self.current_build_data["pull_request_title"]

            if "pull_request_number" in self.current_build_data:
                pull_request_data["number"] = \
                    self.current_build_data["pull_request_number"]

            self.current_job.add_property("pull_request", pull_request_data)

    def parse_job_log(self, job_id):
        """
        Parse Travis CI job log.

        Parameters:
        - job_id : ID of the job to process
        """
        self.parse_job_log_stream(self.connector.download_job_log(job_id))

    def parse_job_log_file(self, filename):
        """
        Open a Travis CI log file and parse it.

        Parameters :
        - filename : filename of Travis CI log
        Returns false if file doesn't exist, true if it was read successfully.
        """
        # load timestamps file
        if not check_file(filename):
            return False

        # read timestamps, calculate stage duration
        with open(filename, 'rb') as file_stream:
            self.parse_job_log_stream(file_stream)

        return True

    def parse_job_log_stream(self, stream):
        """
        Parse Travis CI job log stream.

        Parameters:
        - stream : stream of job log file
        """
        self.travis_substage = TravisSubstage()
        check_timing_tags = self.has_timing_tags()

        for line in stream:
            # convert to str if line is bytes type
            if isinstance(line, bytes):
                line = line.decode('utf-8')
            # parse Travis CI timing tags
            if check_timing_tags and 'travis_' in line:
                self.parse_travis_time_tag(line)
            # parse Travis CI worker tag
            if 'Using worker:' in line:
                self.parse_travis_worker_tag(line)

    def parse_travis_time_tag(self, line):
        """
        Parse and process Travis CI timing tags.

        Parameters:
        - line : line from logfile containing Travis CI tags
        """
        if self.travis_substage is None:
            self.travis_substage = TravisSubstage()

        escaped_line = line.replace('\x0d', '*').replace('\x1b', 'ESC')
        logger.debug('line : %s', escaped_line)

        # parse Travis CI timing tags
        for parse_string in TRAVIS_LOG_PARSE_TIMING_STRINGS:
            result = re.search(parse_string, line)
            if result:
                self.travis_substage.process_parsed_tags(result.groupdict())

                # when finished : log stage and create a new instance
                if self.travis_substage.has_finished():
                    # set substage name, if it is not set
                    if not self.travis_substage.has_name() and \
                            self.travis_substage.has_command():
                        self.travis_substage.set_name(
                            self.get_substage_name(
                                self.travis_substage.get_command()
                            )
                        )

                    # only log complete substages
                    if not self.travis_substage.finished_incomplete:
                        self.current_job.add_stage(self.travis_substage.stage)
                    self.travis_substage = TravisSubstage()

    def parse_travis_worker_tag(self, line):
        """
        Parse and process Travis CI worker tag.

        Parameters:
        - line : line from logfile containing Travis CI tags
        """
        logger.debug('line : %s', line)

        # parse Travis CI worker tags
        result = re.search(TRAVIS_LOG_PARSE_WORKER_STRING, line)
        if not result:
            return

        worker_tags = result.groupdict()

        # check if parameter worker_tags is a dictionary and
        # if it contains all required tags
        tag_list = list({'hostname', 'os'})
        if check_dict(worker_tags, "worker_tags", tag_list):
            logger.debug("Worker tags : %s", worker_tags)
            self.current_job.add_property("worker", worker_tags)

    def has_timing_tags(self):
        """
        Check if Travis CI job log has timing tags.

        Timing tags were introduced on Travis CI starting 2014-08-07,
        check if started_at is more recent.
        """
        started_at = self.current_job.get_property("started_at")
        if started_at is None or "timestamp_seconds" not in started_at:
            return False

        # 1407369600 is epoch timestamp of 2014-08-07T00:00:00Z
        return started_at["timestamp_seconds"] > 1407369600

    def get_job_duration(self):
        """Calculate build job duration."""
        started_at = self.current_job.get_property("started_at")
        finished_at = self.current_job.get_property("finished_at")
        if started_at is None or "timestamp_seconds" not in started_at or \
                finished_at is None or "timestamp_seconds" not in finished_at:
            return 0.0

        timestamp_start = float(started_at["timestamp_seconds"])
        timestamp_end = float(finished_at["timestamp_seconds"])
        return timestamp_end - timestamp_start

    def get_started_at(self):
        """Retrieve timestamp when build was started."""
        if check_dict(self.current_build_data, key_list=["started_at"]):
            return self.current_build_data['started_at']
        else:
            return None

    def get_finished_at(self):
        """Retrieve timestamp when build finished."""
        if check_dict(self.current_build_data, key_list=["finished_at"]):
            return self.current_build_data['finished_at']
        else:
            return None


1			# vim: set expandtab sw=4 ts=4:
2			"""
3			Functions and classes to retrieve and parse Travis CI build data.
4
5			Copyright (C) 2014-2016 Dieter Adriaenssens <[email protected]>
6
7			This file is part of buildtimetrend/python-lib
8			<https://github.com/buildtimetrend/python-lib/>
9
10			This program is free software: you can redistribute it and/or modify
11			it under the terms of the GNU Affero General Public License as published by
12			the Free Software Foundation, either version 3 of the License, or
13			any later version.
14
15			This program is distributed in the hope that it will be useful,
16			but WITHOUT ANY WARRANTY; without even the implied warranty of
17			MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
18			GNU Affero General Public License for more details.
19
20			You should have received a copy of the GNU Affero General Public License
21			along with this program. If not, see <http://www.gnu.org/licenses/>.
22			"""
23			from builtins import str
24			from builtins import object
25			import re
26			import json
27			from buildtimetrend import logger
28			from buildtimetrend.tools import check_file
29			from buildtimetrend.tools import check_dict
30			from buildtimetrend.tools import is_string
31			from buildtimetrend.buildjob import BuildJob
32			from buildtimetrend.collection import Collection
33			from buildtimetrend.travis.connector import TravisOrgConnector
34			from buildtimetrend.travis.connector import TravisConnector
35			from buildtimetrend.travis.substage import TravisSubstage
36			try:
37			# For Python 3.0 and later
38			from urllib.error import HTTPError, URLError
39			except ImportError:
40			# Fall back to Python 2's urllib2
41			from urllib2 import HTTPError, URLError
42
43
44			# strings to parse timestamps in Travis CI log file
45			TRAVIS_LOG_PARSE_TIMING_STRINGS = [
46			r'travis_time:end:(?P<end_hash>.*):start=(?P<start_timestamp>\d+),'
47			r'finish=(?P<finish_timestamp>\d+),duration=(?P<duration>\d+)\x0d\x1b',
48			r'travis_fold:end:(?P<end_stage>\w+)\.(?P<end_substage>\d+)\x0d\x1b',
49			r'travis_fold:start:(?P<start_stage>\w+)\.(?P<start_substage>\d+)\x0d\x1b',
50			r'travis_time:start:(?P<start_hash>.*)\x0d\x1b\[0K',
51			r'\$\ (?P<command>.*)\r',
52			]
53			TRAVIS_LOG_PARSE_WORKER_STRING = r'Using worker:\ (?P<hostname>.):(?P<os>.)'
54
55
56			class TravisData(object):
57
58			"""Gather data from Travis CI using the API."""
59
60			def __init__(self, repo, build_id, connector=None):
61			"""
62			Retrieve Travis CI build data using the API.
63
64			Parameters:
65			- repo : github repository slug (fe. buildtimetrend/python-lib)
66			- build_id : Travis CI build id (fe. 158)
67			- connector : Travis Connector instance
68			"""
69			self.builds_data = {}
70			self.build_jobs = {}
71			self.current_build_data = {}
72			self.current_job = BuildJob()
73			self.travis_substage = None
74			self.repo = repo
75			self.build_id = str(build_id)
76			# set TravisConnector if it is defined
77			if isinstance(connector, TravisConnector):
78			self.connector = connector
79			# use Travis Org connector by default
80			else:
81			self.connector = TravisOrgConnector()
82
83			def get_build_data(self):
84			"""
85			Retrieve Travis CI build data.
86
87			Returns true if retrieving data was succesful, false on error.
88			"""
89			request = 'repos/{repo}/builds?number={build_id}'.format(
90			repo=self.repo, build_id=self.build_id
91			)
92			try:
93			self.builds_data = self.connector.json_request(request)
94			except (HTTPError, URLError) as msg:
95			logger.error("Error getting build data from Travis CI: %s", msg)
96			return False
97
98			# log builds_data
99			logger.debug(
100			"Build #%s data : %s",
101			str(self.build_id),
102			json.dumps(self.builds_data, sort_keys=True, indent=2)
103			)
104
105			return True
106
107			def get_substage_name(self, command):
108			"""
109			Resolve Travis CI substage name that corresponds to a cli command.
110
111			Parameters:
112			- command : cli command
113			"""
114			if not is_string(command):
115			return ""
116
117			if len(self.current_build_data) > 0 and \
118			"config" in self.current_build_data:
119			build_config = self.current_build_data["config"]
120			else:
121			logger.warning(
122			"Travis CI build config is not set"
123			)
124			return ""
125
126			# check if build_config collection is empty
127			if build_config:
128			for stage_name, commands in build_config.items():
129			if type(commands) is list and command in commands:
130			substage_number = commands.index(command) + 1
131			substage_name = "{stage}.{substage:d}".format(
132			stage=stage_name, substage=substage_number
133			)
134			logger.debug(
135			"Substage %s corresponds to '%s'",
136			substage_name, command
137			)
138			return substage_name
139
140			return ""
141
142			def process_build_jobs(self):
143			"""
144			Retrieve Travis CI build job data.
145
146			Method is a generator, iterate result to get each processed build job.
147			"""
148			if len(self.builds_data) > 0 and "builds" in self.builds_data:
149			for build in self.builds_data['builds']:
150			self.current_build_data = build
151
152			if "job_ids" in build:
153			for job_id in build['job_ids']:
154			yield self.process_build_job(job_id)
155
156			# reset current_build_data after builds are processed
157			self.current_build_data = {}
158
159			def process_build_job(self, job_id):
160			"""
161			Retrieve Travis CI build job data.
162
163			Parameters:
164			- job_id : ID of the job to process
165			"""
166			if job_id is None:
167			return None
168
169			# retrieve job data from Travis CI
170			job_data = self.get_job_data(job_id)
171			# process build/job data
172			self.process_job_data(job_data)
173			# parse Travis CI job log file
174			self.parse_job_log(job_id)
175
176			# store build job
177			self.build_jobs[str(job_id)] = self.current_job
178			# create new build job instance
179			self.current_job = BuildJob()
180
181			# return processed build job
182			return self.build_jobs[str(job_id)]
183
184			def get_job_data(self, job_id):
185			"""
186			Retrieve Travis CI job data.
187
188			Parameters:
189			- job_id : ID of the job to process
190			"""
191			request = 'jobs/{:s}'.format(str(job_id))
192			job_data = self.connector.json_request(request)
193
194			# log job_data
195			logger.debug(
196			"Job #%s data : %s",
197			str(job_id),
198			json.dumps(job_data, sort_keys=True, indent=2)
199			)
200
201			return job_data
202
203			def process_job_data(self, job_data):
204			"""
205			Process Job/build data.
206
207			Set build/job properties :
208			- Build/job ID
209			- build result : passed, failed, errored
210			- git repo
211			- git branch
212			- CI platform : Travis
213			- build matrix (language, language version, compiler, ...)
214			- build_trigger : push, pull_request
215			- pull_request (is_pull_request, title, number)
216
217			Parameters:
218			- job_data : dictionary with Travis CI job data
219			"""
220			self.current_job.add_property(
221			"build",
222			# buildnumber is part before "." of job number
223			job_data['job']['number'].split(".")[0]
224			)
225			self.current_job.add_property("job", job_data['job']['number'])
226			self.current_job.add_property("branch", job_data['commit']['branch'])
227			self.current_job.add_property(
228			"repo",
229			job_data['job']['repository_slug']
230			)
231			self.current_job.add_property("ci_platform", 'travis')
232			self.current_job.add_property("result", job_data['job']['state'])
233
234			self.set_build_matrix(job_data)
235
236			self.process_pull_request_data()
237
238			self.current_job.set_started_at(job_data['job']['started_at'])
239			self.current_job.set_finished_at(job_data['job']['finished_at'])
240
241			# calculate job duration from start and finished timestamps
242			# if no timing tags are available
243			if not self.has_timing_tags():
244			self.current_job.add_property("duration", self.get_job_duration())
245
246			def set_build_matrix(self, job_data):
247			"""
248			Retrieve build matrix data from job data and store in properties.
249
250			Properties :
251			- language
252			- language version (if applicable)
253			- compiler (if applicable)
254			- operating system
255			- environment parameters
256
257			Parameters:
258			- job_data : dictionary with Travis CI job data
259			"""
260			# check if job config data exists
261			if 'job' not in job_data or 'config' not in job_data['job']:
262			logger.warning("Job config data doesn't exist")
263			return
264
265			build_matrix = Collection()
266			job_config = job_data['job']['config']
267
268			if 'language' in job_config:
269			language = job_config['language']
270			build_matrix.add_item('language', language)
271
272			# set language version
273			# ('d', 'dart', 'go', 'perl', 'php', 'python', 'rust')
274			if language in job_config:
275			if language == 'android':
276			build_matrix.add_item(
277			"language_components",
278			" ".join(job_config[language]["components"])
279			)
280			else:
281			build_matrix.add_item(
282			'language_version',
283			str(job_config[language])
284			)
285
286			# language specific build matrix parameters
287			parameters = {
288			'ghc': 'ghc', # Haskell
289			'jdk': 'jdk', # Java, Android, Groovy, Ruby, Scala
290			'lein': 'lein', # Clojure
291			'mono': 'mono', # C#, F#, Visual Basic
292			'node_js': 'node_js', # Javascript
293			'otp_release': 'otp_release', # Erlang
294			'rvm': 'rvm', # Ruby, Objective-C
295			'gemfile': 'gemfile', # Ruby, Objective-C
296			'xcode_sdk': 'xcode_sdk', # Objective-C
297			'xcode_scheme': 'xcode_scheme', # Objective-C
298			'compiler': 'compiler', # C, C++
299			'os': 'os',
300			'env': 'parameters'
301			}
302			for parameter, name in parameters.items():
303			if parameter in job_config:
304			build_matrix.add_item(name, str(job_config[parameter]))
305
306			self.current_job.add_property(
307			"build_matrix",
308			build_matrix.get_items_with_summary()
309			)
310
311			def process_pull_request_data(self):
312			"""Retrieve pull request data from Travis CI API."""
313			# check if collection is empty
314			if self.current_build_data:
315			if "event_type" in self.current_build_data:
316			# build trigger (push or pull_request)
317			self.current_job.add_property(
318			"build_trigger",
319			self.current_build_data["event_type"]
320			)
321
322			# pull_request
323			pull_request_data = {}
324			if "pull_request" in self.current_build_data:
325			pull_request_data["is_pull_request"] = \
326			self.current_build_data["pull_request"]
327			else:
328			pull_request_data["is_pull_request"] = False
329
330			if "pull_request_title" in self.current_build_data:
331			pull_request_data["title"] = \
332			self.current_build_data["pull_request_title"]
333
334			if "pull_request_number" in self.current_build_data:
335			pull_request_data["number"] = \
336			self.current_build_data["pull_request_number"]
337
338			self.current_job.add_property("pull_request", pull_request_data)
339
340			def parse_job_log(self, job_id):
341			"""
342			Parse Travis CI job log.
343
344			Parameters:
345			- job_id : ID of the job to process
346			"""
347			self.parse_job_log_stream(self.connector.download_job_log(job_id))
348
349			def parse_job_log_file(self, filename):
350			"""
351			Open a Travis CI log file and parse it.
352
353			Parameters :
354			- filename : filename of Travis CI log
355			Returns false if file doesn't exist, true if it was read successfully.
356			"""
357			# load timestamps file
358			if not check_file(filename):
359			return False
360
361			# read timestamps, calculate stage duration
362			with open(filename, 'rb') as file_stream:
363			self.parse_job_log_stream(file_stream)
364
365			return True
366
367			def parse_job_log_stream(self, stream):
368			"""
369			Parse Travis CI job log stream.
370
371			Parameters:
372			- stream : stream of job log file
373			"""
374			self.travis_substage = TravisSubstage()
375			check_timing_tags = self.has_timing_tags()
376
377			for line in stream:
378			# convert to str if line is bytes type
379			if isinstance(line, bytes):
380			line = line.decode('utf-8')
381			# parse Travis CI timing tags
382			if check_timing_tags and 'travis_' in line:
383			self.parse_travis_time_tag(line)
384			# parse Travis CI worker tag
385			if 'Using worker:' in line:
386			self.parse_travis_worker_tag(line)
387
388			def parse_travis_time_tag(self, line):
389			"""
390			Parse and process Travis CI timing tags.
391
392			Parameters:
393			- line : line from logfile containing Travis CI tags
394			"""
395			if self.travis_substage is None:
396			self.travis_substage = TravisSubstage()
397
398			escaped_line = line.replace('\x0d', '*').replace('\x1b', 'ESC')
399			logger.debug('line : %s', escaped_line)
400
401			# parse Travis CI timing tags
402			for parse_string in TRAVIS_LOG_PARSE_TIMING_STRINGS:
403			result = re.search(parse_string, line)
404			if result:
405			self.travis_substage.process_parsed_tags(result.groupdict())
406
407			# when finished : log stage and create a new instance
408			if self.travis_substage.has_finished():
409			# set substage name, if it is not set
410			if not self.travis_substage.has_name() and \
411			self.travis_substage.has_command():
412			self.travis_substage.set_name(
413			self.get_substage_name(
414			self.travis_substage.get_command()
415			)
416			)
417
418			# only log complete substages
419			if not self.travis_substage.finished_incomplete:
420			self.current_job.add_stage(self.travis_substage.stage)
421			self.travis_substage = TravisSubstage()
422
423			def parse_travis_worker_tag(self, line):
424			"""
425			Parse and process Travis CI worker tag.
426
427			Parameters:
428			- line : line from logfile containing Travis CI tags
429			"""
430			logger.debug('line : %s', line)
431
432			# parse Travis CI worker tags
433			result = re.search(TRAVIS_LOG_PARSE_WORKER_STRING, line)
434			if not result:
435			return
436
437			worker_tags = result.groupdict()
438
439			# check if parameter worker_tags is a dictionary and
440			# if it contains all required tags
441			tag_list = list({'hostname', 'os'})
442			if check_dict(worker_tags, "worker_tags", tag_list):
443			logger.debug("Worker tags : %s", worker_tags)
444			self.current_job.add_property("worker", worker_tags)
445
446			def has_timing_tags(self):
447			"""
448			Check if Travis CI job log has timing tags.
449
450			Timing tags were introduced on Travis CI starting 2014-08-07,
451			check if started_at is more recent.
452			"""
453			started_at = self.current_job.get_property("started_at")
454			if started_at is None or "timestamp_seconds" not in started_at:
455			return False
456
457			# 1407369600 is epoch timestamp of 2014-08-07T00:00:00Z
458			return started_at["timestamp_seconds"] > 1407369600
459
460			def get_job_duration(self):
461			"""Calculate build job duration."""
462			started_at = self.current_job.get_property("started_at")
463			finished_at = self.current_job.get_property("finished_at")
464			if started_at is None or "timestamp_seconds" not in started_at or \
465			finished_at is None or "timestamp_seconds" not in finished_at:
466			return 0.0
467
468			timestamp_start = float(started_at["timestamp_seconds"])
469			timestamp_end = float(finished_at["timestamp_seconds"])
470			return timestamp_end - timestamp_start
471
472			def get_started_at(self):
473			"""Retrieve timestamp when build was started."""
474			if check_dict(self.current_build_data, key_list=["started_at"]):
475			return self.current_build_data['started_at']
476			else:
477			return None
478
479			def get_finished_at(self):
480			"""Retrieve timestamp when build finished."""
481			if check_dict(self.current_build_data, key_list=["finished_at"]):
482			return self.current_build_data['finished_at']
483			else:
484			return None
485

buildtimetrend / python-lib

Push — master ( fec650...1a3e44 )

set_build_matrix() C

Complexity

Size

Duplication

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like