3 """Unittests for the JobSerializer class.
5 Mostly test if the serialized object has the expected content.
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
21 NamedTemporaryFile
= tempfile
.NamedTemporaryFile
22 datetime
= datetime
.datetime
25 class JobSerializerUnittest(unittest
.TestCase
):
26 """Base class as a job serializer unittest"""
30 tko_patches
.append(models
.patch('New spec!', 'Reference?',
33 tko_kernel
= models
.kernel('My Computer', tko_patches
, '1234567')
34 tko_time
= datetime
.now()
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})
40 tko_iteration
= models
.iteration(0, {'2+2':4, '3+3':6},
41 {'4+4':8, '5+5':10, '6+6':12})
43 tko_labels
= ['unittest', 'dummy test', 'autotest']
45 tko_test
= models
.test('/tmp/', 'mocktest', 'Completed', 'N/A',
46 tko_kernel
, 'My Computer', tko_time
,
47 tko_time
, [tko_iteration
,
48 tko_iteration
, tko_iteration
],
49 {'abc':'def'}, tko_labels
)
51 self
.tko_job
= tko_job
52 self
.tko_job
.tests
= [tko_test
, tko_test
, tko_test
]
54 self
.pb_job
= tko_pb2
.Job()
56 self
.expected_afe_job_id
= '1'
58 js
= job_serializer
.JobSerializer()
59 js
.set_pb_job(self
.tko_job
, self
.pb_job
, self
.tag
)
63 self
.assertEqual(self
.tag
, self
.pb_job
.tag
)
66 def test_afe_job_id(self
):
67 self
.assertEqual(self
.expected_afe_job_id
,
68 self
.pb_job
.afe_job_id
)
71 def test_job_dir(self
):
72 """Check if the dir field are the same.
74 self
.assertEqual(self
.tko_job
.dir, self
.pb_job
.dir)
77 def test_number_of_test(self
):
78 """Check if the number of test are the same.
80 self
.assertEqual(len(self
.tko_job
.tests
),
81 len(self
.pb_job
.tests
))
85 """Check if the user field are the same.
87 self
.assertEqual(self
.tko_job
.user
, self
.pb_job
.user
)
90 def test_machine(self
):
91 """Check if the machine fields are the same.
93 self
.assertEqual(self
.tko_job
.machine
, self
.pb_job
.machine
)
96 def test_queued_time(self
):
97 """Check if queued_time are the same.
99 self
.check_time(self
.tko_job
.queued_time
,
100 self
.pb_job
.queued_time
)
103 def test_started_time(self
):
104 """Check if the started_time are the same.
106 self
.check_time(self
.tko_job
.started_time
,
107 self
.pb_job
.started_time
)
110 def test_finished_time(self
):
111 """Check if the finished_time are the same.
113 self
.check_time(self
.tko_job
.finished_time
,
114 self
.pb_job
.finished_time
)
117 def test_machine_owner(self
):
118 """Check if the machine owners are the same.
120 self
.assertEqual(self
.tko_job
.machine_owner
,
121 self
.pb_job
.machine_owner
)
124 def test_machine_group(self
):
125 """Check if the machine groups are the same.
127 self
.assertEqual(self
.tko_job
.machine_group
,
128 self
.pb_job
.machine_group
)
130 def test_aborted_by(self
):
131 """Check if the jobs are aborted by the same person.
133 self
.assertEqual(self
.tko_job
.aborted_by
,
134 self
.pb_job
.aborted_by
)
137 def test_aborted_on(self
):
138 self
.check_time(self
.tko_job
.aborted_on
,
139 self
.pb_job
.aborted_on
)
142 def test_keyval_dict(self
):
143 """Check if the contents of the dictionary are the same.
145 self
.assertEqual(len(self
.tko_job
.keyval_dict
),
146 len(self
.pb_job
.keyval_dict
))
147 self
.check_dict(self
.tko_job
.keyval_dict
,
148 self
.convert_keyval_to_dict(self
.pb_job
,
152 def test_tests(self
):
153 """Check if all the test are the same.
156 for test
, newtest
in zip(self
.tko_job
.tests
,
159 self
.assertEqual(test
.subdir
, newtest
.subdir
)
160 self
.assertEqual(test
.testname
, newtest
.testname
)
161 self
.assertEqual(test
.status
, newtest
.status
)
162 self
.assertEqual(test
.reason
, newtest
.reason
)
163 self
.assertEqual(test
.machine
, newtest
.machine
)
164 self
.assertEqual(test
.labels
, newtest
.labels
)
166 self
.check_time(test
.started_time
, newtest
.started_time
)
167 self
.check_time(test
.finished_time
, newtest
.finished_time
)
169 self
.check_iteration(test
.iterations
, newtest
.iterations
)
171 self
.check_dict(test
.attributes
,
172 self
.convert_keyval_to_dict(newtest
,
175 self
.check_kernel(test
.kernel
, newtest
.kernel
)
178 def check_time(self
, dTime
, stime
):
179 """Check if the datetime object contains the same time value
182 t
= mktime(dTime
.timetuple()) + 1e-6 * dTime
.microsecond
183 self
.assertEqual(long(t
), stime
/1000)
186 def check_iteration(self
, tko_iterations
, pb_iterations
):
187 """Check if the iteration objects are the same.
189 for tko_iteration
, pb_iteration
in zip(tko_iterations
,
192 self
.assertEqual(tko_iteration
.index
, pb_iteration
.index
)
194 self
.check_dict(tko_iteration
.attr_keyval
,
195 self
.convert_keyval_to_dict(pb_iteration
,
198 self
.check_dict(tko_iteration
.perf_keyval
,
199 self
.convert_keyval_to_dict(pb_iteration
,
203 def convert_keyval_to_dict(self
, var
, attr
):
204 """Convert a protocol buffer repeated keyval object into a
208 return dict((keyval
.name
, keyval
.value
) for keyval
in
212 def check_dict(self
, dictionary
, keyval
):
213 """Check if the contents of the dictionary are the same as a
214 repeated keyval pair.
216 for key
, value
in dictionary
.iteritems():
217 self
.assertTrue(key
in keyval
);
218 self
.assertEqual(str(value
), keyval
[key
])
221 def check_kernel(self
, kernel
, newkernel
):
222 """Check if the kernels are the same.
224 self
.assertEqual(kernel
.base
, newkernel
.base
)
225 self
.assertEqual(kernel
.kernel_hash
, newkernel
.kernel_hash
)
228 class ReadBackTest(JobSerializerUnittest
):
229 """Check if convert between models.job and pb job is correct even
230 after being written to binary and read by manually
234 super(ReadBackTest
, self
).setUp()
236 out_binary
= NamedTemporaryFile(mode
='wb')
238 out_binary
.write(self
.pb_job
.SerializeToString())
241 binary
= open(out_binary
.name
, 'rb')
243 self
.pb_job
= tko_pb2
.Job()
244 self
.pb_job
.ParseFromString(binary
.read())
251 class ReadBackGetterTest(JobSerializerUnittest
):
252 """Check if convert between models.job and pb job is correct after
253 using the getter methods in JobSerializer to read back the
258 super(ReadBackGetterTest
, self
).setUp()
260 temp_binary
= NamedTemporaryFile(mode
='wb')
262 temp_binary
.write(self
.pb_job
.SerializeToString())
265 js
= job_serializer
.JobSerializer()
266 self
.from_pb_job
= js
.deserialize_from_binary(temp_binary
.name
)
271 def test_keyval_dict(self
):
272 """Check if the contents of the dictionary are the same. """
274 self
.assertEqual(len(self
.tko_job
.keyval_dict
),
275 len(self
.from_pb_job
.keyval_dict
))
277 self
.check_dict(self
.tko_job
.keyval_dict
,
278 self
.from_pb_job
.keyval_dict
)
281 def test_tests(self
):
282 """Check if all the test are the same.
284 for test
, newtest
in zip(self
.tko_job
.tests
,
285 self
.from_pb_job
.tests
):
287 self
.assertEqual(test
.subdir
, newtest
.subdir
)
288 self
.assertEqual(test
.testname
, newtest
.testname
)
289 self
.assertEqual(test
.status
, newtest
.status
)
290 self
.assertEqual(test
.reason
, newtest
.reason
)
291 self
.assertEqual(test
.machine
, newtest
.machine
)
292 self
.assertEqual(test
.labels
, newtest
.labels
)
294 self
.check_time(test
.started_time
, newtest
.started_time
)
295 self
.check_time(test
.finished_time
, newtest
.finished_time
)
297 self
.check_iteration(test
.iterations
, newtest
.iterations
)
299 self
.check_dict(test
.attributes
, newtest
.attributes
)
301 self
.check_kernel(test
.kernel
, newtest
.kernel
)
304 def check_time(self
, dTime
, sTime
):
305 """Check if the datetime object contains the same time value
308 If sTime is type int or type long, then only convert dTime to
309 microseconds. Else, convert both dTime and sTime to
310 microseconds. Then, compare the two after casting them to
314 t
= mktime(dTime
.timetuple()) + 1e-6 * dTime
.microsecond
315 if isinstance(sTime
, (int, long)):
316 self
.assertEqual(long(t
*1000), sTime
)
318 t1
= mktime(sTime
.timetuple()) + 1e-6 * sTime
.microsecond
319 self
.assertEqual(long(t
*1000), long(t1
*1000))
322 def check_iteration(self
, iterations
, newiterations
):
323 """Check if the iteration objects are the same.
325 for iteration
, newiteration
in zip(iterations
, newiterations
):
326 self
.assertEqual(iteration
.index
, newiteration
.index
)
327 self
.check_dict(iteration
.attr_keyval
,
328 newiteration
.attr_keyval
)
329 self
.check_dict(iteration
.perf_keyval
,
330 newiteration
.perf_keyval
)
333 if __name__
== '__main__':