1# Copyright 2014 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 time 6 7from autotest_lib.client.bin import test 8from autotest_lib.client.bin import utils 9from autotest_lib.client.common_lib import error 10from autotest_lib.client.common_lib.cros.network import apmanager_constants 11from autotest_lib.client.cros import service_stopper 12from autotest_lib.client.cros.networking import apmanager_proxy 13 14class apmanager_CheckAPProcesses(test.test): 15 """Test that required processes are created/terminated when AP service 16 is started/terminated. 17 """ 18 version = 1 19 20 21 POLLING_INTERVAL_SECONDS = 0.2 22 # These services interact with the apmanager in undesirable ways. 23 # For instance, buffet has a bad habit of starting up APs, which 24 # prevents the test from doing likewise. 25 RELATED_SERVICES = ['buffet'] 26 27 28 def _verify_process(self, 29 process_name, 30 expected_running, 31 timeout_seconds=10): 32 """Verify the given process |process_name| is running |running|. 33 34 @param process_name string name of the process 35 @param expected_running bool flag indicating expected process state 36 @param timeout_seconds float number of seconds to wait for the process 37 to transition to the expected state. 38 39 """ 40 endtime = time.time() + timeout_seconds 41 while endtime > time.time(): 42 out = utils.system_output('pgrep %s 2>&1' % process_name, 43 retain_output=True, 44 ignore_status=True).strip() 45 actual_running = (out != '') 46 if expected_running == actual_running: 47 break 48 time.sleep(self.POLLING_INTERVAL_SECONDS) 49 else: 50 raise error.TestFail( 51 'Process %s running status expected %r actual %r' 52 % (process_name, expected_running, actual_running)) 53 54 55 def run_once(self): 56 """Test body.""" 57 with service_stopper.ServiceStopper( 58 services_to_stop=self.RELATED_SERVICES): 59 # AP configuration parameters, only configuring SSID. 60 ap_config = {apmanager_constants.CONFIG_SSID: 'testap'} 61 manager = apmanager_proxy.ApmanagerProxy() 62 service = manager.start_service(ap_config) 63 self._verify_process('hostapd', True) 64 self._verify_process('dnsmasq', True) 65 manager.terminate_service(service) 66 self._verify_process('hostapd', False) 67 self._verify_process('dnsmasq', False) 68