buildtimetrend.get_latest_buildtime() - Code Metrics - Inspection of "reset KeenClient before each test" - buildtimetrend/python-lib - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 232dbd...dd6f0d )

by Dieter

created 2015-12-14 20:18 UTC

buildtimetrend.get_latest_buildtime() F

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	9
dl	0
loc	30
rs	3

# vim: set expandtab sw=4 ts=4:
"""
Interface to Keen IO.

Copyright (C) 2014-2015 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 __future__ import division
from builtins import str
import os
from buildtimetrend import logger
import copy
import keen
import requests
from keen import scoped_keys
from buildtimetrend.settings import Settings
from buildtimetrend.tools import check_file
from buildtimetrend.tools import check_dict
from buildtimetrend.tools import is_list
from buildtimetrend.tools import is_string


TIME_INTERVALS = {
    'week': {'name': 'week', 'timeframe': 'this_7_days', 'max_age': 600},
    'month': {'name': 'month', 'timeframe': 'this_30_days', 'max_age': 600},
    'year': {'name': 'year', 'timeframe': 'this_52_weeks', 'max_age': 1800}
}
KEEN_PROJECT_INFO_NAME = "buildtime_trend"


def keen_has_project_id():
    """Check if Keen.io project ID is set."""
    if "KEEN_PROJECT_ID" in os.environ or keen.project_id is not None:
        return True

    logger.warning("Keen.io Project ID is not set")
    return False


def keen_has_master_key():
    """Check if Keen.io Master API key is set."""
    if "KEEN_MASTER_KEY" in os.environ or keen.master_key is not None:
        return True

    logger.warning("Keen.io Master API Key is not set")
    return False


def keen_has_write_key():
    """Check if Keen.io Write Key is set."""
    if "KEEN_WRITE_KEY" in os.environ or keen.write_key is not None:
        return True

    logger.warning("Keen.io Write Key is not set")
    return False


def keen_has_read_key():
    """Check if Keen.io Read key is set."""
    if "KEEN_READ_KEY" in os.environ or keen.read_key is not None:
        return True

    logger.warning("Keen.io Read Key is not set")
    return False


def keen_is_writable():
    """Check if login keys for Keen IO API are set, to allow writing."""
    if keen_has_project_id() and keen_has_write_key():
        return True

    logger.warning("Keen.io Write Key is not set")
    return False


def keen_is_readable():
    """Check if login keys for Keen IO API are set, to allow reading."""
    if keen_has_project_id() and keen_has_read_key():
        return True

    logger.warning("Keen.io Read Key is not set")
    return False


def keen_io_generate_read_key(repo):
    """
    Create scoped key for reading only the build-stages related data.

    Param repo : github repository slug (fe. buildtimetrend/python-lib)
    """
    if not keen_has_master_key():
        logger.warning("Keen.io Read Key was not created,"
                       " keen.master_key is not defined.")
        return None

    master_key = keen.master_key or os.environ.get("KEEN_MASTER_KEY")

    privileges = {
        "allowed_operations": ["read"]
    }

    if repo is not None:
        privileges["filters"] = [get_repo_filter(repo)]

    logger.info("Keen.io Read Key is created for %s", repo)
    return scoped_keys.encrypt(master_key, privileges)


def keen_io_generate_write_key():
    """Create scoped key for write access to Keen.io database."""
    if not keen_has_master_key():
        logger.warning("Keen.io Write Key was not created,"
                       " keen.master_key is not defined.")
        return None

    master_key = keen.master_key or os.environ.get("KEEN_MASTER_KEY")

    privileges = {
        "allowed_operations": ["write"]
    }

    logger.info("Keen.io Write Key is created")
    return scoped_keys.encrypt(master_key, privileges)


def send_build_data(build):
    """Send build data generated by client to keen.io."""
    if keen_is_writable():
        logger.info("Sending client build data to Keen.io")
        keen_add_event("build_jobs", {"job": build.to_dict()})
        keen_add_events("build_stages", build.stages_to_list())


def send_build_data_service(build):
    """Send build data generated by service to keen.io."""
    if keen_is_writable():
        logger.info("Sending service build data to Keen.io")
        keen_add_event("build_jobs", {"job": build.to_dict()})
        keen_add_events("build_substages", build.stages_to_list())


def keen_add_event(event_collection, payload):
    """
    Wrapper for keen.add_event(), adds project info.

    Param event_collection : collection event data is submitted to
    Param payload : data that is submitted
    """
    # add project info to this event
    payload = add_project_info_dict(payload)

    # submit list of events to Keen.io
    keen.add_event(event_collection, payload)


def keen_add_events(event_collection, payload):
    """
    Wrapper for keen.add_events(), adds project info to each event.

    Param event_collection : collection event data is submitted to
    Param payload : array of events that is submitted
    """
    # add project info to each event
    payload = add_project_info_list(payload)

    # submit list of events to Keen.io
    keen.add_events({event_collection: payload})


def add_project_info_dict(payload):
    """
    Add project info to a dictonary.

    Param payload: dictonary payload
    """
    # check if payload is a dictionary, throws an exception if it isn't
    check_dict(payload, "payload")

    payload_as_dict = copy.deepcopy(payload)

    payload_as_dict[KEEN_PROJECT_INFO_NAME] = Settings().get_project_info()

    if "job" in payload:
        # override project_name, set to build_job repo
        if "repo" in payload["job"]:
            payload_as_dict[KEEN_PROJECT_INFO_NAME]["project_name"] = \
                payload["job"]["repo"]

        # override timestamp, set to finished_at timestamp
        if "finished_at" in payload["job"]:
            payload_as_dict["keen"] = {
                "timestamp": payload["job"]["finished_at"]["isotimestamp"]
            }

    return payload_as_dict


def add_project_info_list(payload):
    """
    Add project info to a list of dictionaries.

    Param payload: list of dictionaries
    """
    # check if payload is a list, throws an exception if it isn't
    is_list(payload, "payload")

    payload_as_list = []

    # loop over dicts in payload and add project info to each one
    for event_dict in payload:
        payload_as_list.append(add_project_info_dict(event_dict))

    return payload_as_list


def get_dashboard_keen_config(repo):
    """
    Generate the Keen.io settings for the configuration of the dashboard.

    The dashboard is Javascript powered HTML file that contains the
    graphs generated by Keen.io.
    """
    # initialise config settings
    keen_config = {}

    if not keen_has_project_id() or not keen_has_master_key():
        logger.warning("Keen.io related config settings could not be created,"
                       " keen.project_id and/or keen.master_key"
                       " are not defined.")
        return keen_config

    # set keen project ID
    if keen.project_id is None:
        keen_project_id = os.environ["KEEN_PROJECT_ID"]
    else:
        keen_project_id = keen.project_id
    keen_config['projectId'] = str(keen_project_id)

    # generate read key
    read_key = keen_io_generate_read_key(repo)
    if read_key is not None:
        # convert bytes to string
        if isinstance(read_key, bytes):
            read_key = read_key.decode('utf-8')

        keen_config['readKey'] = str(read_key)

    # return keen config settings
    return keen_config


def get_dashboard_config_dict(repo, extra=None):
    """
    Generate the configuration settings for the dashboard.

    The dashboard is Javascript powered HTML file that contains the
    graphs generated by Keen.io.

    Parameters:
    - repo : repo name (fe. buildtimetrend/service)
    - extra : dictionary of extra config settings, format : {"name" : "value"}
    """
    # initialise config settings dictionaries
    config = {}

    # add repo and project name
    if repo is not None and not repo == "":
        # merge extra settings into existing config dictionary
        config.update({
            'projectName': str(repo),
            'repoName': str(repo)
        })

    # add extra config parameters
    if extra is not None and check_dict(extra, "extra"):
        config.update(extra)

    return config


def get_dashboard_config(repo, extra=None):
    """
    Generate the configuration settings for the dashboard.

    The dashboard is Javascript powered HTML file that contains the
    graphs generated by Keen.io.

    Parameters:
    - repo : repo name (fe. buildtimetrend/service)
    - extra : dictionary of extra config settings, format : {"name" : "value"}
    """
    # initialise config settings dictionaries
    config = get_dashboard_config_dict(repo, extra)
    keen_config = get_dashboard_keen_config(repo)

    # create configuration as a string
    return "var config = {};\nvar keenConfig = {};".format(config, keen_config)


def generate_dashboard_config_file(repo):
    """
    Generate the configuration file for the dashboard.

    The dashboard is Javascript powered HTML file that contains the
    graphs generated by Keen.io.

    Parameters:
    - repo : repo name (fe. buildtimetrend/service)
    """
    # get config settings
    config_string = get_dashboard_config(repo)

    # write config file
    config_file = Settings().get_setting("dashboard_configfile")
    with open(config_file, 'w') as outfile:
        outfile.write(config_string)

    if check_file(config_file):
        logger.info("Created trends dashboard config file %s", config_file)
        return True
    else:
        logger.warning("The dashboard config file was not created")
        return False


def get_avg_buildtime(repo=None, interval=None):

    """
    Query Keen.io database and retrieve average build time.

    Parameters :
    - repo : repo name (fe. buildtimetrend/service)
    - interval : timeframe, possible values : 'week', 'month', 'year',
                 anything else defaults to 'week'
    """
    if repo is None or not keen_is_readable():
        return -1

    interval_data = check_time_interval(interval)

    try:
        return keen.average(
            "build_jobs",
            target_property="job.duration",
            timeframe=interval_data['timeframe'],
            max_age=interval_data['max_age'],
            filters=[get_repo_filter(repo)]
        )
    except requests.ConnectionError:
        logger.error("Connection to Keen.io API failed")
        return -1
    except keen.exceptions.KeenApiError as msg:
        logger.error("Error in keenio.get_avg_buildtime() : " + str(msg))
        return -1


def get_total_build_jobs(repo=None, interval=None):

    """
    Query Keen.io database and retrieve total number of build jobs.

    Parameters :
    - repo : repo name (fe. buildtimetrend/service)
    - interval : timeframe, possible values : 'week', 'month', 'year',
                 anything else defaults to 'week'
    """
    if repo is None or not keen_is_readable():
        return -1

    interval_data = check_time_interval(interval)

    try:
        return keen.count_unique(
            "build_jobs",
            target_property="job.job",
            timeframe=interval_data['timeframe'],
            max_age=interval_data['max_age'],
            filters=[get_repo_filter(repo)]
        )
    except requests.ConnectionError:
        logger.error("Connection to Keen.io API failed")
        return -1
    except keen.exceptions.KeenApiError as msg:
        logger.error("Error in keenio.get_total_build_jobs() : " + str(msg))
        return -1


def get_passed_build_jobs(repo=None, interval=None):
    """
    Query Keen.io database and retrieve total number of build jobs that passed.

    Parameters :
    - repo : repo name (fe. buildtimetrend/service)
    - interval : timeframe, possible values : 'week', 'month', 'year',
                 anything else defaults to 'week'
    """
    if repo is None or not keen_is_readable():
        return -1

    interval_data = check_time_interval(interval)

    try:
        return keen.count_unique(
            "build_jobs",
            target_property="job.job",
            timeframe=interval_data['timeframe'],
            max_age=interval_data['max_age'],
            filters=[
                get_repo_filter(repo),
                {
                    "property_name": "job.result",
                    "operator": "eq",
                    "property_value": "passed"
                }
            ]
        )
    except requests.ConnectionError:
        logger.error("Connection to Keen.io API failed")
        return -1
    except keen.exceptions.KeenApiError as msg:
        logger.error("Error in keenio.get_passed_build_jobs() : " + str(msg))
        return -1


def get_pct_passed_build_jobs(repo=None, interval=None):
    """
    Calculate percentage of passed build jobs.

    Parameters :
    - repo : repo name (fe. buildtimetrend/service)
    - interval : timeframe, possible values : 'week', 'month', 'year',
                 anything else defaults to 'week'
    """
    total_jobs = get_total_build_jobs(repo, interval)
    passed_jobs = get_passed_build_jobs(repo, interval)

    logger.debug("passed/total build jobs : %d/%d", passed_jobs, total_jobs)

    # calculate percentage if at least one job was executed
    # passed is a valid number (not -1)
    if total_jobs > 0 and passed_jobs >= 0:
        return int(float(passed_jobs) / float(total_jobs) * 100.0)

    return -1


def get_result_color(value=0, ok_thershold=90, warning_thershold=70):
    """
    Get color code that corresponds to the result.

    Parameters:
    - value : value to check
    - ok_thershold : OK threshold value
    - warning_thershold : warning thershold value
    """
    if not(type(value) in (int, float) and
           type(ok_thershold) in (int, float) and
           type(warning_thershold) in (int, float)):
        return "lightgrey"

    # check thresholds
    if value >= ok_thershold:
        return "green"
    elif value >= warning_thershold:
        return "yellow"
    else:
        return "red"


def get_total_builds(repo=None, interval=None):

    """
    Query Keen.io database and retrieve total number of builds.

    Parameters :
    - repo : repo name (fe. buildtimetrend/service)
    - interval : timeframe, possible values : 'week', 'month', 'year',
                 anything else defaults to 'week'
    """
    if repo is None or not keen_is_readable():
        return -1

    interval_data = check_time_interval(interval)

    try:
        return keen.count_unique(
            "build_jobs",
            target_property="job.build",
            timeframe=interval_data['timeframe'],
            max_age=interval_data['max_age'],
            filters=[get_repo_filter(repo)]
        )
    except requests.ConnectionError:
        logger.error("Connection to Keen.io API failed")
        return -1
    except keen.exceptions.KeenApiError as msg:
        logger.error("Error in keenio.get_total_builds() : " + str(msg))
        return -1


def get_latest_buildtime(repo=None):
    """
    Query Keen.io database and retrieve buildtime duration of last build.

    Parameters :
    - repo : repo name (fe. buildtimetrend/python-lib)
    """
    if repo is None or not keen_is_readable():
        return -1

    try:
        result = keen.extraction(
            "build_jobs",
            property_names="job.duration",
            latest=1,
            filters=[get_repo_filter(repo)]
        )
    except requests.ConnectionError:
        logger.error("Connection to Keen.io API failed")
        return -1
    except keen.exceptions.KeenApiError as msg:
        logger.error("Error in keenio.get_latest_buildtime() : " + str(msg))
        return -1

    if is_list(result) and len(result) > 0 and \
            check_dict(result[0], None, ['job']) and \
            check_dict(result[0]['job'], None, ['duration']):
        return result[0]['job']['duration']

    return -1


def has_build_id(repo=None, build_id=None):
    """
    Check if build_id exists in Keen.io database.

    Parameters :
    - repo : repo name (fe. buildtimetrend/python-lib)
    - build_id : ID of the build
    """
    if repo is None or build_id is None:
        logger.error("Repo or build_id is not set")
        raise ValueError("Repo or build_id is not set")
    if not keen_is_readable():
        raise SystemError("Keen.io Project ID or API Read Key is not set")

    try:
        count = keen.count(
            "build_jobs",
            filters=[get_repo_filter(repo), {
                "property_name": "job.build",
                "operator": "eq",
                "property_value": str(build_id)}]
        )
    except requests.ConnectionError:
        logger.error("Connection to Keen.io API failed")
        raise SystemError("Connection to Keen.io API failed")
    except keen.exceptions.KeenApiError as msg:
        logger.error("Error in keenio.has_build_id : " + str(msg))
        raise SystemError(msg)

    return count > 0


def get_all_projects():
    """Query Keen.io database and retrieve a list of all projects."""
    if not keen_is_readable():
        return []

    try:
        result = keen.select_unique(
            "build_jobs",
            "buildtime_trend.project_name",
            max_age=3600 * 24  # cache for 24 hours
        )
    except requests.ConnectionError:
        logger.error("Connection to Keen.io API failed")
        return []
    except keen.exceptions.KeenApiError as msg:
        logger.error("Error in keenio.get_all_projects() : " + str(msg))
        return []

    if type(result) is list:
        return result

    return []


def get_repo_filter(repo=None):
    """
    Return filter for analysis request.

    Parameters
    - repo : repo slug name, fe. buildtimetrend/python-lib
    """
    if repo is None:
        return None

    return {
        "property_name": "%s.project_name" % KEEN_PROJECT_INFO_NAME,
        "operator": "eq",
        "property_value": str(repo)
    }


def check_time_interval(interval=None):
    """
    Check time interval and returns corresponding parameters.

    Parameters :
    - interval : timeframe, possible values : 'week', 'month', 'year',
                 anything else defaults to 'week'
    """
    if is_string(interval):
        # convert to lowercase
        interval = interval.lower()

        if interval in TIME_INTERVALS:
            return TIME_INTERVALS[interval]

    return TIME_INTERVALS['week']


1			# vim: set expandtab sw=4 ts=4:
2			"""
3			Interface to Keen IO.
4
5			Copyright (C) 2014-2015 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
24			from __future__ import division
25			from builtins import str
26			import os
27			from buildtimetrend import logger
28			import copy
29			import keen
30			import requests
31			from keen import scoped_keys
32			from buildtimetrend.settings import Settings
33			from buildtimetrend.tools import check_file
34			from buildtimetrend.tools import check_dict
35			from buildtimetrend.tools import is_list
36			from buildtimetrend.tools import is_string
37
38
39			TIME_INTERVALS = {
40			'week': {'name': 'week', 'timeframe': 'this_7_days', 'max_age': 600},
41			'month': {'name': 'month', 'timeframe': 'this_30_days', 'max_age': 600},
42			'year': {'name': 'year', 'timeframe': 'this_52_weeks', 'max_age': 1800}
43			}
44			KEEN_PROJECT_INFO_NAME = "buildtime_trend"
45
46
47			def keen_has_project_id():
48			"""Check if Keen.io project ID is set."""
49			if "KEEN_PROJECT_ID" in os.environ or keen.project_id is not None:
50			return True
51
52			logger.warning("Keen.io Project ID is not set")
53			return False
54
55
56			def keen_has_master_key():
57			"""Check if Keen.io Master API key is set."""
58			if "KEEN_MASTER_KEY" in os.environ or keen.master_key is not None:
59			return True
60
61			logger.warning("Keen.io Master API Key is not set")
62			return False
63
64
65			def keen_has_write_key():
66			"""Check if Keen.io Write Key is set."""
67			if "KEEN_WRITE_KEY" in os.environ or keen.write_key is not None:
68			return True
69
70			logger.warning("Keen.io Write Key is not set")
71			return False
72
73
74			def keen_has_read_key():
75			"""Check if Keen.io Read key is set."""
76			if "KEEN_READ_KEY" in os.environ or keen.read_key is not None:
77			return True
78
79			logger.warning("Keen.io Read Key is not set")
80			return False
81
82
83			def keen_is_writable():
84			"""Check if login keys for Keen IO API are set, to allow writing."""
85			if keen_has_project_id() and keen_has_write_key():
86			return True
87
88			logger.warning("Keen.io Write Key is not set")
89			return False
90
91
92			def keen_is_readable():
93			"""Check if login keys for Keen IO API are set, to allow reading."""
94			if keen_has_project_id() and keen_has_read_key():
95			return True
96
97			logger.warning("Keen.io Read Key is not set")
98			return False
99
100
101			def keen_io_generate_read_key(repo):
102			"""
103			Create scoped key for reading only the build-stages related data.
104
105			Param repo : github repository slug (fe. buildtimetrend/python-lib)
106			"""
107			if not keen_has_master_key():
108			logger.warning("Keen.io Read Key was not created,"
109			" keen.master_key is not defined.")
110			return None
111
112			master_key = keen.master_key or os.environ.get("KEEN_MASTER_KEY")
113
114			privileges = {
115			"allowed_operations": ["read"]
116			}
117
118			if repo is not None:
119			privileges["filters"] = [get_repo_filter(repo)]
120
121			logger.info("Keen.io Read Key is created for %s", repo)
122			return scoped_keys.encrypt(master_key, privileges)
123
124
125			def keen_io_generate_write_key():
126			"""Create scoped key for write access to Keen.io database."""
127			if not keen_has_master_key():
128			logger.warning("Keen.io Write Key was not created,"
129			" keen.master_key is not defined.")
130			return None
131
132			master_key = keen.master_key or os.environ.get("KEEN_MASTER_KEY")
133
134			privileges = {
135			"allowed_operations": ["write"]
136			}
137
138			logger.info("Keen.io Write Key is created")
139			return scoped_keys.encrypt(master_key, privileges)
140
141
142			def send_build_data(build):
143			"""Send build data generated by client to keen.io."""
144			if keen_is_writable():
145			logger.info("Sending client build data to Keen.io")
146			keen_add_event("build_jobs", {"job": build.to_dict()})
147			keen_add_events("build_stages", build.stages_to_list())
148
149
150			def send_build_data_service(build):
151			"""Send build data generated by service to keen.io."""
152			if keen_is_writable():
153			logger.info("Sending service build data to Keen.io")
154			keen_add_event("build_jobs", {"job": build.to_dict()})
155			keen_add_events("build_substages", build.stages_to_list())
156
157
158			def keen_add_event(event_collection, payload):
159			"""
160			Wrapper for keen.add_event(), adds project info.
161
162			Param event_collection : collection event data is submitted to
163			Param payload : data that is submitted
164			"""
165			# add project info to this event
166			payload = add_project_info_dict(payload)
167
168			# submit list of events to Keen.io
169			keen.add_event(event_collection, payload)
170
171
172			def keen_add_events(event_collection, payload):
173			"""
174			Wrapper for keen.add_events(), adds project info to each event.
175
176			Param event_collection : collection event data is submitted to
177			Param payload : array of events that is submitted
178			"""
179			# add project info to each event
180			payload = add_project_info_list(payload)
181
182			# submit list of events to Keen.io
183			keen.add_events({event_collection: payload})
184
185
186			def add_project_info_dict(payload):
187			"""
188			Add project info to a dictonary.
189
190			Param payload: dictonary payload
191			"""
192			# check if payload is a dictionary, throws an exception if it isn't
193			check_dict(payload, "payload")
194
195			payload_as_dict = copy.deepcopy(payload)
196
197			payload_as_dict[KEEN_PROJECT_INFO_NAME] = Settings().get_project_info()
198
199			if "job" in payload:
200			# override project_name, set to build_job repo
201			if "repo" in payload["job"]:
202			payload_as_dict[KEEN_PROJECT_INFO_NAME]["project_name"] = \
203			payload["job"]["repo"]
204
205			# override timestamp, set to finished_at timestamp
206			if "finished_at" in payload["job"]:
207			payload_as_dict["keen"] = {
208			"timestamp": payload["job"]["finished_at"]["isotimestamp"]
209			}
210
211			return payload_as_dict
212
213
214			def add_project_info_list(payload):
215			"""
216			Add project info to a list of dictionaries.
217
218			Param payload: list of dictionaries
219			"""
220			# check if payload is a list, throws an exception if it isn't
221			is_list(payload, "payload")
222
223			payload_as_list = []
224
225			# loop over dicts in payload and add project info to each one
226			for event_dict in payload:
227			payload_as_list.append(add_project_info_dict(event_dict))
228
229			return payload_as_list
230
231
232			def get_dashboard_keen_config(repo):
233			"""
234			Generate the Keen.io settings for the configuration of the dashboard.
235
236			The dashboard is Javascript powered HTML file that contains the
237			graphs generated by Keen.io.
238			"""
239			# initialise config settings
240			keen_config = {}
241
242			if not keen_has_project_id() or not keen_has_master_key():
243			logger.warning("Keen.io related config settings could not be created,"
244			" keen.project_id and/or keen.master_key"
245			" are not defined.")
246			return keen_config
247
248			# set keen project ID
249			if keen.project_id is None:
250			keen_project_id = os.environ["KEEN_PROJECT_ID"]
251			else:
252			keen_project_id = keen.project_id
253			keen_config['projectId'] = str(keen_project_id)
254
255			# generate read key
256			read_key = keen_io_generate_read_key(repo)
257			if read_key is not None:
258			# convert bytes to string
259			if isinstance(read_key, bytes):
260			read_key = read_key.decode('utf-8')
261
262			keen_config['readKey'] = str(read_key)
263
264			# return keen config settings
265			return keen_config
266
267
268			def get_dashboard_config_dict(repo, extra=None):
269			"""
270			Generate the configuration settings for the dashboard.
271
272			The dashboard is Javascript powered HTML file that contains the
273			graphs generated by Keen.io.
274
275			Parameters:
276			- repo : repo name (fe. buildtimetrend/service)
277			- extra : dictionary of extra config settings, format : {"name" : "value"}
278			"""
279			# initialise config settings dictionaries
280			config = {}
281
282			# add repo and project name
283			if repo is not None and not repo == "":
284			# merge extra settings into existing config dictionary
285			config.update({
286			'projectName': str(repo),
287			'repoName': str(repo)
288			})
289
290			# add extra config parameters
291			if extra is not None and check_dict(extra, "extra"):
292			config.update(extra)
293
294			return config
295
296
297			def get_dashboard_config(repo, extra=None):
298			"""
299			Generate the configuration settings for the dashboard.
300
301			The dashboard is Javascript powered HTML file that contains the
302			graphs generated by Keen.io.
303
304			Parameters:
305			- repo : repo name (fe. buildtimetrend/service)
306			- extra : dictionary of extra config settings, format : {"name" : "value"}
307			"""
308			# initialise config settings dictionaries
309			config = get_dashboard_config_dict(repo, extra)
310			keen_config = get_dashboard_keen_config(repo)
311
312			# create configuration as a string
313			return "var config = {};\nvar keenConfig = {};".format(config, keen_config)
314
315
316			def generate_dashboard_config_file(repo):
317			"""
318			Generate the configuration file for the dashboard.
319
320			The dashboard is Javascript powered HTML file that contains the
321			graphs generated by Keen.io.
322
323			Parameters:
324			- repo : repo name (fe. buildtimetrend/service)
325			"""
326			# get config settings
327			config_string = get_dashboard_config(repo)
328
329			# write config file
330			config_file = Settings().get_setting("dashboard_configfile")
331			with open(config_file, 'w') as outfile:
332			outfile.write(config_string)
333
334			if check_file(config_file):
335			logger.info("Created trends dashboard config file %s", config_file)
336			return True
337			else:
338			logger.warning("The dashboard config file was not created")
339			return False
340
341
342			def get_avg_buildtime(repo=None, interval=None):
			0 ignored issues – show Duplication introduced 2015-11-22 12:44 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Duplicated code is one of the most pungent code smells. If you need to duplicate the same code in three or more different places, we strongly encourage you to look into extracting the code into a single class or operation. You can also find more detailed suggestions in the “Code” section of your repository. Loading history...
