subunit2sql/subunit2sql/tests/db/test_api.py

1446 lines
66 KiB
Python

# Copyright 2015 Hewlett-Packard Development Company, L.P.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import datetime
import types
from six import moves
import testscenarios
from subunit2sql.db import api
from subunit2sql.db import models
from subunit2sql.tests import base
from subunit2sql.tests import db_test_utils
from subunit2sql.tests import subunit2sql_fixtures as fixtures
load_tests = testscenarios.load_tests_apply_scenarios
class TestDatabaseAPI(base.TestCase):
scenarios = [
('mysql', {'dialect': 'mysql'}),
('postgresql', {'dialect': 'postgres'}),
('sqlite', {'dialect': 'sqlite'})
]
def setUp(self):
super(TestDatabaseAPI, self).setUp()
self.useFixture(fixtures.LockFixture(self.dialect))
if not db_test_utils.is_backend_avail(self.dialect):
raise self.skipTest('%s is not available' % self.dialect)
if self.dialect == 'mysql':
self.useFixture(fixtures.MySQLConfFixture())
elif self.dialect == 'postgres':
self.useFixture(fixtures.PostgresConfFixture())
elif self.dialect == 'sqlite':
self.useFixture(fixtures.SqliteConfFixture())
self.useFixture(fixtures.Database())
def test_create_test(self):
api.create_test('1234')
res = api.get_all_tests()
self.assertEqual(len(res), 1)
self.assertEqual(res[0].test_id, '1234')
def test_create_test_and_get_by_test_id(self):
create_res = api.create_test('fake_test', 2, 1, 1, 1.2)
res = api.get_test_by_test_id('fake_test')
self.assertEqual(res.id, create_res.id)
self.assertEqual(res.test_id, 'fake_test')
self.assertEqual(res.run_time, 1.2)
self.assertEqual(res.run_count, 2)
def test_get_test_by_test_id_invalid_id(self):
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_get_test_with_sql_injection(self):
api.create_test("test_terror'); DROP TABLE tests;")
res = api.get_all_tests()
self.assertEqual(len(res), 1)
self.assertEqual(res[0].test_id, "test_terror'); DROP TABLE tests;")
def test_create_run_and_list(self):
res = api.create_run()
self.assertIsNotNone(res)
all_runs = api.get_all_runs()
self.assertEqual(len(all_runs), 1)
self.assertEqual(res.id, all_runs[0].id)
def test_update_run_and_list(self):
res = api.create_run()
dt = datetime.datetime.utcnow()
# NOTE(masayukig): The DateTime column in some DB envs can't store a
# microseconds resolution value. So this removes the microseconds value
# here.
dt = dt.replace(microsecond=0)
dt_dummy = dt - datetime.timedelta(days=1)
res_dummy = api.create_run(run_at=dt_dummy)
values = {'skips': 98, 'fails': 97, 'passes': 96, 'run_time': 1.123,
'artifacts': 'fake_url', 'run_at': dt}
updated_res = api.update_run(values, res.id)
all_runs = api.get_all_runs()
self.assertEqual(len(all_runs), 2)
for run in all_runs:
if run.id == res.id:
self.assertEqual(updated_res.id, run.id)
for key in values:
self.assertEqual(values[key], run[key])
self.assertEqual(run[key], updated_res[key])
elif run.id == res_dummy.id:
self.assertNotEqual(updated_res.id, run.id)
for key in values:
if key == 'run_at':
continue
self.assertNotEqual(values[key], run[key])
self.assertNotEqual(run[key], updated_res[key])
else:
self.fail('an unexpected run(%s) was found' % run.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().replace(microsecond=0)
stop_time = datetime.datetime.utcnow().replace(microsecond=0)
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.uuid)
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().replace(microsecond=0)
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.uuid)
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().replace(microsecond=0)
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.uuid)
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.uuid)
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.uuid)
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')
test_run = api.create_test_run(test.id, run.id, 'fail')
self.assertIsNotNone(test_run)
all_test_runs = api.get_all_test_runs()
self.assertEqual(len(all_test_runs), 1)
self.assertEqual(test_run.id, all_test_runs[0].id)
def test_get_test_runs_dicts_from_run_id_are_in_chrono_order(self):
run = api.create_run()
test_a = api.create_test('fake_test')
test_b = api.create_test('fake_test_2')
test_c = api.create_test('fake_test_3')
api.create_test_run(test_a.id, run.id, 'success',
datetime.datetime.utcnow())
api.create_test_run(test_b.id, run.id, 'success',
datetime.datetime(1914, 6, 28, 10, 45, 0))
api.create_test_run(test_c.id, run.id, 'success',
datetime.datetime(2014, 8, 26, 20, 00, 00))
test_run_dicts = api.get_tests_run_dicts_from_run_id(run.uuid)
self.assertEqual(len(test_run_dicts), 3)
prev = None
for test_run in test_run_dicts:
if prev == None:
prev = test_run
continue
self.assertTrue(test_run_dicts[test_run]['start_time'] >
test_run_dicts[prev]['start_time'])
prev = test_run
def test_get_test_run_duration(self):
start_time = datetime.datetime.utcnow()
stop_time = start_time + datetime.timedelta(minutes=3)
run = api.create_run()
test_a = api.create_test('fake_test')
test_run = api.create_test_run(test_a.id, run.id, 'success',
start_time, stop_time)
dur = api.get_test_run_duration(test_run.id)
self.assertEqual(180.0, dur)
def test_get_id_from_test_id(self):
test_a = api.create_test('fake_test')
id_value = api.get_id_from_test_id('fake_test')
self.assertEqual(test_a.id, id_value)
def test_get_run_times_all_test_runs(self):
timestamp_a = datetime.datetime.utcnow()
timestamp_b = timestamp_a + datetime.timedelta(seconds=3)
run_a = api.create_run()
run_b = api.create_run()
test_a = api.create_test('test_a')
test_b = api.create_test('test_b')
api.create_test_run(test_a.id, run_a.id, 'success', timestamp_a,
timestamp_b)
api.create_test_run(test_a.id, run_b.id, 'success', timestamp_a,
timestamp_b)
api.create_test_run(test_b.id, run_b.id, 'success', timestamp_a,
timestamp_b)
res = api.get_run_times_all_test_runs()
expected_dict = {
'test_a': [3, 3],
'test_b': [3]
}
self.assertEqual(expected_dict, res)
def test_get_run_times_all_test_runs_with_tests_filter(self):
timestamp_a = datetime.datetime.utcnow()
timestamp_b = timestamp_a + datetime.timedelta(seconds=3)
run_a = api.create_run()
run_b = api.create_run()
test_a = api.create_test('test_a')
test_b = api.create_test('test_b')
api.create_test_run(test_a.id, run_a.id, 'success', timestamp_a,
timestamp_b)
api.create_test_run(test_a.id, run_b.id, 'success', timestamp_a,
timestamp_b)
api.create_test_run(test_b.id, run_b.id, 'success', timestamp_a,
timestamp_b)
res = api.get_run_times_all_test_runs(tests=['test_a'])
expected_dict = {
'test_a': [3, 3],
}
self.assertEqual(expected_dict, res)
def test_get_test_runs_by_run_id(self):
run_b = api.create_run()
run_a = api.create_run()
run_c = api.create_run()
test_a = api.create_test('fake_test')
testrun_a = api.create_test_run(test_a.id, run_a.id, 'success',
datetime.datetime.utcnow())
testrun_b = api.create_test_run(test_a.id, run_b.id, 'success',
datetime.datetime.utcnow())
testrun_c = api.create_test_run(test_a.id, run_c.id, 'success',
datetime.datetime.utcnow())
test_runs_a = api.get_test_runs_by_run_id(run_a.uuid)
test_runs_b = api.get_test_runs_by_run_id(run_b.uuid)
test_runs_c = api.get_test_runs_by_run_id(run_c.uuid)
self.assertEqual(len(test_runs_a), 1)
self.assertEqual(testrun_a.id, test_runs_a[0].id)
self.assertEqual(testrun_a.status, test_runs_a[0].status)
self.assertEqual(len(test_runs_b), 1)
self.assertEqual(testrun_b.id, test_runs_b[0].id)
self.assertEqual(testrun_b.status, test_runs_b[0].status)
self.assertEqual(len(test_runs_c), 1)
self.assertEqual(testrun_c.id, test_runs_c[0].id)
self.assertEqual(testrun_c.status, test_runs_c[0].status)
def test_get_runs_by_status_grouped_by_run_metadata(self):
# Generating 20 runs:
# 10 with no failures
# 10 with 10 failures
# 7 in 2010/2011 each, 6 in 2012
# 10 in projecta/projectb each
for i in moves.range(20):
if i % 2 == 1:
fails = 10
else:
fails = 0
year = 2010 + (i % 3)
run_at = datetime.datetime(year, 1, i + 1, 12, 0, 0)
run = api.create_run(fails=fails, passes=10, run_at=run_at)
self.assertIsNotNone(run)
if i < 10:
project = 'projecta'
else:
project = 'projectb'
meta_dict = {'project': project}
api.add_run_metadata(meta_dict, run.id)
result = api.get_runs_by_status_grouped_by_run_metadata(
'project', start_date='2012-01-01', stop_date='2012-12-31')
# There should be two projects
self.assertEqual(2, len(result.keys()))
self.assertTrue('projecta' in result)
self.assertTrue('projectb' in result)
# There should be passes and failures
self.assertEqual(2, len(result['projecta'].keys()))
self.assertTrue('pass' in result['projecta'])
self.assertTrue('fail' in result['projecta'])
self.assertEqual(2, len(result['projectb'].keys()))
self.assertTrue('pass' in result['projectb'])
self.assertTrue('fail' in result['projectb'])
self.assertEqual(2, result['projecta']['pass'])
self.assertEqual(1, result['projecta']['fail'])
self.assertEqual(1, result['projectb']['pass'])
self.assertEqual(2, result['projectb']['fail'])
def test_get_time_series_runs_by_key_value(self):
runs = []
run_at = datetime.datetime.utcnow()
for run_num in moves.range(15):
run = api.create_run(run_num, run_num + 1, run_num + 2, 3,
run_at=run_at)
runs.append(run)
run_meta = {'test_key': 'fun', 'non_test': 'value-%s' % run_num}
if run_num >= 3:
run_meta = {'test_key': 'no-fun',
'non_test': 'value-%s' % run_num}
api.add_run_metadata(run_meta, run.id)
runs_time_series = api.get_time_series_runs_by_key_value('test_key',
'fun')
self.assertEqual(1, len(runs_time_series))
timestamp = list(runs_time_series.keys())[0]
self.assertEqual(3, len(runs_time_series[timestamp]))
for run_num in moves.range(3):
run_dict = {
'skip': run_num,
'fail': run_num + 1,
'pass': run_num + 2,
'id': runs[run_num].uuid,
'run_time': 3.0,
'metadata': {
u'test_key': u'fun',
u'non_test': u'value-%s' % run_num
}
}
self.assertIn(run_dict, runs_time_series[timestamp])
for run_num in moves.range(3, 14):
missing_run_dict = {
'skip': run_num,
'fail': run_num + 1,
'pass': run_num + 2,
'id': runs[run_num].id,
'run_time': 3.0,
'metadata': {
u'test_key': u'fun',
u'non_test': u'value-%s' % run_num
}
}
self.assertNotIn(missing_run_dict, runs_time_series[timestamp])
def test_get_test_runs_test_test_id(self):
run = api.create_run()
test_a = api.create_test('fake_test')
test_b = api.create_test('less_fake_test')
api.create_test_run(test_a.id, run.id, 'success')
api.create_test_run(test_b.id, run.id, 'success')
res = api.get_test_runs_by_test_test_id('less_fake_test')
self.assertEqual(1, len(res))
self.assertEqual(test_b.id, res[0].test_id)
self.assertEqual(run.id, res[0].run_id)
def test_get_test_runs_test_test_id_with_run_metadata(self):
run_a = api.create_run()
run_b = api.create_run()
api.add_run_metadata({'a_key': 'a_value'}, run_a.id)
api.add_run_metadata({'b_key': 'b_value'}, run_b.id)
test_a = api.create_test('fake_test')
test_b = api.create_test('less_fake_test')
api.create_test_run(test_a.id, run_a.id, 'success')
api.create_test_run(test_a.id, run_b.id, 'success')
api.create_test_run(test_b.id, run_a.id, 'success')
api.create_test_run(test_b.id, run_b.id, 'success')
res = api.get_test_runs_by_test_test_id('less_fake_test', key='a_key',
value='a_value')
self.assertEqual(1, len(res))
self.assertEqual(test_b.id, res[0].test_id)
self.assertEqual(run_a.id, res[0].run_id)
def test_get_all_run_metadata_keys(self):
run = api.create_run()
meta_dict = {
'test_a': 'a',
'test_a': 'b',
'test_b': 'a',
'test_c': 'a',
'test_d': 'a',
'test_c': 'b',
}
api.add_run_metadata(meta_dict, run.id)
keys = api.get_all_run_metadata_keys()
self.assertEqual(sorted(['test_a', 'test_b', 'test_c', 'test_d']),
sorted(keys))
def test_get_all_test_metadata_keys(self):
test = api.create_test('fake_test')
meta_dict = {
'test_a': 'a',
'test_a': 'b',
'test_b': 'a',
'test_c': 'a',
'test_d': 'a',
'test_c': 'b',
}
api.add_test_metadata(meta_dict, test.id)
keys = api.get_all_test_metadata_keys()
self.assertEqual(sorted(['test_a', 'test_b', 'test_c', 'test_d']),
sorted(keys))
def test_get_all_test_run_metadata_keys(self):
run = api.create_run()
test = api.create_test('fake_test')
test_run = api.create_test_run(test.id, run.id, 'skip')
meta_dict = {
'test_a': 'a',
'test_a': 'b',
'test_b': 'a',
'test_c': 'a',
'test_d': 'a',
'test_c': 'b',
}
api.add_test_run_metadata(meta_dict, test_run.id)
keys = api.get_all_test_run_metadata_keys()
self.assertEqual(sorted(['test_a', 'test_b', 'test_c', 'test_d']),
sorted(keys))
def test_get_test_run_series(self):
timestamp_a = datetime.datetime.utcnow().replace(microsecond=0)
timestamp_b = timestamp_a + datetime.timedelta(minutes=2)
api.create_run(passes=5, run_at=timestamp_a)
api.create_run(fails=2, run_at=timestamp_b)
result = api.get_test_run_series(key=None, value=None)
self.assertEqual(2, len(result.keys()))
self.assertIn(timestamp_a, result.keys())
self.assertIn(timestamp_b, result.keys())
for timestamp in result:
if timestamp.replace(
microsecond=0) == timestamp_a.replace(microsecond=0):
self.assertEqual(5, result[timestamp])
else:
self.assertEqual(2, result[timestamp])
def test_get_test_run_series_with_meta(self):
timestamp_a = datetime.datetime.utcnow().replace(microsecond=0)
timestamp_b = timestamp_a + datetime.timedelta(minutes=2)
run_a = api.create_run(passes=5, run_at=timestamp_a)
api.create_run(fails=2, run_at=timestamp_b)
api.add_run_metadata({'not_a_key': 'not_a_value'}, run_a.id)
result = api.get_test_run_series(key='not_a_key',
value='not_a_value')
self.assertEqual(1, len(result.keys()))
self.assertIn(timestamp_a, result.keys())
self.assertNotIn(timestamp_b, result.keys())
self.assertEqual(5, result[list(result.keys())[0]])
def test_get_run_times_grouped_by_run_metadata_key(self):
run_a = api.create_run(run_time=2.2, passes=2)
run_b = api.create_run(run_time=3.5, passes=3)
api.add_run_metadata({'key': 'value_a'}, run_a.id)
api.add_run_metadata({'key': 'value_b'}, run_b.id)
res = api.get_run_times_grouped_by_run_metadata_key('key')
expected_res = {'value_a': [2.2], 'value_b': [3.5]}
self.assertEqual(expected_res, res)
def test_get_run_times_grouped_by_run_metadata_key_with_match_filter(self):
run_a = api.create_run(run_time=2.2, passes=2)
run_b = api.create_run(run_time=3.5, passes=3)
run_c = api.create_run(run_time=75.432, passes=11)
api.add_run_metadata({'key': 'value_a'}, run_a.id)
api.add_run_metadata({'match_key': 'value_c'}, run_a.id)
api.add_run_metadata({'key': 'value_b'}, run_b.id)
api.add_run_metadata({'match_key': 'value_c'}, run_b.id)
api.add_run_metadata({'key': 'value_b'}, run_c.id)
res = api.get_run_times_grouped_by_run_metadata_key(
'key', match_key='match_key', match_value='value_c')
expected_res = {'value_a': [2.2], 'value_b': [3.5]}
self.assertEqual(expected_res, res)
def test_get_run_time_series_grouped_by_run_metadata_key(self):
run_at_a = datetime.datetime(1914, 11, 11, 11, 11, 11)
run_a = api.create_run(run_time=2.2, passes=2, run_at=run_at_a)
run_at_b = datetime.datetime.utcnow().replace(microsecond=0)
run_b = api.create_run(run_time=3.5, passes=3, run_at=run_at_b)
api.add_run_metadata({'key': 'value_a'}, run_a.id)
api.add_run_metadata({'key': 'value_b'}, run_b.id)
res = api.get_run_times_time_series_grouped_by_run_metadata_key('key')
expected_res = {run_at_a: {'value_a': 2.2},
run_at_b: {'value_b': 3.5}}
self.assertEqual(expected_res, res)
def test_get_run_ts_grouped_by_run_meta_key_with_match_filter(self):
run_at_a = datetime.datetime(1914, 11, 11, 11, 11, 11)
run_a = api.create_run(run_time=2.2, passes=2, run_at=run_at_a)
run_at_b = datetime.datetime.utcnow().replace(microsecond=0)
run_b = api.create_run(run_time=3.5, passes=3, run_at=run_at_b)
run_c = api.create_run(run_time=75.432, passes=11)
api.add_run_metadata({'key': 'value_a'}, run_a.id)
api.add_run_metadata({'match_key': 'value_c'}, run_a.id)
api.add_run_metadata({'key': 'value_b'}, run_b.id)
api.add_run_metadata({'match_key': 'value_c'}, run_b.id)
api.add_run_metadata({'key': 'value_b'}, run_c.id)
res = api.get_run_times_time_series_grouped_by_run_metadata_key(
'key', match_key='match_key', match_value='value_c')
expected_res = {run_at_a: {'value_a': 2.2},
run_at_b: {'value_b': 3.5}}
self.assertEqual(expected_res, res)
def test_get_test_run_dict_by_run_meta_key_value(self):
timestamp_a = datetime.datetime.utcnow().replace(microsecond=0)
timestamp_b = timestamp_a + datetime.timedelta(minutes=2)
run_a = api.create_run()
run_b = api.create_run()
api.add_run_metadata({'key': 'true'}, run_a.id)
api.add_run_metadata({'key': 'not so true'}, run_b.id)
test_a = api.create_test('fake_test')
api.create_test_run(test_a.id, run_a.id, 'success', timestamp_a,
timestamp_b)
api.create_test_run(test_a.id, run_b.id, 'fail', timestamp_a,
datetime.datetime.utcnow().replace(microsecond=0))
test_run_dicts = api.get_test_run_dict_by_run_meta_key_value('key',
'true')
self.assertEqual(1, len(test_run_dicts))
self.assertEqual([{
'test_id': 'fake_test',
'status': 'success',
'start_time': timestamp_a,
'stop_time': timestamp_b,
}], test_run_dicts)
def test_delete_old_runs(self):
run_a = api.create_run(run_at=datetime.datetime(
1914, 6, 28, 10, 45, 0))
run_b = api.create_run()
api.add_run_metadata({'key': 'value'}, run_b.id)
api.add_run_metadata({'key': 'not_so_much_a_value'}, run_a.id)
api.delete_old_runs()
runs = api.get_all_runs()
self.assertEqual(1, len(runs))
self.assertEqual(1, api.get_session().query(
models.RunMetadata.id).count())
self.assertEqual(run_b.id, runs[0].id)
self.assertEqual(1, len(api.get_run_metadata(run_b.uuid)))
self.assertEqual(0, len(api.get_run_metadata(run_a.uuid)))
def test_delete_old_test_runs(self):
run_a = api.create_run()
run_b = api.create_run()
test = api.create_test('fake_test')
test_run_a = api.create_test_run(test.id, run_a.id, 'fail',
start_time=datetime.datetime(
1914, 6, 28, 10, 45, 0))
test_run_b = api.create_test_run(test.id, run_b.id, 'fail',
start_time=datetime.datetime.utcnow())
api.add_test_run_metadata({'key': 'value'}, test_run_b.id)
api.add_test_run_metadata({'key': 'not_so_much_a_value'},
test_run_a.id)
api.delete_old_test_runs()
test_runs = api.get_all_test_runs()
self.assertEqual(1, len(test_runs))
self.assertEqual(test_run_b.id, test_runs[0].id)
self.assertEqual(1, len(api.get_test_run_metadata(test_run_b.id)))
self.assertEqual(0, len(api.get_test_run_metadata(test_run_a.id)))
def test_delete_run_by_uuid(self):
metadata_a = {'key': 'This is the metadata we\'re keeping'}
metadata_b = {'key': 'This metadata should be removed'}
run_a = api.create_run()
run_b = api.create_run()
uuid_to_delete = run_a.uuid
api.add_run_metadata(metadata_a, run_a.id)
api.add_run_metadata(metadata_b, run_b.id)
api.delete_run_by_uuid(uuid_to_delete)
runs = api.get_all_runs()
self.assertEqual(1, len(runs))
self.assertEqual(1, api.get_session().query(
models.RunMetadata.id).count())
self.assertEqual(run_b.id, runs[0].id)
self.assertEqual(1, len(api.get_run_metadata(run_b.uuid)))
self.assertEqual(0, len(api.get_run_metadata(run_a.uuid)))
def test_delete_test_runs_by_run_uuid(self):
# create runs
run_a = api.create_run()
run_b = api.create_run()
uuid_to_delete = run_a.uuid
# create test runs
test = api.create_test('fake_test')
test_run_a = api.create_test_run(test.id, run_a.id, 'fail',
start_time=datetime.datetime(
1914, 6, 28, 10, 45, 0))
test_run_b = api.create_test_run(test.id, run_b.id, 'fail',
start_time=datetime.datetime.utcnow())
# initialize and metadata to test_runs
metadata_a = {
'key': 'this test belongs to the run we are deleting from'}
metadata_b = {
'key': 'this test belongs to the run we are leaving alone'}
api.add_test_run_metadata(metadata_a, test_run_a.id)
api.add_test_run_metadata(metadata_b, test_run_b.id)
api.delete_test_runs_by_run_uuid(uuid_to_delete)
test_runs = api.get_all_test_runs()
self.assertEqual(1, len(test_runs))
self.assertEqual(test_run_b.id, test_runs[0].id)
self.assertEqual(1, len(api.get_test_run_metadata(test_run_b.id)))
self.assertEqual(0, len(api.get_test_run_metadata(test_run_a.id)))
def test_get_run_metadata(self):
run_a = api.create_run()
run_b = api.create_run()
api.add_run_metadata({'not_a_key': 'not_a_value'}, run_b.id)
a_metadata = api.get_run_metadata(run_a.uuid)
b_metadata = api.get_run_metadata(run_b.uuid)
self.assertEqual([], a_metadata)
self.assertEqual(1, len(b_metadata))
b_metadata = b_metadata[0].to_dict()
self.assertEqual(run_b.id, b_metadata['run_id'])
self.assertEqual('not_a_key', b_metadata['key'])
self.assertEqual('not_a_value', b_metadata['value'])
def test_get_runs_by_key_value(self):
api.create_run()
run_b = api.create_run()
api.add_run_metadata({'not_a_key': 'not_a_value'}, run_b.id)
found_runs = api.get_runs_by_key_value('not_a_key', 'not_a_value')
self.assertEqual(1, len(found_runs))
self.assertEqual(run_b.id, found_runs[0].id)
self.assertEqual(run_b.uuid, found_runs[0].uuid)
def test_get_tests_from_run_id(self):
run_a = api.create_run()
run_b = api.create_run()
test_a = api.create_test('fake_test')
test_b = api.create_test('fake_test2')
api.create_test_run(test_a.id, run_a.id, 'fail',
start_time=datetime.datetime(1914, 6, 28, 10, 45,
0))
api.create_test_run(test_a.id, run_b.id, 'fail',
start_time=datetime.datetime.utcnow())
api.create_test_run(test_b.id, run_a.id, 'success',
start_time=datetime.datetime(1914, 6, 28, 10, 45,
0))
result = api.get_tests_from_run_id(run_a.id)
self.assertEqual(2, len(result))
self.assertIn(test_a.id, [x.id for x in result])
self.assertIn(test_a.test_id, [x.test_id for x in result])
self.assertIn(test_b.id, [x.id for x in result])
self.assertIn(test_b.test_id, [x.test_id for x in result])
def test_get_all_runs_time_series_by_key(self):
time_a = datetime.datetime(1914, 6, 28, 10, 45, 0)
run_a = api.create_run(run_at=time_a)
run_b = api.create_run()
time_c = datetime.datetime(1918, 11, 11, 11, 11, 11)
run_c = api.create_run(run_at=time_c)
api.add_run_metadata({'not_a_key': 'not_a_value'}, run_b.id)
api.add_run_metadata({'a_key': 'a_value'}, run_a.id)
api.add_run_metadata({'a_key': 'c_value'}, run_c.id)
result = api.get_all_runs_time_series_by_key('a_key')
self.assertEqual(2, len(result.keys()))
self.assertIn(time_a.date(), [x.date() for x in result.keys()])
self.assertIn(time_c.date(), [x.date() for x in result.keys()])
def test_get_recent_runs_by_key_value_metadata(self):
run_a = api.create_run()
run_b = api.create_run()
run_c = api.create_run()
api.add_run_metadata({'a_key': 'a_value'}, run_a.id)
api.add_run_metadata({'a_key': 'a_value'}, run_c.id)
api.add_run_metadata({'a_key': 'b_value'}, run_b.id)
result = api.get_recent_runs_by_key_value_metadata('a_key', 'a_value')
self.assertEqual(2, len(result))
self.assertIn(run_a.id, [x.id for x in result])
self.assertNotIn(run_b.id, [x.id for x in result])
self.assertIn(run_c.id, [x.id for x in result])
def test_get_recent_runs_by_key_value_metadata_with_start_date(self):
run_a = api.create_run(run_at=datetime.datetime(
1914, 6, 28, 10, 45, 0))
run_b = api.create_run()
run_c = api.create_run()
api.add_run_metadata({'a_key': 'a_value'}, run_a.id)
api.add_run_metadata({'a_key': 'a_value'}, run_c.id)
api.add_run_metadata({'a_key': 'b_value'}, run_b.id)
result = api.get_recent_runs_by_key_value_metadata(
'a_key', 'a_value', start_date=datetime.datetime(
1918, 11, 11, 11, 11, 11))
self.assertEqual(1, len(result))
self.assertNotIn(run_a.id, [x.id for x in result])
self.assertNotIn(run_b.id, [x.id for x in result])
self.assertIn(run_c.id, [x.id for x in result])
def test_get_recent_runs_by_key_value_metadata_one_run(self):
timestamp = datetime.datetime(1914, 6, 28, 10, 45, 0)
run_a = api.create_run(run_at=timestamp)
run_b = api.create_run()
run_c = api.create_run()
api.add_run_metadata({'a_key': 'a_value'}, run_a.id)
api.add_run_metadata({'a_key': 'a_value'}, run_c.id)
api.add_run_metadata({'a_key': 'b_value'}, run_b.id)
result = api.get_recent_runs_by_key_value_metadata('a_key', 'a_value',
num_runs=1)
self.assertEqual(1, len(result))
self.assertNotIn(run_a.id, [x.id for x in result])
self.assertNotIn(run_b.id, [x.id for x in result])
self.assertIn(run_c.id, [x.id for x in result])
def test_get_test_runs_by_status_for_run_ids_no_meta(self):
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())
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())
result = api.get_test_runs_by_status_for_run_ids(
'fail', [run_a.uuid, run_b.uuid, run_c.uuid])
self.assertEqual(1, len(result))
self.assertEqual({
'test_id': u'fake_test',
'link': u'fake_url',
'start_time': start_timestamp,
'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()
run_c = api.create_run()
test_a = api.create_test('fake_test')
api.add_run_metadata({'a_key': 'b'}, run_b.id)
api.add_run_metadata({'a_key': 'a'}, run_a.id)
api.add_run_metadata({'a_key': 'c'}, run_c.id)
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())
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())
result = api.get_test_runs_by_status_for_run_ids(
'fail', [run_a.uuid, run_b.uuid, run_c.uuid], key='a_key')
self.assertEqual(1, len(result))
self.assertEqual({
'test_id': u'fake_test',
'link': u'fake_url',
'start_time': start_timestamp,
'stop_time': stop_timestamp,
'a_key': 'b',
}, result[0])
def test_get_test_runs_by_status_for_run_ids_with_run_id(self):
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')
api.add_run_metadata({'a_key': 'b'}, run_b.id)
api.add_run_metadata({'a_key': 'a'}, run_a.id)
api.add_run_metadata({'a_key': 'c'}, run_c.id)
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())
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())
result = api.get_test_runs_by_status_for_run_ids(
'fail', [run_a.uuid, run_b.uuid, run_c.uuid], key='a_key',
include_run_id=True)
self.assertEqual(1, len(result))
self.assertEqual({
'test_id': u'fake_test',
'link': u'fake_url',
'start_time': start_timestamp,
'stop_time': stop_timestamp,
'a_key': 'b',
'uuid': run_b.uuid,
}, result[0])
def test_get_all_runs_time_series_by_key_with_overlap(self):
time_a = datetime.datetime(1914, 6, 28, 10, 45, 0)
run_a = api.create_run(run_at=time_a)
run_b = api.create_run()
time_c = datetime.datetime(1918, 11, 11, 11, 11, 11)
run_c = api.create_run(run_at=time_c)
run_d = api.create_run(run_at=time_a)
api.add_run_metadata({'not_a_key': 'not_a_value'}, run_b.id)
api.add_run_metadata({'a_key': 'a_value'}, run_a.id)
api.add_run_metadata({'a_key': 'c_value'}, run_c.id)
api.add_run_metadata({'a_key': 'a_value'}, run_d.id)
result = api.get_all_runs_time_series_by_key('a_key')
self.assertEqual(2, len(result.keys()))
self.assertIn(time_a.date(), [x.date() for x in result.keys()])
self.assertIn(time_c.date(), [x.date() for x in result.keys()])
self.assertIn(time_a.date(), [x.date() for x in result.keys()])
self.assertEqual(len(result[time_a]['a_value']), 2)
def test_get_run_failure_rate_by_key_value_metadata(self):
run_a = api.create_run(fails=100, passes=0)
run_b = api.create_run()
run_c = api.create_run(passes=100, fails=0)
api.add_run_metadata({'a_key': 'a_value'}, run_a.id)
api.add_run_metadata({'a_key': 'a_value'}, run_c.id)
api.add_run_metadata({'a_key': 'b_value'}, run_b.id)
fail_rate = api.get_run_failure_rate_by_key_value_metadata(
'a_key', 'a_value')
self.assertEqual(50, fail_rate)
def test_get_test_prefixes(self):
api.create_test('prefix.token.token')
api.create_test('setUpClass (prefix.token.token)')
api.create_test('other.token.token')
api.create_test('justonetoken')
prefixes = api.get_test_prefixes()
self.assertEqual(len(prefixes), 3)
self.assertIn('prefix', prefixes)
self.assertIn('other', prefixes)
self.assertIn('justonetoken', prefixes)
def test_get_tests_by_prefix(self):
api.create_test('prefix.token.token')
api.create_test('setUpClass (prefix.token.token)')
api.create_test('other.token.token')
api.create_test('justonetoken')
tests = api.get_tests_by_prefix('prefix')
self.assertEqual(len(tests), 2)
self.assertIn('prefix.token.token', [test.test_id for test in tests])
self.assertIn('setUpClass (prefix.token.token)',
[test.test_id for test in tests])
tests = api.get_tests_by_prefix('other')
self.assertEqual(len(tests), 1)
self.assertIn('other.token.token', [test.test_id for test in tests])
tests = api.get_tests_by_prefix('prefix', limit=1, offset=1)
self.assertEqual(len(tests), 1)
self.assertIn('setUpClass (prefix.token.token)',
[test.test_id for test in tests])
tests = api.get_tests_by_prefix('justonetoken')
self.assertEqual(len(tests), 1)
self.assertIn('justonetoken', [test.test_id for test in tests])
def test_update_test(self):
create_res = api.create_test('fake_test')
values = {'run_count': 2, 'run_time': 1.2}
update_res = api.update_test(values, create_res.id)
res = api.get_test_by_test_id('fake_test')
self.assertEqual(res.id, update_res.id)
self.assertEqual(res.test_id, 'fake_test')
self.assertEqual(res.run_time, 1.2)
self.assertEqual(res.run_count, 2)
def test_update_test_run(self):
run = api.create_run()
test = api.create_test('fake_test')
test_run = api.create_test_run(test.id, run.id, 'fail')
start_time = datetime.datetime.utcnow()
stop_time = datetime.datetime.utcnow()
start_time = start_time.replace(microsecond=0)
stop_time = start_time.replace(microsecond=0)
values = {'status': 'success', 'start_time': start_time,
'stop_time': stop_time}
update_test_run = api.update_test_run(values, test_run.id)
all_test_runs = api.get_all_test_runs()
self.assertEqual(len(all_test_runs), 1)
self.assertEqual(update_test_run.id, all_test_runs[0].id)
self.assertEqual(all_test_runs[0].status, 'success')
self.assertEqual(all_test_runs[0].start_time, start_time)
self.assertEqual(all_test_runs[0].stop_time, stop_time)
def test_get_test_run_by_id(self):
run = api.create_run()
test = api.create_test('fake_test')
test_run = api.create_test_run(test.id, run.id, 'fail')
res = api.get_test_run_by_id(test_run.id)
self.assertEqual(res.id, test_run.id)
self.assertEqual(res.status, 'fail')
self.assertEqual(res.run_id, run.id)
def test_get_test_by_id(self):
create_res = api.create_test('fake_test', 2, 1, 1, 1.2)
res = api.get_test_by_id(create_res.id)
self.assertEqual(res.id, create_res.id)
self.assertEqual(res.test_id, 'fake_test')
self.assertEqual(res.run_time, 1.2)
self.assertEqual(res.run_count, 2)
def test_get_run_id_from_uuid(self):
run = api.create_run()
run_id = api.get_run_id_from_uuid(run.uuid)
self.assertEqual(run_id, run.id)
def test_get_run_by_id(self):
run = api.create_run()
res = api.get_run_by_id(run.id)
self.assertEqual(res.id, run.id)
self.assertEqual(res.uuid, run.uuid)
def test_get_runs_by_ids(self):
run_a = api.create_run()
run_b = api.create_run()
res = api.get_runs_by_ids([run_a.id, run_b.id])
self.assertIn(run_a.id, [x.id for x in res])
self.assertIn(run_b.id, [x.id for x in res])
self.assertIn(run_a.uuid, [x.uuid for x in res])
self.assertIn(run_b.uuid, [x.uuid for x in res])
def test_get_test_runs_by_test_id(self):
run_a = api.create_run()
run_b = api.create_run()
run_c = api.create_run()
test = api.create_test('fake_test')
api.create_test_run(test.id, run_a.id, 'success')
api.create_test_run(test.id, run_b.id, 'fail')
api.create_test_run(test.id, run_c.id, 'success')
res = api.get_test_runs_by_test_id(test.id)
self.assertEqual(3, len(res))
self.assertIn(run_a.id, [x.run_id for x in res])
self.assertIn(run_b.id, [x.run_id for x in res])
self.assertIn(run_c.id, [x.run_id for x in res])
def test_get_test_metadata(self):
test = api.create_test('fake_test')
test_meta = {
'test_a': 'a',
'test_b': 'b',
'test_c': 'c',
}
api.add_test_metadata(test_meta, test.id)
test_metadata = api.get_test_metadata(test.id)
self.assertEqual(3, len(test_metadata))
for meta in test_metadata:
self.assertIn(meta.key, test_meta.keys())
self.assertIn(meta.value, test_meta.values())
def test_get_recent_failed_runs_by_run_metadata_no_start_date(self):
run_a = api.create_run(fails=1)
api.create_run()
run_c = api.create_run(fails=2)
api.add_run_metadata({'fake_key': 'fake_value'}, run_a.id)
api.add_run_metadata({'zeon': 'zaku'}, run_c.id)
results = api.get_recent_failed_runs_by_run_metadata('zeon', 'zaku')
self.assertEqual(1, len(results))
self.assertEqual(run_c.id, results[0].id)
def test_get_recent_failed_runs_by_run_metadata_with_start_date(self):
run_a = api.create_run(fails=1)
api.create_run()
run_c = api.create_run(fails=2)
run_d = api.create_run(fails=3,
run_at=datetime.datetime(1914, 6, 28,
10, 45, 0))
api.add_run_metadata({'fake_key': 'fake_value'}, run_a.id)
api.add_run_metadata({'zeon': 'zaku'}, run_c.id)
api.add_run_metadata({'zeon': 'zaku'}, run_d.id)
results = api.get_recent_failed_runs_by_run_metadata(
'zeon', 'zaku', start_date=datetime.date(1970, 1, 1))
self.assertEqual(1, len(results))
self.assertEqual(run_c.id, results[0].id)
def test_get_runs_count(self):
for i in range(0, 100):
api.create_run()
count = api.get_runs_count()
self.assertEqual(100, count)
def test_get_runs_count_zero(self):
self.assertEqual(0, api.get_runs_count())
def test_get_runs_counts_by_run_metadata_no_start_date(self):
run_a = api.create_run(fails=1)
api.create_run()
run_c = api.create_run(fails=2)
api.add_run_metadata({'fake_key': 'fake_value'}, run_a.id)
api.add_run_metadata({'zeon': 'zaku'}, run_c.id)
results = api.get_runs_counts_by_run_metadata('zeon', 'zaku')
self.assertEqual(1, results)
results = api.get_runs_counts_by_run_metadata('zeon', 'dom')
self.assertEqual(0, results)
def test_get_runs_counts_by_run_metadata_with_start_date(self):
run_a = api.create_run(fails=1)
api.create_run()
run_c = api.create_run(fails=2)
run_d = api.create_run(fails=3,
run_at=datetime.datetime(1914, 6, 28,
10, 45, 0))
api.add_run_metadata({'fake_key': 'fake_value'}, run_a.id)
api.add_run_metadata({'zeon': 'zaku'}, run_c.id)
api.add_run_metadata({'zeon': 'zaku'}, run_d.id)
results = api.get_runs_counts_by_run_metadata(
'zeon', 'zaku', start_date=datetime.date(1970, 1, 1))
self.assertEqual(1, results)
results = api.get_runs_counts_by_run_metadata(
'zeon', 'zaku', start_date=datetime.date(1914, 6, 27))
self.assertEqual(2, results)
results = api.get_runs_counts_by_run_metadata(
'zeon', 'dom', start_date=datetime.date(1914, 6, 27))
self.assertEqual(0, results)
def test_get_test_run_metadata(self):
run = api.create_run()
test = api.create_test('fake_test')
test_run = api.create_test_run(test.id, run.id, 'success',
datetime.datetime.utcnow(),
datetime.datetime.utcnow())
run_meta = {
'key_a': 'value_b',
'key_b': 'value_a',
}
api.add_test_run_metadata(run_meta, test_run.id)
test_run_metadata = api.get_test_run_metadata(test_run.id)
self.assertEqual(2, len(test_run_metadata))
for meta in test_run_metadata:
self.assertIn(meta.key, run_meta.keys())
self.assertIn(meta.value, run_meta.values())
def test_get_all_runs_by_date(self):
timestamp_a = datetime.datetime.utcnow().replace(microsecond=0)
timestamp_b = timestamp_a + datetime.timedelta(minutes=10)
timestamp_c = timestamp_a + datetime.timedelta(minutes=20)
api.create_run(run_at=timestamp_a)
api.create_run(run_at=timestamp_b)
api.create_run(run_at=timestamp_c)
res = api.get_all_runs_by_date()
self.assertEqual(3, len(res))
self.assertIn(timestamp_a, [x.run_at for x in res])
self.assertIn(timestamp_b, [x.run_at for x in res])
self.assertIn(timestamp_c, [x.run_at for x in res])
res = api.get_all_runs_by_date(start_date=timestamp_c)
self.assertEqual(1, len(res))
self.assertEqual(timestamp_c, res[0].run_at)
res = api.get_all_runs_by_date(stop_date=timestamp_a)
self.assertEqual(1, len(res))
self.assertEqual(timestamp_a, res[0].run_at)
res = api.get_all_runs_by_date(start_date=timestamp_a,
stop_date=timestamp_b)
self.assertEqual(2, len(res))
self.assertIn(timestamp_a, [x.run_at for x in res])
self.assertIn(timestamp_b, [x.run_at for x in res])
def test_get_latest_run(self):
timestamp_a = datetime.datetime.utcnow().replace(microsecond=0)
timestamp_b = timestamp_a + datetime.timedelta(minutes=10)
timestamp_c = timestamp_a + datetime.timedelta(minutes=20)
api.create_run(run_at=timestamp_a)
api.create_run(run_at=timestamp_b)
api.create_run(run_at=timestamp_c)
res = api.get_latest_run()
self.assertEqual(timestamp_c, res.run_at)
def test_get_failing_from_run(self):
timestamp_a = datetime.datetime.utcnow().replace(microsecond=0)
timestamp_b = timestamp_a + datetime.timedelta(minutes=10)
timestamp_c = timestamp_a + datetime.timedelta(minutes=20)
run_a = api.create_run()
run_b = api.create_run()
test_a = api.create_test('fake_test')
test_b = api.create_test('fake_test2')
api.create_test_run(test_a.id, run_a.id, 'fail',
start_time=timestamp_a)
api.create_test_run(test_a.id, run_b.id, 'fail',
start_time=timestamp_a)
api.create_test_run(test_b.id, run_b.id, 'fail',
start_time=timestamp_b)
api.create_test_run(test_b.id, run_a.id, 'success',
start_time=timestamp_c)
res = api.get_failing_from_run(run_a.id)
self.assertEqual(1, len(res))
self.assertEqual(test_a.id, res[0].test_id)
self.assertEqual(timestamp_a, res[0].start_time)
res = api.get_failing_from_run(run_b.id)
self.assertEqual(2, len(res))
self.assertEqual('fail', res[0].status)
self.assertIn(timestamp_a, [x.start_time for x in res])
self.assertIn(timestamp_b, [x.start_time for x in res])
def test_get_test_run_time_series(self):
timestamp_a = datetime.datetime.utcnow().replace(microsecond=0)
timestamp_b = timestamp_a + datetime.timedelta(minutes=10)
timestamp_c = timestamp_a + datetime.timedelta(minutes=15)
run_a = api.create_run()
run_b = api.create_run()
test = api.create_test('fake_test')
api.create_test_run(test.id, run_a.id, 'success',
start_time=timestamp_a,
end_time=timestamp_b)
api.create_test_run(test.id, run_b.id, 'success',
start_time=timestamp_b,
end_time=timestamp_c)
res = api.get_test_run_time_series(test.id)
self.assertEqual(res[timestamp_a], 600)
self.assertEqual(res[timestamp_b], 300)
def test_get_test_status_time_series(self):
timestamp_a = datetime.datetime.utcnow().replace(microsecond=0)
timestamp_b = timestamp_a + datetime.timedelta(minutes=10)
timestamp_c = timestamp_a + datetime.timedelta(minutes=20)
run_a = api.create_run()
run_b = api.create_run()
test = api.create_test('fake_test')
api.create_test_run(test.id, run_a.id, 'success',
start_time=timestamp_a,
end_time=timestamp_b)
api.create_test_run(test.id, run_b.id, 'fail',
start_time=timestamp_b,
end_time=timestamp_c)
res = api.get_test_status_time_series(test.id)
self.assertEqual(res[timestamp_a], 'success')
self.assertEqual(res[timestamp_b], 'fail')
def test_get_recent_failed_runs(self):
run_a = api.create_run(fails=1)
api.create_run()
run_c = api.create_run(fails=2)
run_d = api.create_run(fails=1)
res = api.get_recent_failed_runs()
self.assertEqual(3, len(res))
self.assertIn(run_a.uuid, res)
self.assertIn(run_c.uuid, res)
self.assertIn(run_d.uuid, res)
def test_get_recent_failed_runs_with_start_date(self):
api.create_run(fails=1, run_at=datetime.datetime(
1914, 6, 28, 10, 45, 0))
api.create_run()
run_c = api.create_run(fails=2)
run_d = api.create_run(fails=1)
res = api.get_recent_failed_runs(start_date=datetime.datetime(
1918, 11, 11, 11, 11, 11))
self.assertEqual(2, len(res))
self.assertIn(run_c.uuid, res)
self.assertIn(run_d.uuid, res)
def test_get_recent_successful_runs(self):
run_a = api.create_run(passes=1)
run_b = api.create_run()
run_c = api.create_run(passes=2)
api.create_run(fails=1)
res = api.get_recent_successful_runs()
self.assertEqual(3, len(res))
self.assertIn(run_a.uuid, res)
self.assertIn(run_b.uuid, res)
self.assertIn(run_c.uuid, res)
def test_get_recent_successful_runs_with_start_date(self):
api.create_run(passes=1, run_at=datetime.datetime(
1914, 6, 28, 10, 45, 0))
run_b = api.create_run()
run_c = api.create_run(passes=2)
api.create_run(fails=1)
res = api.get_recent_successful_runs(start_date=datetime.datetime(
1918, 11, 11, 11, 11, 11))
self.assertEqual(2, len(res))
self.assertIn(run_b.uuid, res)
self.assertIn(run_c.uuid, res)
def test_get_test_counts_in_date_range_as_str(self):
timestamp_str_a = 'Dec 01 2015'
timestamp_str_b = 'Dec 20 2015'
timestamp_a = datetime.datetime(2015, 12, 2, 10, 00, 00)
timestamp_b = timestamp_a + datetime.timedelta(minutes=10)
timestamp_c = timestamp_a + datetime.timedelta(minutes=20)
timestamp_d = datetime.datetime(2015, 12, 22, 10, 00, 00)
run_a = api.create_run()
run_b = api.create_run()
run_c = api.create_run()
run_d = api.create_run()
test = api.create_test('fake_test')
api.create_test_run(test.id, run_a.id, 'success',
timestamp_a, timestamp_b)
api.create_test_run(test.id, run_b.id, 'fail',
timestamp_a, timestamp_c)
api.create_test_run(test.id, run_c.id, 'success',
timestamp_a, timestamp_d)
api.create_test_run(test.id, run_d.id, 'skip',
timestamp_c, timestamp_d)
res = api.get_test_counts_in_date_range(test.id,
timestamp_str_a,
timestamp_str_b)
self.assertEqual(1, res['success'])
self.assertEqual(1, res['failure'])
self.assertEqual(0, res['skips'])
def test_get_test_counts_in_date_range(self):
timestamp_a = datetime.datetime(2015, 12, 2, 10, 00, 00)
timestamp_b = timestamp_a + datetime.timedelta(minutes=10)
timestamp_c = timestamp_a + datetime.timedelta(minutes=20)
timestamp_d = datetime.datetime(2015, 12, 22, 10, 00, 00)
timerange_a = datetime.datetime(2015, 12, 1)
timerange_b = datetime.datetime(2015, 12, 20)
run_a = api.create_run()
run_b = api.create_run()
run_c = api.create_run()
run_d = api.create_run()
test = api.create_test('fake_test')
api.create_test_run(test.id, run_a.id, 'success',
timestamp_a, timestamp_b)
api.create_test_run(test.id, run_b.id, 'fail',
timestamp_a, timestamp_c)
api.create_test_run(test.id, run_c.id, 'success',
timestamp_a, timestamp_d)
api.create_test_run(test.id, run_d.id, 'skip',
timestamp_c, timestamp_d)
res = api.get_test_counts_in_date_range(test.id,
timerange_a,
timerange_b)
self.assertEqual(1, res['success'])
self.assertEqual(1, res['failure'])
self.assertEqual(0, res['skips'])
def test_get_failing_test_ids_from_runs_by_key_value(self):
test_a = api.create_test('fake_test')
test_b = api.create_test('fake_test1')
test_c = api.create_test('fake_test2')
test_d = api.create_test('fake_test3')
run_a = api.create_run()
run_b = api.create_run()
run_c = api.create_run()
api.add_run_metadata({'a_key': 'a_value'}, run_a.id)
api.add_run_metadata({'a_key': 'a_value'}, run_b.id)
api.add_run_metadata({'a_key': 'b_value'}, run_c.id)
api.create_test_run(test_a.id, run_a.id, 'fail')
api.create_test_run(test_b.id, run_b.id, 'success')
api.create_test_run(test_c.id, run_b.id, 'fail')
api.create_test_run(test_d.id, run_c.id, 'fail')
res = api.get_failing_test_ids_from_runs_by_key_value('a_key',
'a_value')
self.assertEqual(2, len(res))
self.assertIn(test_a.test_id, res)
self.assertIn(test_c.test_id, res)
def test_add_test_run_attachments(self):
test = api.create_test('fake_test')
run = api.create_run()
test_run = api.create_test_run(test.id, run.id, 'success')
attach_dict = {'attach_label': b'attach',
'attach_label_a': b'attach_a'}
res = api.add_test_run_attachments(attach_dict, test_run.id)
self.assertEqual(2, len(res))
self.assertEqual(test_run.id, res[0].test_run_id)
self.assertIn('attach_label', [x.label for x in res])
self.assertIn('attach_label_a', [x.label for x in res])
self.assertIn(b'attach', [x.attachment for x in res])
self.assertIn(b'attach_a', [x.attachment for x in res])
def test_get_ids_for_all_tests(self):
test_a = api.create_test('fake_test')
test_b = api.create_test('fake_test1')
test_c = api.create_test('fake_test2')
res = api.get_ids_for_all_tests()
self.assertIsInstance(res, types.GeneratorType)
res = list(res)
self.assertIn((test_c.id,), res)
self.assertIn((test_b.id,), res)
self.assertIn((test_a.id,), res)
def test_get_recent_successful_runs_by_run_metadata_no_start_date(self):
run_a = api.create_run(fails=0, passes=1)
api.create_run()
run_c = api.create_run(fails=0, passes=2)
api.add_run_metadata({'fake_key': 'fake_value'}, run_a.id)
api.add_run_metadata({'zeon': 'zaku'}, run_c.id)
results = api.get_recent_successful_runs_by_run_metadata('zeon',
'zaku')
self.assertEqual(1, len(results))
self.assertEqual(run_c.id, results[0].id)
def test_get_recent_successful_runs_by_run_metadata_with_start_date(self):
run_a = api.create_run(fails=0, passes=1)
api.create_run()
run_c = api.create_run(fails=0, passes=2)
run_d = api.create_run(fails=0, passes=3,
run_at=datetime.datetime(1914, 6, 28,
10, 45, 0))
api.add_run_metadata({'fake_key': 'fake_value'}, run_a.id)
api.add_run_metadata({'zeon': 'zaku'}, run_c.id)
api.add_run_metadata({'zeon': 'zaku'}, run_d.id)
results = api.get_recent_successful_runs_by_run_metadata(
'zeon', 'zaku', start_date=datetime.date(1970, 1, 1))
self.assertEqual(1, len(results))
self.assertEqual(run_c.id, results[0].id)