1#!/usr/bin/env python3
2#
3#   Copyright 2019 - The Android Open Source Project
4#
5#   Licensed under the Apache License, Version 2.0 (the "License");
6#   you may not use this file except in compliance with the License.
7#   You may obtain a copy of the License at
8#
9#       http://www.apache.org/licenses/LICENSE-2.0
10#
11#   Unless required by applicable law or agreed to in writing, software
12#   distributed under the License is distributed on an "AS IS" BASIS,
13#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14#   See the License for the specific language governing permissions and
15#   limitations under the License.
16
17import importlib
18import logging
19import os
20import signal
21import subprocess
22import traceback
23
24from functools import wraps
25from grpc import RpcError
26
27from acts import asserts, signals
28from acts.context import get_current_context
29from acts.base_test import BaseTestClass
30
31from cert.async_subprocess_logger import AsyncSubprocessLogger
32from cert.os_utils import get_gd_root
33from cert.os_utils import read_crash_snippet_and_log_tail
34from cert.os_utils import is_subprocess_alive
35from cert.os_utils import make_ports_available
36from cert.os_utils import TerminalColor
37from cert.gd_device import MOBLY_CONTROLLER_CONFIG_NAME as CONTROLLER_CONFIG_NAME
38from facade import rootservice_pb2 as facade_rootservice
39from cert.gd_base_test_lib import setup_class_core
40from cert.gd_base_test_lib import teardown_class_core
41from cert.gd_base_test_lib import setup_test_core
42from cert.gd_base_test_lib import teardown_test_core
43from cert.gd_base_test_lib import dump_crashes_core
44
45
46class GdBaseTestClass(BaseTestClass):
47
48    SUBPROCESS_WAIT_TIMEOUT_SECONDS = 10
49
50    def setup_class(self, dut_module, cert_module):
51        self.log_path_base = get_current_context().get_full_output_path()
52        self.verbose_mode = bool(self.user_params.get('verbose_mode', False))
53        for config in self.controller_configs[CONTROLLER_CONFIG_NAME]:
54            config['verbose_mode'] = self.verbose_mode
55
56        self.info = setup_class_core(
57            dut_module=dut_module,
58            cert_module=cert_module,
59            verbose_mode=self.verbose_mode,
60            log_path_base=self.log_path_base,
61            controller_configs=self.controller_configs)
62        self.dut_module = self.info['dut_module']
63        self.cert_module = self.info['cert_module']
64        self.rootcanal_running = self.info['rootcanal_running']
65        self.rootcanal_logpath = self.info['rootcanal_logpath']
66        self.rootcanal_process = self.info['rootcanal_process']
67
68        if 'rootcanal' in self.controller_configs:
69            asserts.assert_true(self.info['rootcanal_exist'],
70                                "Root canal does not exist at %s" % self.info['rootcanal'])
71            asserts.assert_true(self.info['make_rootcanal_ports_available'],
72                                "Failed to make root canal ports available")
73
74            self.log.debug("Running %s" % " ".join(self.info['rootcanal_cmd']))
75            asserts.assert_true(
76                self.info['is_rootcanal_process_started'],
77                msg="Cannot start root-canal at " + str(self.info['rootcanal']))
78            asserts.assert_true(self.info['is_subprocess_alive'], msg="root-canal stopped immediately after running")
79
80            self.rootcanal_logger = self.info['rootcanal_logger']
81            self.controller_configs = self.info['controller_configs']
82
83        # Parse and construct GD device objects
84        self.register_controller(importlib.import_module('cert.gd_device'), builtin=True)
85        self.dut = self.gd_devices[1]
86        self.cert = self.gd_devices[0]
87
88    def teardown_class(self):
89        teardown_class_core(
90            rootcanal_running=self.rootcanal_running,
91            rootcanal_process=self.rootcanal_process,
92            rootcanal_logger=self.rootcanal_logger,
93            subprocess_wait_timeout_seconds=self.SUBPROCESS_WAIT_TIMEOUT_SECONDS)
94
95    def setup_test(self):
96        setup_test_core(dut=self.dut, cert=self.cert, dut_module=self.dut_module, cert_module=self.cert_module)
97
98    def teardown_test(self):
99        teardown_test_core(cert=self.cert, dut=self.dut)
100
101    def __getattribute__(self, name):
102        attr = super().__getattribute__(name)
103        if not callable(attr) or not GdBaseTestClass.__is_entry_function(name):
104            return attr
105
106        @wraps(attr)
107        def __wrapped(*args, **kwargs):
108            try:
109                return attr(*args, **kwargs)
110            except RpcError as e:
111                exception_info = "".join(traceback.format_exception(e.__class__, e, e.__traceback__))
112                raise signals.TestFailure(
113                    "RpcError during test\n\nRpcError:\n\n%s\n%s" % (exception_info, self.__dump_crashes()))
114
115        return __wrapped
116
117    __ENTRY_METHODS = {"setup_class", "teardown_class", "setup_test", "teardown_test"}
118
119    @staticmethod
120    def __is_entry_function(name):
121        return name.startswith("test_") or name in GdBaseTestClass.__ENTRY_METHODS
122
123    def __dump_crashes(self):
124        """
125        return: formatted stack traces if found, or last few lines of log
126        """
127        crash_detail = dump_crashes_core(
128            dut=self.dut,
129            cert=self.cert,
130            rootcanal_running=self.rootcanal_running,
131            rootcanal_process=self.rootcanal_process,
132            rootcanal_logpath=self.rootcanal_logpath)
133        return crash_detail
134