client/site_tests/apmanager_CheckAPProcesses/apmanager_CheckAPProcesses.py

# Copyright 2014 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 time

from autotest_lib.client.bin import test
from autotest_lib.client.bin import utils
from autotest_lib.client.common_lib import error
from autotest_lib.client.common_lib.cros.network import apmanager_constants
from autotest_lib.client.cros import service_stopper
from autotest_lib.client.cros.networking import apmanager_proxy

class apmanager_CheckAPProcesses(test.test):
    """Test that required processes are created/terminated when AP service
    is started/terminated.
    """
    version = 1


    POLLING_INTERVAL_SECONDS = 0.2
    # These services interact with the apmanager in undesirable ways.
    # For instance, privetd has a bad habit of starting up APs, which
    # prevents the test from doing likewise.
    RELATED_SERVICES = ['privetd']


    def _verify_process(self,
                        process_name,
                        expected_running,
                        timeout_seconds=10):
        """Verify the given process |process_name| is running |running|.

        @param process_name string name of the process
        @param expected_running bool flag indicating expected process state
        @param timeout_seconds float number of seconds to wait for the process
               to transition to the expected state.

        """
        endtime = time.time() + timeout_seconds
        while endtime > time.time():
            out = utils.system_output('pgrep %s 2>&1' % process_name,
                                      retain_output=True,
                                      ignore_status=True).strip()
            actual_running = (out != '')
            if expected_running == actual_running:
                break
            time.sleep(self.POLLING_INTERVAL_SECONDS)
        else:
            raise error.TestFail(
                    'Process %s running status expected %r actual %r'
                    % (process_name, expected_running, actual_running))


    def run_once(self):
        """Test body."""
        with service_stopper.ServiceStopper(
                services_to_stop=self.RELATED_SERVICES):
            # AP configuration parameters, only configuring SSID.
            ap_config = {apmanager_constants.CONFIG_SSID: 'testap'}
            manager = apmanager_proxy.ApmanagerProxy()
            service = manager.start_service(ap_config)
            self._verify_process('hostapd', True)
            self._verify_process('dnsmasq', True)
            manager.terminate_service(service)
            self._verify_process('hostapd', False)
            self._verify_process('dnsmasq', False)