path: root/tko/job_serializer_unittest.py

#!/usr/bin/python

"""Unittests for the JobSerializer class.

Mostly test if the serialized object has the expected content.

"""

import common
import time
import os
import datetime
import tempfile

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('tubes', tko_patches, '1234567')
        tko_time = datetime.now()

        tko_job = models.job('/tmp/', 'root', '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']

        tko_test = models.test('/tmp/', 'mocktest', 'PASS', '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()
        js = job_serializer.JobSerializer()
        js.set_pb_job(self.tko_job, self.pb_job)


    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)

        self.check_patches(kernel.patches, newkernel.patches)


    def check_patches(self, patches, newpatches):
        for patch, newpatch in zip(patches, newpatches):
            self.assertEqual(patch.spec, newpatch.spec)
            self.assertEqual(patch.reference, newpatch.reference)
            self.assertEqual(patch.hash, newpatch.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()