blob: d578a17ead4e55dae195c9bc65f993117840552a [file] [log] [blame]
# Copyright (c) 2012 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
import logging
import time
from autotest_lib.client.common_lib import error
from autotest_lib.server import site_wifitest
from autotest_lib.server import site_host_route
from autotest_lib.server.cros import stress
class ONCTest(site_wifitest.WiFiTest):
""" This class includes executing the ONC specific commands and is used
in conjunction with WiFiTest for configuring the routers."""
STATE_LIST = {
'STATE_LOGGED_IN': '/tmp/NETWORKONC_logged_in',
'STATE_ONC_SET':'/tmp/NETWORKONC_onc_set',
'STATE_EXIT':'/tmp/NETWORKONC_exit'
}
WAIT_FOR_CLIENT_TIMEOUT = 200
def init_profile(self):
# ONC type tests require chrome which seems to only like the default
# profile. For ONC tests, profile cleanup occurs when running
# the client side test.
pass
def cleanup(self, params={}):
self.client_logout()
super(ONCTest, self).cleanup(params)
def run_onc_client_test(self, params):
if not params.get('test_type'):
params['test_type'] = 'test_simple_set_user_onc'
logging.info('Server: starting client test "%s"' % params['test_type'])
self.client_at.run_test('network_ONC', **params)
def set_user_onc(self, params):
if not params.get('onc'):
params['onc'] = ''
params['test_type'] = 'test_simple_set_user_onc'
# Start the thread that logs into chrome and sets the
# device policy.
self.client_test = stress.CountedStressor(
lambda: self.run_onc_client_test(params),
on_exit=lambda:self._client_logout())
self.client_test.start(1)
def _wait_for_client_state(self, state, timeout=WAIT_FOR_CLIENT_TIMEOUT):
""" Waits till the client's state matches that of state.
This is done via checking a series of state files the client
writes to /tmp. """
ending_time = time.time()+timeout
while time.time() < ending_time:
output = self.client.run('ls %s' % self.STATE_LIST[state],
ignore_status=True)
if output.exit_status == 0:
break
time.sleep(1)
else:
raise error.TestError('TIMEOUT waiting for ONC to be ready')
def connect_wifi_onc(self, params):
""" Connect to the configured AP. """
# Wait for the ONC configuration to be ready before connecting.
self._wait_for_client_state('STATE_ONC_SET')
super(ONCTest, self).connect(params)
def __add_host_route(self, host):
""" Adding local and remote ip route. """
# What is the local address we use to get to the test host?
local_ip = site_host_route.LocalHostRoute(host.ip).route_info["src"]
# How does the test host currently get to this local address?
host_route = site_host_route.RemoteHostRoute(host, local_ip).route_info
# Flatten the returned dict into a single string
route_args = " ".join(" ".join(x) for x in host_route.iteritems())
self.host_route_args[host.ip] = "%s %s" % (local_ip, route_args)
host.run("ip route add %s" % self.host_route_args[host.ip])
def connect_vpn_onc(self, params):
""" Connect to the configured VPN. """
# Wait for the ONC configuration to be ready before connecting.
self._wait_for_client_state('STATE_ONC_SET')
self.vpn_client_kill({}) # Must be first. Relies on self.vpn_kind.
# Starting up the VPN client may cause the DUT's routing table (esp.
# the default route) to change. Set up a host route backwards so
# we don't lose our control connection in that event.
self.__add_host_route(self.client)
# Service is connectable after onc import
result = self.client.run('%s/test/connect-service '
' l2tpipsec-psk ' %
(self.client_cmd_flimflam_lib))
def _client_logout(self):
# To end the client thread, it relies on state exit to be
# present.
self.client.run('touch %s' % self.STATE_LIST['STATE_EXIT'],
ignore_status=True)
def client_logout(self, params=None):
if not hasattr(self, 'client_test'):
return
self._client_logout()
self.client_test.join(self.WAIT_FOR_CLIENT_TIMEOUT)
self.client_test.reraise()