Merge "Add DB API function to get multiple tests by test_id"

This commit is contained in:
Jenkins 2017-01-10 23:29:41 +00:00 committed by Gerrit Code Review
commit 4d1a5aa6a5
3 changed files with 40 additions and 0 deletions

View File

@ -0,0 +1,4 @@
---
features:
- Add a new DB API function, get_tests_by_test_ids, to return a list of Test
model objects give an list of test_ids

View File

@ -723,6 +723,20 @@ def get_test_by_test_id(test_id, session=None):
return test
def get_tests_by_test_ids(test_ids, session=None):
"""Get tests that match input test_ids
:param list test_ids: A list of test_ids (aka the test name) for the test
:param session: Optional session object if one isn't provided a new session
will be acquired for the duration of this operation
:return: A list of the specified test objects
:rtype: list
"""
session = session or get_session()
return db_utils.model_query(models.Test, session).filter(
models.Test.test_id.in_(test_ids)).all()
def get_run_id_from_uuid(uuid, session=None):
"""Get the id for a run by it's uuid

View File

@ -66,6 +66,28 @@ class TestDatabaseAPI(base.TestCase):
res = api.get_test_by_test_id('fake_test')
self.assertIsNone(res)
def test_get_tests_by_test_ids(self):
test_a = api.create_test('fake_test1', 2, 1, 1, 1.2)
test_b = api.create_test('fake_test2', 4, 2, 2, 2.3)
test_c = api.create_test('fake_test3', 6, 3, 3, 3.3)
test_d = api.create_test('fake_test4', 8, 4, 4, 4.3)
result = api.get_tests_by_test_ids(
['fake_test1', 'fake_test2', 'fake_test3'])
result_ids = [x.id for x in result]
self.assertIn(test_a.id, result_ids)
self.assertIn(test_b.id, result_ids)
self.assertIn(test_c.id, result_ids)
self.assertNotIn(test_d.id, result_ids)
def test_get_tests_by_test_ids_no_matches(self):
api.create_test('fake_test5', 2, 1, 1, 1.2)
api.create_test('fake_test6', 4, 2, 2, 2.3)
api.create_test('fake_test7', 6, 3, 3, 3.3)
api.create_test('fake_test8', 8, 4, 4, 4.3)
result = api.get_tests_by_test_ids(
['fake_test1', 'fake_test2', 'fake_test3'])
self.assertEqual([], result)
def test_create_run_and_list(self):
res = api.create_run()
self.assertIsNotNone(res)