server/site_tests/audio_RebootChameleon/audio_RebootChameleon.py

# Lint as: python2, python3
# Copyright 2023 The ChromiumOS Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""
This is a server side test which reboots chameleon to keep it healthy.
"""

from autotest_lib.client.common_lib import error
from autotest_lib.client.cros.chameleon import chameleon
from autotest_lib.server.cros.audio import audio_test

import logging
import os
import time

WAIT_AFTER_RETRY = 10
MAX_RETRIES = 4

class audio_RebootChameleon(audio_test.AudioTest):
    """Reboot chameleon.

    This test provides a periodic soft-reboot mechanism for chameleon per suite.

    This test will:
    a. reboot chameleon device and restart chameleond
    b. make polling to the chameleon device
    c. recreate the SSH tunneling between drone and chameleon
    d. execute a simple health check to make sure the XMLRPC service is up and running.
    """
    version = 1

    def _reboot_chameleon(self):
        """Reboots chameleon."""
        # Reboot chameleon will break the SSH tunnel between drone and chameleon, so
        # the retry mechanism in ChameleonConnection cannot find chameleon after reboot.
        # Here we ignore those exceptions generated by ChameleonConnection, and later
        # the connection will be recreated and have health checked after reboot.
        try:
            self.host.chameleon.reboot()
        except Exception:
            pass

    def _polling_with_ping(self, connAddr):
        """Polling chameleon with ping."""
        retries = 0
        while retries < MAX_RETRIES:
            response = os.system("ping -c 1 " + connAddr)
            if response == 0:
                logging.info("Chameleon is alive. (attempt #%d)", retries+1)
                break
            logging.info("Chameleon has no response at attempt #%d.", retries+1)
            retries += 1
            logging.info("Wait for %s seconds for next attempt.", WAIT_AFTER_RETRY)
            time.sleep(WAIT_AFTER_RETRY)

        if retries == MAX_RETRIES:
            logging.error("Chameleon has no response after reboot.")
            raise error.TestFail("Chameleon has no response after reboot.")

    def _reconnect(self, connAddr):
        """Recreate the SSH tunnel between drone and chameleon."""
        conn = chameleon.ChameleonConnection(connAddr)
        self.host.chameleon = chameleon.ChameleonBoard(conn)

    def _health_check(self):
        """Make XMLRPC call to chameleon as health check."""
        try:
            return str([p.port_id for p in self.host.chameleon.get_all_ports()])
        except Exception as e:
            raise error.TestFail("Unable to execute health check after reboot: " + str(e))

    def run_once(self):
        """Reboot to maintain chameleon healthness."""
        logging.info("Rebooting Chameleon.")

        if not self.host.chameleon:
            raise error.TestFail("Chameleon Not Found.")

        address = self.host.chameleon.get_network_address()
        logging.info("Chameleon address: %s", address)

        self._reboot_chameleon()
        self._polling_with_ping(address)

        logging.info("Successfully rebooted.")