Merge "GdCert: Kill root-canal process if SIGINT failed"

import subprocess

from acts import asserts

from acts import context

from acts.context import get_current_context

from acts.base_test import BaseTestClass

from cert.os_utils import get_gd_root, is_subprocess_alive

from facade import rootservice_pb2 as facade_rootservice

class GdBaseTestClass(BaseTestClass):

    SUBPROCESS_WAIT_TIMEOUT_SECONDS = 10

    def setup_class(self, dut_module, cert_module):

        self.dut_module = dut_module

        self.cert_module = cert_module

        self.log_path_base = get_current_context().get_full_output_path()

        gd_devices = self.controller_configs.get("GdDevice")

        self.log_path_base = context.get_current_context().get_full_output_path(

        )

        # Start root-canal if needed

        self.rootcanal_running = False

        if 'rootcanal' in self.controller_configs:

            self.rootcanal_running = True

            asserts.assert_true(

                is_subprocess_alive(self.rootcanal_process),

                msg="root-canal stopped immediately after running")

            for gd_device in gd_devices:

                gd_device["rootcanal_port"] = rootcanal_hci_port

            # Modify the device config to include the correct root-canal port

            for gd_device_config in self.controller_configs.get("GdDevice"):

                gd_device_config["rootcanal_port"] = rootcanal_hci_port

        # Parse and construct GD device objects

        self.register_controller(

            importlib.import_module('cert.gd_device'), builtin=True)

        self.dut = self.gd_devices[1]

        self.cert = self.gd_devices[0]

    def teardown_class(self):

        if self.rootcanal_running:

            self.rootcanal_process.send_signal(signal.SIGINT)

            rootcanal_return_code = self.rootcanal_process.wait()

            self.rootcanal_logs.close()

            if rootcanal_return_code != 0 and\

                rootcanal_return_code != -signal.SIGINT:

            stop_signal = signal.SIGINT

            self.rootcanal_process.send_signal(stop_signal)

            try:

                return_code = self.rootcanal_process.wait(

                    timeout=self.SUBPROCESS_WAIT_TIMEOUT_SECONDS)

            except subprocess.TimeoutExpired:

                logging.error(

                    "rootcanal stopped with code: %d" % rootcanal_return_code)

                return False

                    "Failed to interrupt root canal via SIGINT, sending SIGKILL"

                )

                stop_signal = signal.SIGKILL

                self.rootcanal_process.kill()

                try:

                    return_code = self.rootcanal_process.wait(

                        timeout=self.SUBPROCESS_WAIT_TIMEOUT_SECONDS)

                except subprocess.TimeoutExpired:

                    logging.error("Failed to kill root canal")

                    return_code = -65536

            if return_code != 0 and return_code != -stop_signal:

                logging.error("rootcanal stopped with code: %d" % return_code)

            self.rootcanal_logs.close()

    def setup_test(self):

        self.dut.rootservice.StartStack(

        self.cert.wait_channel_ready()

    def teardown_test(self):

        self.dut.rootservice.StopStack(facade_rootservice.StopStackRequest())

        self.cert.rootservice.StopStack(facade_rootservice.StopStackRequest())

        self.dut.rootservice.StopStack(facade_rootservice.StopStackRequest())