opendev/subunit2sql
Code Issues Proposed changes

Add a db api method to get a time series of runs by key

Browse Source 
This commit adds a new db api method that will get a time series
of a summary view of runs which are grouped by a metadata key.

Change-Id: Id04241249cf4ff14883354b0e69a7685c7f770cf
This commit is contained in:
Matthew Treinish
2015-09-16 00:21:08 -04:00
parent 8fe45ec7fe
commit 05f176eb3c
1 changed files with 49 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
subunit2sql/db/api.py
View File

@@ -930,6 +930,55 @@ def get_failing_test_ids_from_runs_by_key_value(key, value, session=None):
return [test_id[0] for test_id in test_ids]
def get_all_runs_time_series_by_key(key, start_date=None,
stop_date=None, session=None):
"""Get a time series of run summaries grouped by a key
This method will get a time series dictionary of run summary views which
are grouped by the values of the specified key
:param str key: the key to use for grouping the run summaries
:param str start_date: Optional start date to filter results on
:param str stop_date: Optional stop date to filter results on
:param session: optional session object if one isn't provided a new session
will be acquired for the duration of this operation
:return runs: a time series dictionary of runs grouped by values of the
specified key
:rtype: dict
"""
session = session or get_session()
runs_query = db_utils.model_query(models.Run, session).join(
models.RunMetadata).filter(models.RunMetadata.key == key)
runs_query = _filter_runs_by_date(runs_query, start_date, stop_date)
runs_query = runs_query.values(models.Run.run_at,
models.Run.passes,
models.Run.fails,
models.Run.skips,
models.RunMetadata.value)
runs = {}
for run in runs_query:
if run[0] not in runs:
runs[run[0]] = {run[4]: [{
'pass': run[1],
'fail': run[2],
'skip': run[3],
}]}
else:
if run[4] not in runs[run[0]].keys():
runs[run[0]][run[4]] = [{
'pass': run[1],
'fail': run[2],
'skip': run[3],
}]
else:
runs[run[0]][run[4]].append({
'pass': run[1],
'fail': run[2],
'skip': run[3],
})
return runs
def add_test_run_attachments(attach_dict, test_run_id, session=None):
"""Add attachments a specific test run.