#!/usr/bin/python
|
|
"""Unittests for the JobSerializer class.
|
|
Mostly test if the serialized object has the expected content.
|
|
"""
|
|
import datetime
|
import tempfile
|
import time
|
import unittest
|
|
import common
|
from autotest_lib.tko import tko_pb2
|
from autotest_lib.tko import job_serializer
|
from autotest_lib.tko import models
|
|
NamedTemporaryFile = tempfile.NamedTemporaryFile
|
datetime = datetime.datetime
|
mktime = time.mktime
|
|
class JobSerializerUnittest(unittest.TestCase):
|
"""Base class as a job serializer unittest"""
|
|
def setUp(self):
|
tko_patches = []
|
tko_patches.append(models.patch('New spec!', 'Reference?',
|
123456))
|
|
tko_kernel = models.kernel('My Computer', tko_patches, '1234567')
|
tko_time = datetime.now()
|
|
tko_job = models.job('/tmp/', 'autotest', 'test', 'My Computer',
|
tko_time, tko_time, tko_time, 'root',
|
'www', 'No one', tko_time, {'1+1':2})
|
tko_job.afe_parent_job_id = 111
|
tko_job.build_version = 'R1-1.0.0'
|
tko_job.suite = 'bvt'
|
tko_job.board = 'alex'
|
tko_job.index = 2
|
|
tko_iteration = models.iteration(0, {'2+2':4, '3+3':6},
|
{'4+4':8, '5+5':10, '6+6':12})
|
|
tko_labels = ['unittest', 'dummy test', 'autotest']
|
|
# See the comment about the models.test constructor in
|
# job_serializer.py, where the models.test constructor is used.
|
tko_test = models.test('/tmp/', 'mocktest', 'Completed', 'N/A',
|
tko_kernel, 'My Computer', tko_time,
|
tko_time, [tko_iteration,
|
tko_iteration, tko_iteration],
|
{'abc':'def'}, [], tko_labels)
|
tko_test.test_idx = 3
|
self.tko_job = tko_job
|
self.tko_job.tests = [tko_test, tko_test, tko_test]
|
|
self.pb_job = tko_pb2.Job()
|
self.tag = '1-abc./.'
|
self.expected_afe_job_id = '1'
|
|
js = job_serializer.JobSerializer()
|
js.set_pb_job(self.tko_job, self.pb_job, self.tag)
|
|
|
def test_tag(self):
|
"""Test serializing tag field."""
|
self.assertEqual(self.tag, self.pb_job.tag)
|
|
|
def test_afe_job_id(self):
|
"""Test serializing afe_job_id field."""
|
self.assertEqual(self.expected_afe_job_id,
|
self.pb_job.afe_job_id)
|
|
|
def test_job_dir(self):
|
"""Check if the dir field are the same.
|
"""
|
self.assertEqual(self.tko_job.dir, self.pb_job.dir)
|
|
|
def test_number_of_test(self):
|
"""Check if the number of test are the same.
|
"""
|
self.assertEqual(len(self.tko_job.tests),
|
len(self.pb_job.tests))
|
|
|
def test_user(self):
|
"""Check if the user field are the same.
|
"""
|
self.assertEqual(self.tko_job.user, self.pb_job.user)
|
|
|
def test_machine(self):
|
"""Check if the machine fields are the same.
|
"""
|
self.assertEqual(self.tko_job.machine, self.pb_job.machine)
|
|
|
def test_queued_time(self):
|
"""Check if queued_time are the same.
|
"""
|
self.check_time(self.tko_job.queued_time,
|
self.pb_job.queued_time)
|
|
|
def test_started_time(self):
|
"""Check if the started_time are the same.
|
"""
|
self.check_time(self.tko_job.started_time,
|
self.pb_job.started_time)
|
|
|
def test_finished_time(self):
|
"""Check if the finished_time are the same.
|
"""
|
self.check_time(self.tko_job.finished_time,
|
self.pb_job.finished_time)
|
|
|
def test_machine_owner(self):
|
"""Check if the machine owners are the same.
|
"""
|
self.assertEqual(self.tko_job.machine_owner,
|
self.pb_job.machine_owner)
|
|
|
def test_machine_group(self):
|
"""Check if the machine groups are the same.
|
"""
|
self.assertEqual(self.tko_job.machine_group,
|
self.pb_job.machine_group)
|
|
def test_aborted_by(self):
|
"""Check if the jobs are aborted by the same person.
|
"""
|
self.assertEqual(self.tko_job.aborted_by,
|
self.pb_job.aborted_by)
|
|
|
def test_aborted_on(self):
|
"""Test serializing aborted_on field."""
|
self.check_time(self.tko_job.aborted_on,
|
self.pb_job.aborted_on)
|
|
|
def test_keyval_dict(self):
|
"""Check if the contents of the dictionary are the same.
|
"""
|
self.assertEqual(len(self.tko_job.keyval_dict),
|
len(self.pb_job.keyval_dict))
|
self.check_dict(self.tko_job.keyval_dict,
|
self.convert_keyval_to_dict(self.pb_job,
|
'keyval_dict'))
|
|
|
def test_job_idx(self):
|
"""Test serializing job_idx field."""
|
self.assertEqual(self.tko_job.index,
|
self.pb_job.job_idx)
|
|
|
def test_afe_parent_job_id(self):
|
"""Test serializing afe_parent_job_id field."""
|
self.assertEqual(str(self.tko_job.afe_parent_job_id),
|
self.pb_job.afe_parent_job_id)
|
|
|
def test_build_version(self):
|
"""Test serializing build_version field."""
|
self.assertEqual(self.tko_job.build_version,
|
self.pb_job.build_version)
|
|
|
def test_suite(self):
|
"""Test serializing suite field."""
|
self.assertEqual(self.tko_job.suite,
|
self.pb_job.suite)
|
|
|
def test_board(self):
|
"""Test serializing board field."""
|
self.assertEqual(self.tko_job.board,
|
self.pb_job.board)
|
|
|
def test_tests(self):
|
"""Check if all the test are the same.
|
"""
|
|
for test, newtest in zip(self.tko_job.tests,
|
self.pb_job.tests):
|
|
self.assertEqual(test.subdir, newtest.subdir)
|
self.assertEqual(test.testname, newtest.testname)
|
self.assertEqual(test.status, newtest.status)
|
self.assertEqual(test.reason, newtest.reason)
|
self.assertEqual(test.machine, newtest.machine)
|
self.assertEqual(test.labels, newtest.labels)
|
self.assertEqual(test.test_idx, newtest.test_idx)
|
|
self.check_time(test.started_time, newtest.started_time)
|
self.check_time(test.finished_time, newtest.finished_time)
|
|
self.check_iteration(test.iterations, newtest.iterations)
|
|
self.check_dict(test.attributes,
|
self.convert_keyval_to_dict(newtest,
|
'attributes'))
|
|
self.check_kernel(test.kernel, newtest.kernel)
|
|
|
def check_time(self, dTime, stime):
|
"""Check if the datetime object contains the same time value
|
in microseconds.
|
|
@param dTime: The datetime.
|
@param stime: The original time.
|
"""
|
t = mktime(dTime.timetuple()) + 1e-6 * dTime.microsecond
|
self.assertEqual(long(t), stime/1000)
|
|
|
def check_iteration(self, tko_iterations, pb_iterations):
|
"""Check if the iteration objects are the same.
|
|
@param tko_iterations: The list of iterations.
|
@param pb_iterations: The proto iterations.
|
"""
|
for tko_iteration, pb_iteration in zip(tko_iterations,
|
pb_iterations):
|
|
self.assertEqual(tko_iteration.index, pb_iteration.index)
|
|
self.check_dict(tko_iteration.attr_keyval,
|
self.convert_keyval_to_dict(pb_iteration,
|
'attr_keyval'))
|
|
self.check_dict(tko_iteration.perf_keyval,
|
self.convert_keyval_to_dict(pb_iteration,
|
'perf_keyval'))
|
|
|
def convert_keyval_to_dict(self, var, attr):
|
"""Convert a protocol buffer repeated keyval object into a
|
python dict.
|
|
@param var: The variable name.
|
@param attr: The attribute name.
|
"""
|
|
return dict((keyval.name, keyval.value) for keyval in
|
getattr(var,attr))
|
|
|
def check_dict(self, dictionary, keyval):
|
"""Check if the contents of the dictionary are the same as a
|
repeated keyval pair.
|
|
@param dictionary: The dict object.
|
@param keyval: The keyval object.
|
"""
|
for key, value in dictionary.iteritems():
|
self.assertTrue(key in keyval);
|
self.assertEqual(str(value), keyval[key])
|
|
|
def check_kernel(self, kernel, newkernel):
|
"""Check if the kernels are the same.
|
|
@param kernel: The kernel object.
|
@param newkernel: The proto kernel object.
|
"""
|
self.assertEqual(kernel.base, newkernel.base)
|
self.assertEqual(kernel.kernel_hash, newkernel.kernel_hash)
|
|
|
class ReadBackTest(JobSerializerUnittest):
|
"""Check if convert between models.job and pb job is correct even
|
after being written to binary and read by manually
|
"""
|
|
def setUp(self):
|
"""Setup the test."""
|
super(ReadBackTest, self).setUp()
|
|
out_binary = NamedTemporaryFile(mode='wb')
|
try:
|
out_binary.write(self.pb_job.SerializeToString())
|
out_binary.flush()
|
|
binary = open(out_binary.name, 'rb')
|
try:
|
self.pb_job = tko_pb2.Job()
|
self.pb_job.ParseFromString(binary.read())
|
finally:
|
binary.close()
|
finally:
|
out_binary.close()
|
|
|
class ReadBackGetterTest(JobSerializerUnittest):
|
"""Check if convert between models.job and pb job is correct after
|
using the getter methods in JobSerializer to read back the
|
data.
|
"""
|
|
def setUp(self):
|
super(ReadBackGetterTest, self).setUp()
|
|
temp_binary = NamedTemporaryFile(mode='wb')
|
try:
|
temp_binary.write(self.pb_job.SerializeToString())
|
temp_binary.flush()
|
|
js = job_serializer.JobSerializer()
|
self.from_pb_job = js.deserialize_from_binary(temp_binary.name)
|
finally:
|
temp_binary.close()
|
|
|
def test_keyval_dict(self):
|
"""Check if the contents of the dictionary are the same. """
|
|
self.assertEqual(len(self.tko_job.keyval_dict),
|
len(self.from_pb_job.keyval_dict))
|
|
self.check_dict(self.tko_job.keyval_dict,
|
self.from_pb_job.keyval_dict)
|
|
|
def test_tests(self):
|
"""Check if all the test are the same.
|
"""
|
for test, newtest in zip(self.tko_job.tests,
|
self.from_pb_job.tests):
|
|
self.assertEqual(test.subdir, newtest.subdir)
|
self.assertEqual(test.testname, newtest.testname)
|
self.assertEqual(test.status, newtest.status)
|
self.assertEqual(test.reason, newtest.reason)
|
self.assertEqual(test.machine, newtest.machine)
|
self.assertEqual(test.labels, newtest.labels)
|
|
self.check_time(test.started_time, newtest.started_time)
|
self.check_time(test.finished_time, newtest.finished_time)
|
|
self.check_iteration(test.iterations, newtest.iterations)
|
|
self.check_dict(test.attributes, newtest.attributes)
|
|
self.check_kernel(test.kernel, newtest.kernel)
|
|
|
def check_time(self, dTime, sTime):
|
"""Check if the datetime object contains the same time value
|
in microseconds.
|
|
If sTime is type int or type long, then only convert dTime to
|
microseconds. Else, convert both dTime and sTime to
|
microseconds. Then, compare the two after casting them to
|
long.
|
"""
|
|
t = mktime(dTime.timetuple()) + 1e-6 * dTime.microsecond
|
if isinstance(sTime, (int, long)):
|
self.assertEqual(long(t*1000), sTime)
|
else:
|
t1 = mktime(sTime.timetuple()) + 1e-6 * sTime.microsecond
|
self.assertEqual(long(t*1000), long(t1*1000))
|
|
|
def check_iteration(self, iterations, newiterations):
|
"""Check if the iteration objects are the same.
|
"""
|
for iteration, newiteration in zip(iterations, newiterations):
|
self.assertEqual(iteration.index, newiteration.index)
|
self.check_dict(iteration.attr_keyval,
|
newiteration.attr_keyval)
|
self.check_dict(iteration.perf_keyval,
|
newiteration.perf_keyval)
|
|
|
if __name__ == '__main__':
|
unittest.main()
|