343			"""
344			Query Keen.io database and retrieve average build time.
345
346			Parameters :
347			- repo : repo name (fe. buildtimetrend/service)
348			- interval : timeframe, possible values : 'week', 'month', 'year',
349			anything else defaults to 'week'
350			"""
351			if repo is None or not keen_is_readable():
352			return -1
353
354			interval_data = check_time_interval(interval)
355
356			try:
357			return keen.average(
358			"build_jobs",
359			target_property="job.duration",
360			timeframe=interval_data['timeframe'],
361			max_age=interval_data['max_age'],
362			filters=[get_repo_filter(repo)]
363			)
364			except requests.ConnectionError:
365			logger.error("Connection to Keen.io API failed")
366			return -1
367			except keen.exceptions.KeenApiError as msg:
368			logger.error("Error in keenio.get_avg_buildtime() : " + str(msg))
369			return -1
370
371
372			def get_total_build_jobs(repo=None, interval=None):
			0 ignored issues – show Duplication introduced 2015-11-22 12:44 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Duplicated code is one of the most pungent code smells. If you need to duplicate the same code in three or more different places, we strongly encourage you to look into extracting the code into a single class or operation. You can also find more detailed suggestions in the “Code” section of your repository. Loading history...
373			"""
374			Query Keen.io database and retrieve total number of build jobs.
375
376			Parameters :
377			- repo : repo name (fe. buildtimetrend/service)
378			- interval : timeframe, possible values : 'week', 'month', 'year',
379			anything else defaults to 'week'
380			"""
381			if repo is None or not keen_is_readable():
382			return -1
383
384			interval_data = check_time_interval(interval)
385
386			try:
387			return keen.count_unique(
388			"build_jobs",
389			target_property="job.job",
390			timeframe=interval_data['timeframe'],
391			max_age=interval_data['max_age'],
392			filters=[get_repo_filter(repo)]
393			)
394			except requests.ConnectionError:
395			logger.error("Connection to Keen.io API failed")
396			return -1
397			except keen.exceptions.KeenApiError as msg:
398			logger.error("Error in keenio.get_total_build_jobs() : " + str(msg))
399			return -1
400
401
402			def get_passed_build_jobs(repo=None, interval=None):
403			"""
404			Query Keen.io database and retrieve total number of build jobs that passed.
405
406			Parameters :
407			- repo : repo name (fe. buildtimetrend/service)
408			- interval : timeframe, possible values : 'week', 'month', 'year',
409			anything else defaults to 'week'
410			"""
411			if repo is None or not keen_is_readable():
412			return -1
413
414			interval_data = check_time_interval(interval)
415
416			try:
417			return keen.count_unique(
418			"build_jobs",
419			target_property="job.job",
420			timeframe=interval_data['timeframe'],
421			max_age=interval_data['max_age'],
422			filters=[
423			get_repo_filter(repo),
424			{
425			"property_name": "job.result",
426			"operator": "eq",
427			"property_value": "passed"
428			}
429			]
430			)
431			except requests.ConnectionError:
432			logger.error("Connection to Keen.io API failed")
433			return -1
434			except keen.exceptions.KeenApiError as msg:
435			logger.error("Error in keenio.get_passed_build_jobs() : " + str(msg))
436			return -1
437
438
439			def get_pct_passed_build_jobs(repo=None, interval=None):
440			"""
441			Calculate percentage of passed build jobs.
442
443			Parameters :
444			- repo : repo name (fe. buildtimetrend/service)
445			- interval : timeframe, possible values : 'week', 'month', 'year',
446			anything else defaults to 'week'
447			"""
448			total_jobs = get_total_build_jobs(repo, interval)
449			passed_jobs = get_passed_build_jobs(repo, interval)
450
451			logger.debug("passed/total build jobs : %d/%d", passed_jobs, total_jobs)
452
453			# calculate percentage if at least one job was executed
454			# passed is a valid number (not -1)
455			if total_jobs > 0 and passed_jobs >= 0:
456			return int(float(passed_jobs) / float(total_jobs) * 100.0)
457
458			return -1
459
460
461			def get_result_color(value=0, ok_thershold=90, warning_thershold=70):
462			"""
463			Get color code that corresponds to the result.
464
465			Parameters:
466			- value : value to check
467			- ok_thershold : OK threshold value
468			- warning_thershold : warning thershold value
469			"""
470			if not(type(value) in (int, float) and
471			type(ok_thershold) in (int, float) and
472			type(warning_thershold) in (int, float)):
473			return "lightgrey"
474
475			# check thresholds
476			if value >= ok_thershold:
477			return "green"
478			elif value >= warning_thershold:
479			return "yellow"
480			else:
481			return "red"
482
483
484			def get_total_builds(repo=None, interval=None):
			0 ignored issues – show Duplication introduced 2015-11-22 12:44 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Duplicated code is one of the most pungent code smells. If you need to duplicate the same code in three or more different places, we strongly encourage you to look into extracting the code into a single class or operation. You can also find more detailed suggestions in the “Code” section of your repository. Loading history...
