Merge "Add attachments flag to get_test_runs_by_status"

This commit is contained in:
Zuul 2019-07-18 05:52:22 +00:00 committed by Gerrit Code Review
commit aeb9d0940f
3 changed files with 140 additions and 20 deletions

View File

@ -0,0 +1,6 @@
---
features:
- An 'include_attachments' boolean option has been added to the db api
method, get_test_runs_by_status_for_run_ids(). When enabled, a list
of attachments with labels in each of the test runs dictionaries will
be returned.

View File

@ -1852,7 +1852,8 @@ def get_runs_by_status_grouped_by_run_metadata(key, start_date=None,
def get_test_runs_by_status_for_run_ids(status, run_ids, key=None,
session=None, include_run_id=False):
session=None, include_run_id=False,
include_attachments=False):
"""Get a list of test run dicts by status for all the specified runs
:param str status: The test status to filter the returned test runs on
@ -1864,6 +1865,8 @@ def get_test_runs_by_status_for_run_ids(status, run_ids, key=None,
will be acquired for the duration of this operation
:param bool include_run_id: boolean flag to enable including the run uuid
in the test run dicts returned
:param bool include_attachments: boolean flag to enable including a list
of attachments with labels in the test run dicts returned
:return test_runs: A list of dicts for the test_runs and associated data
:rtype: list
@ -1874,34 +1877,43 @@ def get_test_runs_by_status_for_run_ids(status, run_ids, key=None,
models.Test, models.TestRun.test_id == models.Test.id).join(
models.Run, models.TestRun.run_id == models.Run.id).filter(
models.Run.uuid.in_(run_ids))
query_values = [models.Test.test_id, models.Run.artifacts,
models.TestRun.start_time,
models.TestRun.start_time_microsecond,
models.TestRun.stop_time,
models.TestRun.stop_time_microsecond,
models.Run.uuid]
if include_attachments:
query = query.join(
models.Attachments,
models.TestRun.id == models.Attachments.test_run_id)
query_values.extend([models.Attachments.label,
models.Attachments.attachment])
if key:
query = query.join(
models.RunMetadata,
models.TestRun.run_id == models.RunMetadata.run_id).filter(
models.RunMetadata.key == key)
results = query.values(models.Test.test_id, models.Run.artifacts,
models.TestRun.start_time,
models.TestRun.start_time_microsecond,
models.TestRun.stop_time,
models.TestRun.stop_time_microsecond,
models.RunMetadata.value,
models.Run.uuid)
else:
results = query.values(models.Test.test_id, models.Run.artifacts,
models.TestRun.start_time,
models.TestRun.start_time_microsecond,
models.TestRun.stop_time,
models.TestRun.stop_time_microsecond,
models.Run.uuid)
test_runs = []
query_values.append(models.RunMetadata.value)
results = query.values(*query_values)
test_runs = {}
for result in results:
test_run = {
'test_id': result.test_id,
'link': result.artifacts,
'start_time': result.start_time,
'stop_time': result.stop_time,
'stop_time': result.stop_time
}
test_run_key = result.uuid + result.test_id
if include_attachments:
attachment_dict = {'label': result.label,
'attachment': result.attachment}
if test_run_key in test_runs:
test_run = test_runs[test_run_key]
test_run['attachments'].append(attachment_dict)
continue
test_run['attachments'] = [attachment_dict]
if include_run_id:
test_run['uuid'] = result.uuid
if result.start_time_microsecond is not None:
@ -1912,8 +1924,8 @@ def get_test_runs_by_status_for_run_ids(status, run_ids, key=None,
microsecond=result.stop_time_microsecond)
if hasattr(result, "value"):
test_run[key] = result.value
test_runs.append(test_run)
return test_runs
test_runs[test_run_key] = test_run
return list(test_runs.values())
def get_all_run_metadata_keys(session=None):

View File

@ -779,6 +779,108 @@ class TestDatabaseAPI(base.TestCase):
'stop_time': stop_timestamp,
}, result[0])
def test_get_test_runs_by_status_for_run_ids_with_attachments(self):
attach_dict = {'attach_label': b'attach',
'attach_label_a': b'attach_a'}
run_b = api.create_run(artifacts='fake_url')
run_a = api.create_run()
run_c = api.create_run()
test_a = api.create_test('fake_test')
start_timestamp = datetime.datetime(1914, 6, 28, 10, 45, 0)
stop_timestamp = datetime.datetime(1914, 6, 28, 10, 50, 0)
api.create_test_run(test_a.id, run_a.id, 'success',
datetime.datetime.utcnow())
test_run_ab = api.create_test_run(test_a.id, run_b.id, 'fail',
start_timestamp, stop_timestamp)
api.create_test_run(test_a.id, run_c.id, 'success',
datetime.datetime.utcnow())
api.add_test_run_attachments(attach_dict, test_run_ab.id)
result = api.get_test_runs_by_status_for_run_ids(
'fail',
[run_a.uuid, run_b.uuid, run_c.uuid],
include_attachments=True)
self.assertEqual(1, len(result))
expected_attachments = [{'label': 'attach_label',
'attachment': b'attach'},
{'label': 'attach_label_a',
'attachment': b'attach_a'}]
self.assertItemsEqual(expected_attachments, result[0]['attachments'])
def test_get_many_test_runs_by_status_for_run_ids_with_attachments(self):
attach_dict_test_b_run_a = {'attach_label': b'attach1',
'run_a': b'test_b'}
attach_dict_test_a_run_b = {'attach_label': b'attach2',
'run_b': b'test_a'}
attach_dict_test_a_run_c = {'attach_label': b'attach3',
'run_c': b'test_a'}
attach_dict_test_b_run_c = {'attach_label': b'attach4',
'run_c': b'test_b'}
run_a = api.create_run(artifacts='fake_url')
run_b = api.create_run(artifacts='fake_url')
run_c = api.create_run(artifacts='fake_url')
test_a = api.create_test('fake_test_a')
test_b = api.create_test('fake_test_b')
start_timestamp = datetime.datetime(1914, 6, 28, 11, 45, 1)
stop_timestamp = datetime.datetime(1914, 6, 28, 11, 50, 1)
test_b_run_a = api.create_test_run(test_b.id, run_a.id, 'fail',
start_timestamp, stop_timestamp)
test_a_run_b = api.create_test_run(test_a.id, run_b.id, 'fail',
start_timestamp, stop_timestamp)
test_a_run_c = api.create_test_run(test_a.id, run_c.id, 'fail',
start_timestamp, stop_timestamp)
test_b_run_c = api.create_test_run(test_b.id, run_c.id, 'fail',
start_timestamp, stop_timestamp)
api.add_test_run_attachments(attach_dict_test_b_run_a, test_b_run_a.id)
api.add_test_run_attachments(attach_dict_test_a_run_b, test_a_run_b.id)
api.add_test_run_attachments(attach_dict_test_a_run_c, test_a_run_c.id)
api.add_test_run_attachments(attach_dict_test_b_run_c, test_b_run_c.id)
results = api.get_test_runs_by_status_for_run_ids(
'fail',
[run_a.uuid, run_b.uuid, run_c.uuid],
include_attachments=True,
include_run_id=True)
self.assertEqual(4, len(results))
expected_attachments_test_b_run_a = [{'label': u'attach_label',
'attachment': b'attach1'},
{'label': u'run_a',
'attachment': b'test_b'}]
expected_attachments_test_a_run_b = [{'label': u'attach_label',
'attachment': b'attach2'},
{'label': u'run_b',
'attachment': b'test_a'}]
expected_attachments_test_a_run_c = [{'label': u'attach_label',
'attachment': b'attach3'},
{'label': u'run_c',
'attachment': b'test_a'}]
expected_attachments_test_b_run_c = [{'label': u'attach_label',
'attachment': b'attach4'},
{'label': u'run_c',
'attachment': b'test_b'}]
test_b_run_a_called = False
test_a_run_b_called = False
test_a_run_c_called = False
test_b_run_c_called = False
for result in results:
uuid = result['uuid']
test_id = result['test_id']
if test_id == test_b.test_id and uuid == run_a.uuid:
expected_attachments = expected_attachments_test_b_run_a
test_b_run_a_called = True
elif test_id == test_a.test_id and uuid == run_b.uuid:
expected_attachments = expected_attachments_test_a_run_b
test_a_run_b_called = True
elif test_id == test_a.test_id and uuid == run_c.uuid:
expected_attachments = expected_attachments_test_a_run_c
test_a_run_c_called = True
elif test_id == test_b.test_id and uuid == run_c.uuid:
expected_attachments = expected_attachments_test_b_run_c
test_b_run_c_called = True
self.assertItemsEqual(expected_attachments, result['attachments'])
self.assertTrue(test_b_run_a_called)
self.assertTrue(test_a_run_b_called)
self.assertTrue(test_a_run_c_called)
self.assertTrue(test_b_run_c_called)
def test_get_test_runs_by_status_for_run_ids_with_meta(self):
run_b = api.create_run(artifacts='fake_url')
run_a = api.create_run()