OLD | NEW |
1 # Copyright (c) 2010 The Chromium OS Authors. All rights reserved. | 1 # Copyright (c) 2010 The Chromium OS Authors. All rights reserved. |
2 # Use of this source code is governed by a BSD-style license that can be | 2 # Use of this source code is governed by a BSD-style license that can be |
3 # found in the LICENSE file. | 3 # found in the LICENSE file. |
4 | 4 |
5 import common, fnmatch, logging, os, re, string, threading, time | 5 import common, fnmatch, logging, os, re, string, threading, time |
6 | 6 |
7 from autotest_lib.server import autotest, hosts, subcommand | 7 from autotest_lib.server import autotest, hosts, subcommand |
8 from autotest_lib.server import site_bsd_router | 8 from autotest_lib.server import site_bsd_router |
9 from autotest_lib.server import site_linux_router | 9 from autotest_lib.server import site_linux_router |
10 from autotest_lib.server import site_host_attributes | 10 from autotest_lib.server import site_host_attributes |
11 from autotest_lib.server import site_eap_tls | 11 from autotest_lib.server import site_eap_tls |
| 12 from autotest_lib.server import test |
12 from autotest_lib.client.common_lib import error | 13 from autotest_lib.client.common_lib import error |
13 | 14 |
14 class NotImplemented(Exception): | 15 class NotImplemented(Exception): |
15 def __init__(self, what): | 16 def __init__(self, what): |
16 self.what = what | 17 self.what = what |
17 | 18 |
18 | 19 |
19 def __str__(self): | 20 def __str__(self): |
20 return repr("Test method '%s' not implemented" % self.what) | 21 return repr("Test method '%s' not implemented" % self.what) |
21 | 22 |
(...skipping 781 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
803 if server_addr is None and hasattr(client_attributes, 'server_addr'): | 804 if server_addr is None and hasattr(client_attributes, 'server_addr'): |
804 server_addr = client_attributes.server_addr | 805 server_addr = client_attributes.server_addr |
805 if server_addr is not None: | 806 if server_addr is not None: |
806 server['addr'] = server_addr; | 807 server['addr'] = server_addr; |
807 # TODO(sleffler) check for wifi_addr when no control address | 808 # TODO(sleffler) check for wifi_addr when no control address |
808 | 809 |
809 # tag jobs w/ the router's address on the control network | 810 # tag jobs w/ the router's address on the control network |
810 config['tagname'] = router['addr'] | 811 config['tagname'] = router['addr'] |
811 | 812 |
812 return config | 813 return config |
| 814 |
| 815 def run_test_dir(test_name, job, args, machines): |
| 816 # convert autoserv args to something usable |
| 817 opts = dict([[k, v] for (k, e, v) in [x.partition('=') for x in args]]) |
| 818 |
| 819 config_file = opts.get('config_file', 'wifi_testbed_config') |
| 820 test_pat = opts.get('test_pat', '[0-9]*') |
| 821 router_addr = opts.get('router_addr', None) |
| 822 server_addr = opts.get('server_addr', None) |
| 823 |
| 824 config = read_wifi_testbed_config( |
| 825 os.path.join(job.configdir, config_file), |
| 826 client_addr = machines[0], # NB: take client identity from command li
ne |
| 827 router_addr = router_addr, |
| 828 server_addr = server_addr) |
| 829 server = config['server'] |
| 830 router = config['router'] |
| 831 |
| 832 logging.info("Client %s, Server %s, AP %s" % \ |
| 833 (machines[0], server.get('addr', 'N/A'), router['addr'])) |
| 834 |
| 835 test_dir = os.path.join(job.serverdir, "site_tests", test_name) |
| 836 |
| 837 for t in read_tests(test_dir, test_pat): |
| 838 job.run_test(test_name, testcase=t, config=config, tag=t['file']) |
| 839 |
| 840 class test(test.test): |
| 841 """ |
| 842 Base class for network_WiFi* classes that are created in the control |
| 843 directory for each test suite |
| 844 """ |
| 845 version = 1 |
| 846 |
| 847 def expect_failure(self, name, reason): |
| 848 if reason is None: |
| 849 reason = "no reason given" |
| 850 logging.info("%s: ignore failure (%s)", name, reason) |
| 851 |
| 852 |
| 853 # The testcase config, setup, etc are done out side the individual |
| 854 # test loop, in the control file. |
| 855 def run_once(self, testcase, config): |
| 856 name = testcase['name'] |
| 857 try: |
| 858 if 'skip_test' in testcase: |
| 859 logging.info("%s: SKIP: %s", name, testcase['skip_test']) |
| 860 else: |
| 861 wt = WiFiTest(name, testcase['steps'], config) |
| 862 wt.run() |
| 863 wt.write_keyvals(self) |
| 864 except error.TestFail: |
| 865 if 'expect_failure' in testcase: |
| 866 self.expect_failure(name, testcase['expect_failure']) |
| 867 else: |
| 868 raise |
| 869 except Exception, e: |
| 870 if 'expect_failure' in testcase: |
| 871 self.expect_failure(name, testcase['expect_failure']) |
| 872 else: |
| 873 raise error.TestFail(e) |
OLD | NEW |