1# Copyright 2015 The Chromium OS Authors. All rights reserved.
2# Use of this source code is governed by a BSD-style license that can be
3# found in the LICENSE file.
4
5import dbus
6import logging
7import tempfile
8import time
9
10from autotest_lib.client.bin import test
11from autotest_lib.client.bin import utils
12from autotest_lib.client.common_lib import error
13from autotest_lib.client.common_lib.cros.network \
14         import wifi_rack_constants as constants
15from autotest_lib.client.cros.networking import wifi_proxy
16from autotest_lib.client.cros.networking.chrome_testing \
17        import chrome_networking_test_api as cnta
18from autotest_lib.client.cros.networking.chrome_testing \
19        import chrome_networking_test_context as cntc
20
21
22class network_RackWiFiConnect(test.test):
23    """Client test to connect to various network services on WiFi rack.
24
25    After connection, we assert access to pages only accessible through the
26    connected network.
27
28    """
29    version = 1
30
31
32    def _assert_access(self, test):
33        """Asset user can access page.
34
35        Verification URLs are either pages on WiFi rack's Apache server or
36        general Internet.
37
38        @param test string - testname of NetworkServices namedtuple
39
40        @return boolean - True if able to access, False otherwise
41
42        """
43        for service_test in constants.NETWORK_SERVICES_TESTS:
44            if test == service_test.testname:
45                url, pattern = service_test.url, service_test.pattern
46                break
47
48        # Since this test runs OTA, allow 15 seconds of leeway
49        time.sleep(15)
50
51        wget_cmd = 'wget -O /tmp/wget.log %s' % url
52        for retry in range(3):
53            exit_status = utils.system(wget_cmd, ignore_status=True)
54            if not exit_status:
55                logging.debug('Able to wget URL.')
56                break
57            logging.error('Could not wget URL; trying again.')
58        grep_url_cmd = 'cat /tmp/wget.log | grep %s' % pattern
59        output_status = utils.system(grep_url_cmd, ignore_status=True)
60        if output_status:
61            logging.debug('Unable to access correct URL for %s',
62                          service_test.testname)
63            return False
64        return True
65
66
67    def _connect(self, ssid, uname):
68        """Connect to particular network and assert access to page.
69
70        @param ssid string - predefined SSID from user's preferred networks
71        @param uname string - predefined username of managed user
72
73        @return boolean - True if able to connect, False otherwise
74
75        """
76        start_time = time.time()
77        with cntc.ChromeNetworkingTestContext(username=uname,
78                  password=constants.PASSWORD, gaia_login=True) as \
79                  testing_context:
80            net_provider = cnta.ChromeNetworkProvider(testing_context)
81            enabled_devices = net_provider.get_enabled_devices()
82            if net_provider.WIFI_DEVICE not in enabled_devices:
83                net_provider.enable_network_device(net_provider.WIFI_DEVICE)
84            logging.info('Scanning for networks')
85            connect_to_service = None
86            while time.time() - start_time < constants.SCAN_RETRY_TIMEOUT:
87                net_provider.scan_for_networks(timeout=20)
88                logging.info('Attempting to connect to %s', ssid)
89                networks = net_provider.get_wifi_networks()
90                for service in networks:
91                    if service['Name'] == ssid:
92                        connect_to_service = service
93                if not connect_to_service:
94                    logging.error('Unable to find %s', ssid)
95                    continue
96                try:
97                    net_provider.connect_to_network(connect_to_service)
98                    logging.info('Successfully connected to network %s', ssid)
99                    return True
100                except error.TestFail as e:
101                    logging.error('Unable to connect to %s', ssid)
102                    continue
103            return False
104
105
106    def _connect_and_assert(self, test, ssid, user):
107        """Verify connect and assert and write results to results/.
108
109        @param test string - testname of NetworkServices namedtuple
110        @param ssid string - predefined SSID from user's preferred networks
111        @param user string - predefined username of managed user
112
113        """
114        tf = tempfile.NamedTemporaryFile(suffix='.txt',
115                                         prefix='connect_%s_' % test,
116                                         dir=self.resultsdir,
117                                         delete=False)
118        with tf as results:
119            if not self._connect(ssid, user):
120                results.write('%s FAILED to connect to SSID\n\n' % test)
121            elif not self._assert_access(test):
122                results.write('%s FAILED to access\n\n' % test)
123            else:
124                results.write('%s passed\n\n' % test)
125
126
127    def _to_wifi(self, proxy):
128        """Set service order to WiFi before Ethernet.
129
130        @param proxy WiFi Proxy object
131
132        """
133        logging.info('Setting order to WiFi, prioritized over Ethernet.')
134        proxy.manager.SetServiceOrder(dbus.String('wifi,ethernet'))
135
136
137    def _to_ethernet(self, proxy):
138        """Set service order to default Ethernet before WiFi
139
140        @param proxy WiFi Proxy object
141
142        """
143        logging.info('Setting back to default service order.')
144        proxy.manager.SetServiceOrder(dbus.String('ethernet,wifi'))
145
146
147    def run_once(self, test):
148        """Run the test.
149
150        @param test string - Set by the client test control file
151
152        """
153        client_proxy = wifi_proxy.WifiProxy()
154        if test is not 'all':
155            logging.info('Running an individual control file.')
156            self._to_wifi(client_proxy)
157            for service_test in constants.NETWORK_SERVICES_TESTS:
158                if service_test.testname == test:
159                    self._connect_and_assert(service_test.testname,
160                                             service_test.ssid,
161                                             service_test.user)
162                    self._to_ethernet(client_proxy)
163                    return
164        for service_test in constants.NETWORK_SERVICES_TESTS:
165            logging.info('==== Running current test %s ====',
166                         service_test.testname)
167            self._to_wifi(client_proxy)
168            self._connect_and_assert(service_test.testname,
169                                     service_test.ssid,
170                                     service_test.user)
171            self._to_ethernet(client_proxy)
172
173        # Ensure DUT returns to normal service state
174        self._to_ethernet(client_proxy)
175