Fixed bug in devicespec parsing
[qa-tools:ots.git] / ots.server / ots / server / xmlrpc / tests.py
1 # ***** BEGIN LICENCE BLOCK *****
2 # This file is part of OTS
3 #
4 # Copyright (C) 2010 Nokia Corporation and/or its subsidiary(-ies).
5 #
6 # Contact: Mikko Makinen <mikko.al.makinen@nokia.com>
7 #
8 # This library is free software; you can redistribute it and/or
9 # modify it under the terms of the GNU Lesser General Public License
10 # version 2.1 as published by the Free Software Foundation.
11 #
12 # This library is distributed in the hope that it will be useful, but
13 # WITHOUT ANY WARRANTY; without even the implied warranty of
14 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
15 # Lesser General Public License for more details.
16 #
17 # You should have received a copy of the GNU Lesser General Public
18 # License along with this library; if not, write to the Free Software
19 # Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
20 # 02110-1301 USA
21 # ***** END LICENCE BLOCK *****
22
23 """Unit tests for xmlrpc interface and handleprocesses modules"""
24
25 import unittest
26
27 import logging
28
29 from ots.server.xmlrpc.public import request_sync
30 from ots.server.xmlrpc.public import REQUEST_ERROR as REQUEST_ERROR
31 from ots.server.xmlrpc.public import _repack_options
32 from ots.server.xmlrpc.public import _string_2_list
33 from ots.server.xmlrpc.public import _run_test 
34 from ots.server.xmlrpc.public import _check_testruns_result_values
35 from ots.server.xmlrpc.public import _parse_multiple_devicegroup_specs
36 from ots.server.xmlrpc.public import _read_queues
37 from ots.server.xmlrpc.public import _prepare_testrun
38 from ots.server.xmlrpc.public import _create_testruns
39 from ots.server.xmlrpc.public import _validate_devicespecs
40 from ots.server.xmlrpc.handleprocesses import HandleProcesses
41
42 from multiprocessing import Process, Queue
43 import ots.server.xmlrpc.public as public
44
45 class TestrunHost(object):
46     """Mocked TestRunHost"""
47
48     def __init__(self):
49         print "Fake TestRunHost initialised"
50
51     def init_testrun(self,
52                      request,
53                      program,
54                      options,
55                      notify_list,
56                      test_packages,
57                      image_url,
58                      rootstrap_url):
59
60         self.request = request
61         self.program = program
62         self.options = options
63         self.notify_list = notify_list
64         self.test_packages = test_packages
65         self.image_url = image_url
66         self.rootstrap_url = rootstrap_url
67
68         global testdata_init_testrun
69         testdata_init_testrun = [request,
70                                  program,
71                                  options,
72                                  notify_list,
73                                  test_packages,
74                                  image_url,
75                                  rootstrap_url]
76
77     def register_ta_plugins(self):
78         pass
79
80     def register_results_plugins(self):
81         pass
82
83     def run(self):
84         pass
85
86     def testrun_result(self):
87         return "PASS"
88
89 #################################
90 # Stubs
91 ################################
92
93 class DummyHost(object):
94     """Another mocked TestRunHost"""
95
96     def init_testrun(self, request, testrun_id, program,
97                            options, notify_list, test_packages,
98                            image_url, rootstrap_url):
99         pass
100
101     def register_ta_plugins(self):
102         pass
103
104     def register_results_plugins(self):
105         pass
106
107     def run(self):
108         pass
109
110     def publish_result_links(self):
111         pass
112
113     def cleanup(self):
114         pass
115
116     def testrun_result(self):
117         return "PASS"
118
119
120 def _get_dummy_host(*args):
121     return DummyHost()
122
123
124 class DummyHostRaises(DummyHost):
125
126     def run(self):
127         raise
128
129 def _get_dummy_host_raises(*args):
130     return DummyHostRaises()
131
132 class DummyHostFails(DummyHost):
133     
134     def testrun_result(self):
135         return "NOT_PASS"
136
137
138 def _get_dummy_host_fails(*args):
139     return DummyHostFails()
140
141
142 def _get_dummy_host_None(*args):
143     return None
144
145 #########################################
146
147 class Test_xmlrpc_interface(unittest.TestCase):
148
149     def test_request_sync(self):
150         self.req_id = "123007"
151
152         self.image_url = 'http://image_url'
153         self.rootstrap_url = 'http://rootstrap_url'
154
155         self.sw_product = 'swproduct'
156         self.email_list = ['nonexisting@notexistingdomain.foobar']
157
158         self.interface_options = dict()
159         self.interface_options['device'] = 'devicegroup:not_real_device'
160         self.interface_options['packages'] = "foobar-tests foobaz-tests  "  #Here are the extra spaces!
161         self.interface_options['testfilter'] = "testsuite=testsuite2"
162
163         self.interface_options['image'] = self.image_url
164         self.interface_options['rootstrap'] = self.rootstrap_url
165         self.interface_options['execute'] = "false"
166
167         result = request_sync(self.sw_product, self.req_id, self.email_list, self.interface_options)
168
169         self.assertEquals(result, 'ERROR')
170
171     def test_request_sync_multiple_devicegroups_with_empy_child_processes_list(self):
172         """
173         Try request_sync with errorneous 'device' option. Should return ERROR since no process instances are created for
174         multiple devicegroup.
175         """
176         self.req_id = "123007"
177
178         self.image_url = 'http://image_url'
179         self.rootstrap_url = 'http://rootstrap_url'
180
181         self.sw_product = 'swproduct'
182         self.email_list = ['nonexisting@notexistingdomain.foobar']
183
184         self.interface_options = dict()
185         self.interface_options['device'] = 'syntaxerror:foobazored_device;syntaxerror:bogus_device;syntaxerror:huuhaa_device'
186         self.interface_options['packages'] = "foobar-tests foobaz-tests  "  #Here are the extra spaces!
187         self.interface_options['testfilter'] = "testsuite=testsuite2"
188
189         self.interface_options['image'] = self.image_url
190         self.interface_options['rootstrap'] = self.rootstrap_url
191
192         result = request_sync(self.sw_product, self.req_id, self.email_list, self.interface_options)
193
194         self.assertEquals(result, 'ERROR')
195
196     def test_request_sync_multiple_devicegroups_with_invalid_devicegroup_syntax(self):
197         """
198         New format for devicegroup is 'devicegroup:groupname property:value', meaning that properties for devicegroup are separated with whitespace,
199         it's not valid to add whitespace after devicegroup value: 'devicegroup: groupname property:value'
200         """
201         self.req_id = "123007"
202
203         self.image_url = 'http://image_url'
204         self.rootstrap_url = 'http://rootstrap_url'
205
206         self.sw_product = 'swproduct'
207         self.email_list = ['nonexisting@notexistingdomain.foobar']
208
209         self.interface_options = dict()
210         self.interface_options['device'] = 'devicegroup: foobazored_device'
211         self.interface_options['packages'] = "foobar-tests foobaz-tests  "  #Here are the extra spaces!
212         self.interface_options['testfilter'] = "testsuite=testsuite2"
213
214         self.interface_options['image'] = self.image_url
215         self.interface_options['rootstrap'] = self.rootstrap_url
216
217         result = request_sync(self.sw_product, self.req_id, self.email_list, self.interface_options)
218
219         self.assertEquals(result, 'ERROR')
220
221     def test_repack_options(self):
222         options = {'hosttest': 'this is a hosttest',
223                    'engine': 'this is the engine',
224                    'device': 'a:1 b:2 c:3',
225                    'emmc': 'EMMC',
226                    'email-attachments': 'off'}
227
228         expected = {'engine': ['this', 'is', 'the', 'engine'], 
229                     'device': [{'a': '1', 'c': '3', 'b': '2'}], 
230                     'emmc': 'EMMC', 
231                     'hosttest': ['this', 'is', 'a', 'hosttest'],
232                     'email-attachments': False}
233         self.assertEquals(expected, _repack_options(options))
234
235     def test_repack_options_extra_options(self): 
236         options = {'hosttest': 'this is a hosttest',
237                    'engine': 'this is the engine',
238                    'device': 'a:1 b:2 c:3',
239                    'emmc': 'EMMC',
240                    'foo': 'FOO',
241                    'bar': 'BAR'}
242
243         expected = {'engine': ['this', 'is', 'the', 'engine'], 
244                     'device': [{'a': '1', 'c': '3', 'b': '2'}], 
245                     'emmc': 'EMMC', 
246                     'hosttest': ['this', 'is', 'a', 'hosttest'],
247                     'foo':'FOO',
248                     'bar':'BAR'}
249         self.assertEquals(expected, _repack_options(options))
250
251     def test_repack_options_multiple_device_groups(self):
252         options = {'hosttest': 'this is a hosttest',
253                    'engine': 'this is the engine',
254                    'device': 'devicegroup:foobar;devicegroup:foobazor b:2',
255                    'emmc': 'EMMC',
256                    'email-attachments': 'off'}
257
258         expected = {'engine': ['this', 'is', 'the', 'engine'],
259                     'device': [{'devicegroup':'foobar'}, \
260                               {'b': '2', 'devicegroup': 'foobazor'}],
261                     'emmc': 'EMMC',
262                     'hosttest': ['this', 'is', 'a', 'hosttest'],
263                     'email-attachments': False}
264         self.assertEquals(expected, _repack_options(options))
265
266     def test_repack_options_email_attachments(self):
267         options = {'email-attachments': 'on'}
268         expected = {'email-attachments': True}
269         self.assertEquals(expected, _repack_options(options))
270         options = {'email-attachments': 'off'}
271         expected = {'email-attachments': False}
272         self.assertEquals(expected, _repack_options(options))
273
274     def test_string_2_list(self):
275         expected = ['mary', 'had', 'a', 'little', 'lamb']
276         self.assertEquals(expected, 
277                           _string_2_list("mary had a little lamb"))
278
279     def test_parse_multiple_devicegroup_specs(self):
280         expected = [{'devicegroup': 'group1'}, {'devicegroup': 'group2', 'sim': 'operator'}]
281         args = "devicegroup:group1;devicegroup:group2 sim:operator"
282         self.assertEquals(expected,
283            _parse_multiple_devicegroup_specs(args))
284
285     def test_parse_multiple_devicegroup_specs_invalid_data(self):
286         """
287         Whitespaces are used to separate devicegroup properties and are not allowed between
288         devicegroup key and it's value
289         """
290         args = "devicegroup: group1 sim:operator2;devicegroup: group2 sim:operator2"
291         try:
292             _parse_multiple_devicegroup_specs(args)
293         except IndexError:
294             pass
295
296     def test_one_devicespec_devicegroup(self):
297         options = {'device': [{'devicegroup': 'foobar_1'}], \
298                    'timeout': '1'}
299         testrun_list, process_queues = \
300             _create_testruns(options, 'request', 'program', 'noemail@nodomain.nodot', \
301                 'foobar-tests', 'https://image_url', 'http://roostrap_url')
302
303         self.assertTrue(len(testrun_list) == 1)
304         self.assertTrue(len(process_queues) == 1)
305
306     def test_one_devicespec_devicegroup_two_devicegroups(self):
307         options = {'device': [{'devicegroup': 'foobar_1'}, \
308                    {'devicegroup': 'foobar_2'}], 'timeout': '1'}
309         testrun_list, process_queues = \
310             _create_testruns(options, 'request', 'program', 'noemail@nodomain.nodot', \
311                 'foobar-tests', 'https://image_url', 'http://roostrap_url')
312
313         self.assertTrue(len(testrun_list) == 2)
314         self.assertTrue(len(process_queues) == 2)
315
316     def test_multiple_devicespecs(self):
317         options = {'device': [{'devicegroup': 'foobar_1', 'devicename': 'mydevice', \
318                    'deviceid': '1'}], 'timeout': '1'}
319         testrun_list, process_queues = \
320             _create_testruns(options, 'request', 'program', 'noemail@nodomain.nodot', \
321                 'foobar-tests', 'https://image_url', 'http://roostrap_url')
322
323         self.assertTrue(len(testrun_list) == 1)
324         self.assertTrue(len(process_queues) == 1)
325
326     def test_multiple_devicespecs_two_devicegroups(self):
327         options = {'device': [{'devicegroup': 'foobar_1', 'devicename': 'device1', \
328                    'deviceid': '1'}, {'devicegroup': 'foobar_2', 'devicename': 'device2', \
329                    'deviceid': '2'}], 'timeout': '1'}
330         testrun_list, process_queues = \
331             _create_testruns(options, 'request', 'program', 'noemail@nodomain.nodot', \
332                 'foobar-tests', 'https://image_url', 'http://roostrap_url')
333
334         self.assertTrue(len(testrun_list) == 2)
335         self.assertTrue(len(process_queues) == 2)
336
337     def test_with_bogus_devicespec(self):
338         options = {'device': [{'bogusspec': 'foobazor'}], \
339                    'timeout': '1'}
340         testrun_list, process_queues = \
341             _create_testruns(options, 'request', 'program', 'noemail@nodomain.nodot', \
342                 'foobar-tests', 'https://image_url', 'http://roostrap_url')
343
344         self.assertTrue(len(testrun_list) == 0)
345
346     def test_with_devicename_and_deviceid_devicespec(self):
347         options = {'device': [{'devicename': 'test1', 'deviceid': '1'}], \
348                    'timeout': '1'}
349         testrun_list, process_queues = \
350             _create_testruns(options, 'request', 'program', 'noemail@nodomain.nodot', \
351                 'foobar-tests', 'https://image_url', 'http://roostrap_url')
352
353         self.assertTrue(len(testrun_list) == 1)
354
355     def test_with_devicename_devicespec(self):
356         options = {'device': [{'devicename': 'test1'}], \
357                    'timeout': '1'}
358         testrun_list, process_queues = \
359             _create_testruns(options, 'request', 'program', 'noemail@nodomain.nodot', \
360                 'foobar-tests', 'https://image_url', 'http://roostrap_url')
361         
362         self.assertTrue(len(testrun_list) == 1)
363
364     def test_with_deviceid_devicespec(self):
365         options = {'device': [{'deviceid': '1'}], \
366                    'timeout': '1'}
367         testrun_list, process_queues = \
368             _create_testruns(options, 'request', 'program', 'noemail@nodomain.nodot', \
369                 'foobar-tests', 'https://image_url', 'http://roostrap_url')
370        
371         self.assertTrue(len(testrun_list) == 1)
372
373     def test_validate_devicespecs_true(self):
374         devicespecs = ['devicegroup', 'devicename', 'deviceid']
375         self.assertTrue(_validate_devicespecs(devicespecs))
376         self.assertTrue(_validate_devicespecs([devicespecs[0]]))
377         self.assertTrue(_validate_devicespecs([devicespecs[1]]))
378         self.assertTrue(_validate_devicespecs([devicespecs[2]]))
379
380     def test_validate_devicespecs_false(self):
381         devicespecs = ['devicegroup', 'devicename', 'bogusspec']
382         self.assertFalse(_validate_devicespecs(devicespecs))
383
384         devicespecs = ['foobarspec']
385         self.assertFalse(_validate_devicespecs(devicespecs))
386
387     def test_prepare(self):
388         options = {'hosttest': 'this is a hosttest',
389                    'engine': 'this is the engine',
390                    'device': 'a:1 b:2 c:3',
391                    'emmc': 'EMMC'}
392
393         new_options, pq = _prepare_testrun(options)
394         self.assertEquals(type(pq), type(Queue()))
395         self.assertEquals(options, new_options)
396         self.assertFalse(options is new_options)
397
398     def test_check_result_values(self):
399         return_error = ['ERROR', 'FAIL', 'PASS']
400         self.assertEqual('ERROR', _check_testruns_result_values(return_error))
401
402         return_error = ['FAIL', 'PASS']
403         self.assertEqual('FAIL', _check_testruns_result_values(return_error))
404
405         return_error = ['PASS','PASS']
406         self.assertEqual('PASS', _check_testruns_result_values(return_error))
407
408     def test_run_test_passed(self):
409         #Override host with stub
410         public._get_testrun_host = _get_dummy_host
411
412         log = logging.getLogger(__name__)
413         options = {'hosttest': 'this is a hosttest',
414                    'engine': 'this is the engine',
415                    'device': 'a:1 b:2 c:3',
416                    'emmc': 'EMMC'}
417
418         self.assertEquals("PASS", 
419                           _run_test(log, "request", "testrun_id", 
420                                     "program", options, 
421                                     ["foo@bar.com"], ["p1", "p2"], 
422                                      "image", "rootstrap"))
423         
424     def test_run_test_raises(self):
425         #Override host with stub
426         public._get_testrun_host = _get_dummy_host_raises
427
428         log = logging.getLogger(__name__)
429         options = {'hosttest': 'this is a hosttest',
430                    'engine': 'this is the engine',
431                    'device': 'a:1 b:2 c:3',
432                    'emmc': 'EMMC'}
433
434         self.assertEquals("ERROR", 
435                           _run_test(log, "request", "testrun_id", 
436                                     "program", options, 
437                                     ["foo@bar.com"], ["p1", "p2"], 
438                                      "image", "rootstrap"))
439
440     def test_run_test_fails(self):
441         #Override host with stub
442         public._get_testrun_host = _get_dummy_host_fails
443
444         log = logging.getLogger(__name__)
445         options = {'hosttest': 'this is a hosttest',
446                    'engine': 'this is the engine',
447                    'device': 'a:1 b:2 c:3',
448                    'emmc': 'EMMC'}
449
450         self.assertEquals("FAIL", 
451                           _run_test(log, "request", "testrun_id", 
452                                     "program", options, 
453                                     ["foo@bar.com"], ["p1", "p2"], 
454                                      "image", "rootstrap"))
455         
456     def test_run_no_test_host(self):
457         #Override host with stub
458         public._get_testrun_host = _get_dummy_host_None
459
460         log = logging.getLogger(__name__)
461         options = {'hosttest': 'this is a hosttest',
462                    'engine': 'this is the engine',
463                    'device': 'a:1 b:2 c:3',
464                    'emmc': 'EMMC'}
465
466         self.assertEquals("FAIL", 
467                           _run_test(log, "request", "testrun_id", 
468                                     "program", options, 
469                                     ["foo@bar.com"], ["p1", "p2"], 
470                                      "image", "rootstrap"))
471         
472
473 class Test_handleprocess(unittest.TestCase):
474
475     def test_start_processes_and_read_process_queues(self):
476         def _run_process(pq, id, string):
477             pq.put({id:string})
478
479         process_queues = []
480         process_handler = HandleProcesses()
481         p1q = Queue()
482         p2q = Queue()
483         process_handler.add_process(_run_process, (p1q, "1", "ping"))
484         process_handler.add_process(_run_process, (p2q, "2", "pong"))
485         process_queues.append(p1q)
486         process_queues.append(p2q)
487
488         process_handler.start_processes()
489         ret = _read_queues(process_queues)
490         process_handler.join_processes()
491
492         self.assertEquals(ret["1"], "ping")
493         self.assertEquals(ret["2"], "pong")
494
495 if __name__ == '__main__':
496     unittest.main()