From a2c496f77524db674b56c5d42b612faf548f0a2e Mon Sep 17 00:00:00 2001 From: Matthew Treinish Date: Wed, 23 Sep 2015 00:33:16 -0400 Subject: [PATCH] Add tests for get_tests_run_dicts_from_run_id() and fix issues 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 --- subunit2sql/db/api.py | 18 ++++++-- subunit2sql/tests/db/test_api.py | 78 ++++++++++++++++++++++++++++++++ 2 files changed, 91 insertions(+), 5 deletions(-) diff --git a/subunit2sql/db/api.py b/subunit2sql/db/api.py index 75486a2..86d23fe 100644 --- a/subunit2sql/db/api.py +++ b/subunit2sql/db/api.py @@ -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, diff --git a/subunit2sql/tests/db/test_api.py b/subunit2sql/tests/db/test_api.py index 149de96..2ba1087 100644 --- a/subunit2sql/tests/db/test_api.py +++ b/subunit2sql/tests/db/test_api.py @@ -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')