• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/python
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(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