Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _update_test(self, test_dict, session, start_time, stop_time):
test_id = utils.cleanup_test_name(test_dict['id'])
db_test = db_api.get_test_by_test_id(test_id, session)
if not db_test:
if test_dict['status'] == 'success':
success = 1
fails = 0
elif test_dict['status'] == 'fail':
fails = 1
success = 0
else:
fails = 0
success = 0
run_time = read_subunit.get_duration(start_time, stop_time)
db_test = db_api.create_test(test_id, (success + fails), success,
fails, run_time,
session)
else:
def _handle_test(self, test_dict):
start, stop = test_dict['timestamps']
if test_dict['status'] == 'exists' or None in (start, stop):
return
test_id = utils.cleanup_test_name(test_dict['id'])
self._times[test_id] = str((stop - start).total_seconds())
def _get_test_times(self, test_ids):
result = {}
# TODO(mtreinish): after subunit2sql adds a bulk query for getting
# multiple tests by test_id at once remove the for loop
session = self.session_factory()
for test_id in test_ids:
stripped_test_id = utils.cleanup_test_name(test_id)
test = db_api.get_test_by_test_id(stripped_test_id,
session=session)
if test:
# NOTE(mtreinish): We need to make sure the test_id with attrs
# is used in the output dict, otherwise the scheduler won't
# see it
result[test_id] = test.run_time
session.close()
return result
def _get_test_times(self, test_ids):
# May be too slow, but build and iterate.
# 'c' because an existing repo may be missing a file.
try:
db = my_dbm.open(self._path('times.dbm'), 'c')
except my_dbm.error:
os.remove(self._path('times.dbm'))
db = my_dbm.open(self._path('times.dbm'), 'c')
try:
result = {}
for test_id in test_ids:
if type(test_id) != str:
test_id = test_id.encode('utf8')
stripped_test_id = utils.cleanup_test_name(test_id)
# gdbm does not support get().
try:
duration = db[stripped_test_id]
except KeyError:
duration = None
if duration is not None:
result[test_id] = float(duration)
return result
finally:
db.close()