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