485			"""
486			Query Keen.io database and retrieve total number of builds.
487
488			Parameters :
489			- repo : repo name (fe. buildtimetrend/service)
490			- interval : timeframe, possible values : 'week', 'month', 'year',
491			anything else defaults to 'week'
492			"""
493			if repo is None or not keen_is_readable():
494			return -1
495
496			interval_data = check_time_interval(interval)
497
498			try:
499			return keen.count_unique(
500			"build_jobs",
501			target_property="job.build",
502			timeframe=interval_data['timeframe'],
503			max_age=interval_data['max_age'],
504			filters=[get_repo_filter(repo)]
505			)
506			except requests.ConnectionError:
507			logger.error("Connection to Keen.io API failed")
508			return -1
509			except keen.exceptions.KeenApiError as msg:
510			logger.error("Error in keenio.get_total_builds() : " + str(msg))
511			return -1
512
513
514			def get_latest_buildtime(repo=None):
515			"""
516			Query Keen.io database and retrieve buildtime duration of last build.
517
518			Parameters :
519			- repo : repo name (fe. buildtimetrend/python-lib)
520			"""
521			if repo is None or not keen_is_readable():
522			return -1
523
524			try:
525			result = keen.extraction(
526			"build_jobs",
527			property_names="job.duration",
528			latest=1,
529			filters=[get_repo_filter(repo)]
530			)
531			except requests.ConnectionError:
532			logger.error("Connection to Keen.io API failed")
533			return -1
534			except keen.exceptions.KeenApiError as msg:
535			logger.error("Error in keenio.get_latest_buildtime() : " + str(msg))
536			return -1
537
538			if is_list(result) and len(result) > 0 and \
539			check_dict(result[0], None, ['job']) and \
540			check_dict(result[0]['job'], None, ['duration']):
541			return result[0]['job']['duration']
542
543			return -1
544
545
546			def has_build_id(repo=None, build_id=None):
547			"""
548			Check if build_id exists in Keen.io database.
549
550			Parameters :
551			- repo : repo name (fe. buildtimetrend/python-lib)
552			- build_id : ID of the build
553			"""
554			if repo is None or build_id is None:
555			logger.error("Repo or build_id is not set")
556			raise ValueError("Repo or build_id is not set")
557			if not keen_is_readable():
558			raise SystemError("Keen.io Project ID or API Read Key is not set")
559
560			try:
561			count = keen.count(
562			"build_jobs",
563			filters=[get_repo_filter(repo), {
564			"property_name": "job.build",
565			"operator": "eq",
566			"property_value": str(build_id)}]
567			)
568			except requests.ConnectionError:
569			logger.error("Connection to Keen.io API failed")
570			raise SystemError("Connection to Keen.io API failed")
571			except keen.exceptions.KeenApiError as msg:
572			logger.error("Error in keenio.has_build_id : " + str(msg))
573			raise SystemError(msg)
574
575			return count > 0
576
577
578			def get_all_projects():
579			"""Query Keen.io database and retrieve a list of all projects."""
580			if not keen_is_readable():
581			return []
582
583			try:
584			result = keen.select_unique(
585			"build_jobs",
586			"buildtime_trend.project_name",
587			max_age=3600 * 24 # cache for 24 hours
588			)
589			except requests.ConnectionError:
590			logger.error("Connection to Keen.io API failed")
591			return []
592			except keen.exceptions.KeenApiError as msg:
593			logger.error("Error in keenio.get_all_projects() : " + str(msg))
594			return []
595
596			if type(result) is list:
597			return result
598
599			return []
600
601
602			def get_repo_filter(repo=None):
603			"""
604			Return filter for analysis request.
605
606			Parameters
607			- repo : repo slug name, fe. buildtimetrend/python-lib
608			"""
609			if repo is None:
610			return None
611
612			return {
613			"property_name": "%s.project_name" % KEEN_PROJECT_INFO_NAME,
614			"operator": "eq",
615			"property_value": str(repo)
616			}
617
618
619			def check_time_interval(interval=None):
620			"""
621			Check time interval and returns corresponding parameters.
622
623			Parameters :
624			- interval : timeframe, possible values : 'week', 'month', 'year',
625			anything else defaults to 'week'
626			"""
627			if is_string(interval):
628			# convert to lowercase
629			interval = interval.lower()
630
631			if interval in TIME_INTERVALS:
632			return TIME_INTERVALS[interval]
633
634			return TIME_INTERVALS['week']
635

buildtimetrend / python-lib

Push — master ( 232dbd...dd6f0d )

buildtimetrend.get_latest_buildtime() F

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like