tko/job_serializer_unittest.py

   1 #!/usr/bin/python
   2
   3 """Unittests for the JobSerializer class.
   4
   5 Mostly test if the serialized object has the expected content.
   6
   7 """
   8
   9 import common
  10 import datetime
  11 import os
  12 import re
  13 import tempfile
  14 import time
  15
  16 from autotest_lib.tko import tko_pb2
  17 from autotest_lib.tko import job_serializer
  18 from autotest_lib.tko import models
  19 from autotest_lib.client.common_lib.test_utils import unittest
  20
  21 NamedTemporaryFile = tempfile.NamedTemporaryFile
  22 datetime = datetime.datetime
  23 mktime = time.mktime
  24
  25 class JobSerializerUnittest(unittest.TestCase):
  26     """Base class as a job serializer unittest"""
  27
  28     def setUp(self):
  29         tko_patches = []
  30         tko_patches.append(models.patch('New spec!', 'Reference?',
  31                                         123456))
  32
  33         tko_kernel = models.kernel('My Computer', tko_patches, '1234567')
  34         tko_time = datetime.now()
  35
  36         tko_job = models.job('/tmp/', 'autotest', 'test', 'My Computer',
  37                              tko_time, tko_time, tko_time, 'root',
  38                              'www', 'No one', tko_time, {'1+1':2})
  39
  40         tko_iteration = models.iteration(0, {'2+2':4, '3+3':6},
  41                                    {'4+4':8, '5+5':10, '6+6':12})
  42
  43         tko_labels = ['unittest', 'dummy test', 'autotest']
  44
  45         tko_test = models.test('/tmp/', 'mocktest', 'Completed', 'N/A',
  46                                tko_kernel, 'My Computer', tko_time,
  47                                tko_time, [tko_iteration,
  48                                tko_iteration, tko_iteration],
  49                                {'abc':'def'}, tko_labels)
  50
  51         self.tko_job = tko_job
  52         self.tko_job.tests = [tko_test, tko_test, tko_test]
  53
  54         self.pb_job = tko_pb2.Job()
  55         self.tag = '1-abc./.'
  56         self.expected_afe_job_id = '1'
  57
  58         js = job_serializer.JobSerializer()
  59         js.set_pb_job(self.tko_job, self.pb_job, self.tag)
  60
  61
  62     def test_tag(self):
  63         self.assertEqual(self.tag, self.pb_job.tag)
  64
  65
  66     def test_afe_job_id(self):
  67         self.assertEqual(self.expected_afe_job_id,
  68                          self.pb_job.afe_job_id)
  69
  70
  71     def test_job_dir(self):
  72         """Check if the dir field are the same.
  73         """
  74         self.assertEqual(self.tko_job.dir, self.pb_job.dir)
  75
  76
  77     def test_number_of_test(self):
  78         """Check if the number of test are the same.
  79         """
  80         self.assertEqual(len(self.tko_job.tests),
  81                          len(self.pb_job.tests))
  82
  83
  84     def test_user(self):
  85         """Check if the user field are the same.
  86         """
  87         self.assertEqual(self.tko_job.user, self.pb_job.user)
  88
  89
  90     def test_machine(self):
  91         """Check if the machine fields are the same.
  92         """
  93         self.assertEqual(self.tko_job.machine, self.pb_job.machine)
  94
  95
  96     def test_queued_time(self):
  97         """Check if queued_time are the same.
  98         """
  99         self.check_time(self.tko_job.queued_time,
 100                         self.pb_job.queued_time)
 101
 102
 103     def test_started_time(self):
 104         """Check if the started_time are the same.
 105         """
 106         self.check_time(self.tko_job.started_time,
 107                         self.pb_job.started_time)
 108
 109
 110     def test_finished_time(self):
 111         """Check if the finished_time are the same.
 112         """
 113         self.check_time(self.tko_job.finished_time,
 114                         self.pb_job.finished_time)
 115
 116
 117     def test_machine_owner(self):
 118         """Check if the machine owners are the same.
 119         """
 120         self.assertEqual(self.tko_job.machine_owner,
 121                          self.pb_job.machine_owner)
 122
 123
 124     def test_machine_group(self):
 125         """Check if the machine groups are the same.
 126         """
 127         self.assertEqual(self.tko_job.machine_group,
 128                          self.pb_job.machine_group)
 129
 130     def test_aborted_by(self):
 131         """Check if the jobs are aborted by the same person.
 132         """
 133         self.assertEqual(self.tko_job.aborted_by,
 134                          self.pb_job.aborted_by)
 135
 136
 137     def test_aborted_on(self):
 138         self.check_time(self.tko_job.aborted_on,
 139                         self.pb_job.aborted_on)
 140
 141
 142     def test_keyval_dict(self):
 143         """Check if the contents of the dictionary are the same.
 144         """
 145         self.assertEqual(len(self.tko_job.keyval_dict),
 146                          len(self.pb_job.keyval_dict))
 147         self.check_dict(self.tko_job.keyval_dict,
 148                         self.convert_keyval_to_dict(self.pb_job,
 149                         'keyval_dict'))
 150
 151
 152     def test_tests(self):
 153         """Check if all the test are the same.
 154         """
 155
 156         for test, newtest in zip(self.tko_job.tests,
 157                                  self.pb_job.tests):
 158
 159             self.assertEqual(test.subdir, newtest.subdir)
 160             self.assertEqual(test.testname, newtest.testname)
 161             self.assertEqual(test.status, newtest.status)
 162             self.assertEqual(test.reason, newtest.reason)
 163             self.assertEqual(test.machine, newtest.machine)
 164             self.assertEqual(test.labels, newtest.labels)
 165
 166             self.check_time(test.started_time, newtest.started_time)
 167             self.check_time(test.finished_time, newtest.finished_time)
 168
 169             self.check_iteration(test.iterations, newtest.iterations)
 170
 171             self.check_dict(test.attributes,
 172                             self.convert_keyval_to_dict(newtest,
 173                             'attributes'))
 174
 175             self.check_kernel(test.kernel, newtest.kernel)
 176
 177
 178     def check_time(self, dTime, stime):
 179         """Check if the datetime object contains the same time value
 180         in microseconds.
 181         """
 182         t = mktime(dTime.timetuple()) + 1e-6 * dTime.microsecond
 183         self.assertEqual(long(t), stime/1000)
 184
 185
 186     def check_iteration(self, tko_iterations, pb_iterations):
 187         """Check if the iteration objects are the same.
 188         """
 189         for tko_iteration, pb_iteration in zip(tko_iterations,
 190                                                pb_iterations):
 191
 192             self.assertEqual(tko_iteration.index, pb_iteration.index)
 193
 194             self.check_dict(tko_iteration.attr_keyval,
 195                             self.convert_keyval_to_dict(pb_iteration,
 196                                                         'attr_keyval'))
 197
 198             self.check_dict(tko_iteration.perf_keyval,
 199                             self.convert_keyval_to_dict(pb_iteration,
 200                                                         'perf_keyval'))
 201
 202
 203     def convert_keyval_to_dict(self, var, attr):
 204         """Convert a protocol buffer repeated keyval object into a
 205         python dict.
 206         """
 207
 208         return dict((keyval.name, keyval.value) for keyval in
 209                     getattr(var,attr))
 210
 211
 212     def check_dict(self, dictionary, keyval):
 213         """Check if the contents of the dictionary are the same as a
 214         repeated keyval pair.
 215         """
 216         for key, value in dictionary.iteritems():
 217             self.assertTrue(key in keyval);
 218             self.assertEqual(str(value), keyval[key])
 219
 220
 221     def check_kernel(self, kernel, newkernel):
 222         """Check if the kernels are the same.
 223         """
 224         self.assertEqual(kernel.base, newkernel.base)
 225         self.assertEqual(kernel.kernel_hash, newkernel.kernel_hash)
 226
 227
 228 class ReadBackTest(JobSerializerUnittest):
 229     """Check if convert between models.job and pb job is correct even
 230     after being written to binary and read by manually
 231     """
 232
 233     def setUp(self):
 234         super(ReadBackTest, self).setUp()
 235
 236         out_binary = NamedTemporaryFile(mode='wb')
 237         try:
 238             out_binary.write(self.pb_job.SerializeToString())
 239             out_binary.flush()
 240
 241             binary = open(out_binary.name, 'rb')
 242             try:
 243                 self.pb_job = tko_pb2.Job()
 244                 self.pb_job.ParseFromString(binary.read())
 245             finally:
 246                 binary.close()
 247         finally:
 248             out_binary.close()
 249
 250
 251 class ReadBackGetterTest(JobSerializerUnittest):
 252     """Check if convert between models.job and pb job is correct after
 253     using the getter methods in JobSerializer to read back the
 254     data.
 255     """
 256
 257     def setUp(self):
 258         super(ReadBackGetterTest, self).setUp()
 259
 260         temp_binary = NamedTemporaryFile(mode='wb')
 261         try:
 262             temp_binary.write(self.pb_job.SerializeToString())
 263             temp_binary.flush()
 264
 265             js = job_serializer.JobSerializer()
 266             self.from_pb_job = js.deserialize_from_binary(temp_binary.name)
 267         finally:
 268             temp_binary.close()
 269
 270
 271     def test_keyval_dict(self):
 272         """Check if the contents of the dictionary are the same.  """
 273
 274         self.assertEqual(len(self.tko_job.keyval_dict),
 275                          len(self.from_pb_job.keyval_dict))
 276
 277         self.check_dict(self.tko_job.keyval_dict,
 278                         self.from_pb_job.keyval_dict)
 279
 280
 281     def test_tests(self):
 282         """Check if all the test are the same.
 283         """
 284         for test, newtest in zip(self.tko_job.tests,
 285                                  self.from_pb_job.tests):
 286
 287             self.assertEqual(test.subdir, newtest.subdir)
 288             self.assertEqual(test.testname, newtest.testname)
 289             self.assertEqual(test.status, newtest.status)
 290             self.assertEqual(test.reason, newtest.reason)
 291             self.assertEqual(test.machine, newtest.machine)
 292             self.assertEqual(test.labels, newtest.labels)
 293
 294             self.check_time(test.started_time, newtest.started_time)
 295             self.check_time(test.finished_time, newtest.finished_time)
 296
 297             self.check_iteration(test.iterations, newtest.iterations)
 298
 299             self.check_dict(test.attributes, newtest.attributes)
 300
 301             self.check_kernel(test.kernel, newtest.kernel)
 302
 303
 304     def check_time(self, dTime, sTime):
 305         """Check if the datetime object contains the same time value
 306         in microseconds.
 307
 308         If sTime is type int or type long, then only convert dTime to
 309         microseconds. Else, convert both dTime and sTime to
 310         microseconds. Then, compare the two after casting them to
 311         long.
 312         """
 313
 314         t = mktime(dTime.timetuple()) + 1e-6 * dTime.microsecond
 315         if isinstance(sTime, (int, long)):
 316             self.assertEqual(long(t*1000), sTime)
 317         else:
 318             t1 = mktime(sTime.timetuple()) + 1e-6 * sTime.microsecond
 319             self.assertEqual(long(t*1000), long(t1*1000))
 320
 321
 322     def check_iteration(self, iterations, newiterations):
 323         """Check if the iteration objects are the same.
 324         """
 325         for iteration, newiteration in zip(iterations, newiterations):
 326             self.assertEqual(iteration.index, newiteration.index)
 327             self.check_dict(iteration.attr_keyval,
 328                             newiteration.attr_keyval)
 329             self.check_dict(iteration.perf_keyval,
 330                             newiteration.perf_keyval)
 331
 332
 333 if __name__ == '__main__':
 334     unittest.main()