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