opendev/subunit2sql
Code Issues Proposed changes

Add tests for get_tests_run_dicts_from_run_id() and fix issues

Browse Source 
This commit adds unit tests for the get_tests_run_dicts_from_run_id()
db api method. It adds tests for a variety of different data scenarios
where different pieces of data could be missing.

In the process of adding the tests a number of bugs with the function
were uncovered. If there was no test run metadata for a particular
test run a result wouldn't be returned for that test run because of
the use of an inner join. If either timestamp field was missing the
method would fail because it would attempt to set the microseconds
on a NoneType object. All of these were addressed as part of this
commit to enable tests to work.

Change-Id: Ifde4158137fd4e985ce9f38608cea553348cd2aa
This commit is contained in:
Matthew Treinish 2015-09-23 00:33:16 -04:00
parent dafc36a4ca
commit a2c496f775
No known key found for this signature in database
GPG Key ID: FD12A0F214C9E177
2 changed files with 91 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
subunit2sql/db/api.py
View File

@ -679,7 +679,7 @@ def get_tests_run_dicts_from_run_id(run_id, session=None):
"""
session = session or get_session()
query = db_utils.model_query(models.Test, session=session).join(
models.TestRun).filter_by(run_id=run_id).join(
models.TestRun).filter(models.TestRun.run_id == run_id).outerjoin(
models.TestRunMetadata).values(
models.Test.test_id,
models.TestRun.status,
@ -692,10 +692,18 @@ def get_tests_run_dicts_from_run_id(run_id, session=None):
test_runs = {}
for test_run in query:
if test_run[0] not in test_runs:
start_time = test_run[2]
start_time = start_time.replace(microsecond=test_run[3])
stop_time = test_run[4]
stop_time = stop_time.replace(microsecond=test_run[5])
# If there is no start_time set to None
if test_run[2]:
start_time = test_run[2]
start_time = start_time.replace(microsecond=test_run[3])
else:
start_time = None
# If there is no stop_time set to None
if test_run[4]:
stop_time = test_run[4]
stop_time = stop_time.replace(microsecond=test_run[5])
else:
stop_time = None
test_runs[test_run[0]] = {
'status': test_run[1],
'start_time': start_time,

78
subunit2sql/tests/db/test_api.py
View File

@ -12,6 +12,8 @@
# License for the specific language governing permissions and limitations
# under the License.
import datetime
import testscenarios
from subunit2sql.db import api
@ -59,6 +61,82 @@ class TestDatabaseAPI(base.TestCase):
self.assertEqual(len(all_runs), 1)
self.assertEqual(res.id, all_runs[0].id)
def test_get_test_runs_dicts_with_no_meta(self):
run = api.create_run()
test_a = api.create_test('fake_test')
start_time = datetime.datetime.utcnow()
stop_time = datetime.datetime.utcnow()
api.create_test_run(test_a.id, run.id, 'success',
start_time, stop_time)
test_run_dict = api.get_tests_run_dicts_from_run_id(run.id)
self.assertEqual(1, len(test_run_dict))
self.assertIn('fake_test', test_run_dict)
self.assertEqual(test_run_dict['fake_test']['status'], 'success')
self.assertEqual(test_run_dict['fake_test']['start_time'], start_time)
self.assertEqual(test_run_dict['fake_test']['stop_time'], stop_time)
self.assertNotIn('metadata', test_run_dict['fake_test'])
def test_get_test_runs_dicts_with_no_stop_time(self):
run = api.create_run()
test_a = api.create_test('fake_test')
start_time = datetime.datetime.utcnow()
stop_time = None
api.create_test_run(test_a.id, run.id, 'success',
start_time, stop_time)
test_run_dict = api.get_tests_run_dicts_from_run_id(run.id)
self.assertEqual(1, len(test_run_dict))
self.assertIn('fake_test', test_run_dict)
self.assertEqual(test_run_dict['fake_test']['status'], 'success')
self.assertEqual(test_run_dict['fake_test']['start_time'], start_time)
self.assertEqual(test_run_dict['fake_test']['stop_time'], stop_time)
def test_get_test_runs_dicts_with_no_start_time(self):
run = api.create_run()
test_a = api.create_test('fake_test')
stop_time = datetime.datetime.utcnow()
start_time = None
api.create_test_run(test_a.id, run.id, 'success',
start_time, stop_time)
test_run_dict = api.get_tests_run_dicts_from_run_id(run.id)
self.assertEqual(1, len(test_run_dict))
self.assertIn('fake_test', test_run_dict)
self.assertEqual(test_run_dict['fake_test']['status'], 'success')
self.assertEqual(test_run_dict['fake_test']['start_time'], start_time)
self.assertEqual(test_run_dict['fake_test']['stop_time'], stop_time)
def test_get_test_runs_dicts_with_no_start_or_stop_time(self):
run = api.create_run()
test_a = api.create_test('fake_test')
stop_time = None
start_time = None
api.create_test_run(test_a.id, run.id, 'success',
start_time, stop_time)
test_run_dict = api.get_tests_run_dicts_from_run_id(run.id)
self.assertEqual(1, len(test_run_dict))
self.assertIn('fake_test', test_run_dict)
self.assertEqual(test_run_dict['fake_test']['status'], 'success')
self.assertEqual(test_run_dict['fake_test']['start_time'], start_time)
self.assertEqual(test_run_dict['fake_test']['stop_time'], stop_time)
def test_get_test_runs_dicts_with_meta(self):
run = api.create_run()
test_a = api.create_test('fake_test')
test_run = api.create_test_run(test_a.id, run.id, 'success',
datetime.datetime.utcnow(),
datetime.datetime.utcnow())
run_meta = {
'key_a': 'value_b',
'key_b': 'value_a',
'attrs': 'test,smoke,notatest',
}
api.add_test_run_metadata(run_meta, test_run.id)
test_run_dict = api.get_tests_run_dicts_from_run_id(run.id)
self.assertEqual(3, len(test_run_dict['fake_test']['metadata']))
for meta in run_meta:
self.assertIn(meta, test_run_dict['fake_test']['metadata'])
self.assertEqual(run_meta[meta],
test_run_dict['fake_test']['metadata'][meta])
def test_create_test_run_and_list(self):
run = api.create_run()
test = api.create_test('fake_test')