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 common
10import datetime
11import os
12import re
13import tempfile
14import time
15
16from autotest_lib.tko import tko_pb2
17from autotest_lib.tko import job_serializer
18from autotest_lib.tko import models
19from autotest_lib.client.common_lib.test_utils import unittest
20
21NamedTemporaryFile = tempfile.NamedTemporaryFile
22datetime = datetime.datetime
23mktime = time.mktime
24
25class 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
230class 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
253class 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
335if __name__ == '__main__':
336    unittest.main()
337