#!/usr/bin/python
"""Unittests for the JobSerializer class.
Mostly test if the serialized object has the expected content.
"""
import common
import datetime
import os
import re
import tempfile
import time
from autotest_lib.tko import tko_pb2
from autotest_lib.tko import job_serializer
from autotest_lib.tko import models
from autotest_lib.client.common_lib.test_utils import unittest
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_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)
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):
self.assertEqual(self.tag, self.pb_job.tag)
def test_afe_job_id(self):
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):
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_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.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.
"""
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.
"""
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.
"""
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.
"""
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.
"""
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):
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()