• Home
  • History
  • Annotate
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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 datetime
10 import tempfile
11 import time
12 import unittest
13 
14 import common
15 from autotest_lib.tko import tko_pb2
16 from autotest_lib.tko import job_serializer
17 from autotest_lib.tko import models
18 
19 NamedTemporaryFile = tempfile.NamedTemporaryFile
20 datetime = datetime.datetime
21 mktime = time.mktime
22 
23 class 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(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 
282 class 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 
306 class 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 
388 if __name__ == '__main__':
389     unittest.main()
390