Port GD Cert tests to Blueberry/Mobly

#   See the License for the specific language governing permissions and

#   limitations under the License.

from blueberry.tests.gd.cert.metadata import metadata

from mobly import asserts

from mobly import signals

from mobly import test_runner

from mobly import base_test

from cert.metadata import metadata

from cert.cert_self_test_lib import *

#!/usr/bin/env python3

#

#   Copyright 2018 - The Android Open Source Project

#

#   Licensed under the Apache License, Version 2.0 (the "License");

#   you may not use this file except in compliance with the License.

#   You may obtain a copy of the License at

#

#       http://www.apache.org/licenses/LICENSE-2.0

#

#   Unless required by applicable law or agreed to in writing, software

#   distributed under the License is distributed on an "AS IS" BASIS,

#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

#   See the License for the specific language governing permissions and

#   limitations under the License.

import enum

import logging

import os

class ContextLevel(enum.IntEnum):

    ROOT = 0

    TESTCLASS = 1

    TESTCASE = 2

def get_current_context(depth=None):

    """Get the current test context at the specified depth.

    Pulls the most recently created context, with a level at or below the given

    depth, from the _contexts stack.

    Args:

        depth: The desired context level. For example, the TESTCLASS level would

            yield the current test class context, even if the test is currently

            within a test case.

    Returns: An instance of TestContext.

    """

    if depth is None:

        return _contexts[-1]

    return _contexts[min(depth, len(_contexts) - 1)]

class TestContext(object):

    """An object representing the current context in which a test is executing.

    The context encodes the current state of the test runner with respect to a

    particular scenario in which code is being executed. For example, if some

    code is being executed as part of a test case, then the context should

    encode information about that test case such as its name or enclosing

    class.

    The subcontext specifies a relative path in which certain outputs,

    e.g. logcat, should be kept for the given context.

    The full output path is given by

    <base_output_path>/<context_dir>/<subcontext>.

    Attributes:

        _base_output_paths: a dictionary mapping a logger's name to its base

                            output path

        _subcontexts: a dictionary mapping a logger's name to its

                      subcontext-level output directory

    """

    _base_output_paths = {}

    _subcontexts = {}

    def get_base_output_path(self, log_name=None):

        """Gets the base output path for this logger.

        The base output path is interpreted as the reporting root for the

        entire test runner.

        If a path has been added with add_base_output_path, it is returned.

        Otherwise, a default is determined by _get_default_base_output_path().

        Args:

            log_name: The name of the logger.

        Returns:

            The output path.

        """

        if log_name in self._base_output_paths:

            return self._base_output_paths[log_name]

        return self._get_default_base_output_path()

    def get_subcontext(self, log_name=None):

        """Gets the subcontext for this logger.

        The subcontext is interpreted as the directory, relative to the

        context-level path, where all outputs of the given logger are stored.

        If a path has been added with add_subcontext, it is returned.

        Otherwise, the empty string is returned.

        Args:

            log_name: The name of the logger.

        Returns:

            The output path.

        """

        return self._subcontexts.get(log_name, '')

    def get_full_output_path(self, log_name=None):

        """Gets the full output path for this context.

        The full path represents the absolute path to the output directory,

        as given by <base_output_path>/<context_dir>/<subcontext>

        Args:

            log_name: The name of the logger. Used to specify the base output

                      path and the subcontext.

        Returns:

            The output path.

        """

        path = os.path.join(

            self.get_base_output_path(log_name), self._get_default_context_dir(), self.get_subcontext(log_name))

        os.makedirs(path, exist_ok=True)

        return path

    def _get_default_base_output_path(self):

        """Gets the default base output path.

        This will attempt to use logging path set up in the global

        logger.

        Returns:

            The logging path.

        Raises:

            EnvironmentError: If logger has not been initialized.

        """

        try:

            return logging.log_path

        except AttributeError as e:

            raise EnvironmentError('The Mobly logger has not been set up and'

                                   ' "base_output_path" has not been set.') from e

    def _get_default_context_dir(self):

        """Gets the default output directory for this context."""

        raise NotImplementedError()

class RootContext(TestContext):

    """A TestContext that represents a test run."""

    @property

    def identifier(self):

        return 'root'

    def _get_default_context_dir(self):

        """Gets the default output directory for this context.

        Logs at the root level context are placed directly in the base level

        directory, so no context-level path exists."""

        return ''

# stack for keeping track of the current test context

_contexts = [RootContext()]

#!/usr/bin/env python3

#

#   Copyright 2019 - The Android Open Source Project

#

#   Licensed under the Apache License, Version 2.0 (the "License");

#   you may not use this file except in compliance with the License.

#   You may obtain a copy of the License at

#

#       http://www.apache.org/licenses/LICENSE-2.0

#

#   Unless required by applicable law or agreed to in writing, software

#   distributed under the License is distributed on an "AS IS" BASIS,

#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

#   See the License for the specific language governing permissions and

#   limitations under the License.

import importlib

import logging

import traceback

from functools import wraps

from grpc import RpcError

from cert.gd_base_test_lib import setup_class_core

from cert.gd_base_test_lib import teardown_class_core

from cert.gd_base_test_lib import setup_test_core

from cert.gd_base_test_lib import teardown_test_core

from cert.gd_base_test_lib import dump_crashes_core

from blueberry.tests.gd.cert.context import get_current_context

from blueberry.tests.gd.cert.gd_device import MOBLY_CONTROLLER_CONFIG_NAME as CONTROLLER_CONFIG_NAME

from blueberry.tests.gd.cert.tracelogger import TraceLogger

from mobly import asserts, signals

from mobly import base_test

class GdBaseTestClass(base_test.BaseTestClass):

    SUBPROCESS_WAIT_TIMEOUT_SECONDS = 10

    def setup_class(self, dut_module, cert_module):

        self.log = TraceLogger(logging.getLogger())

        self.log_path_base = get_current_context().get_full_output_path()

        self.verbose_mode = bool(self.user_params.get('verbose_mode', False))

        for config in self.controller_configs[CONTROLLER_CONFIG_NAME]:

            config['verbose_mode'] = self.verbose_mode

        self.info = setup_class_core(

            dut_module=dut_module,

            cert_module=cert_module,

            verbose_mode=self.verbose_mode,

            log_path_base=self.log_path_base,

            controller_configs=self.controller_configs)

        self.dut_module = self.info['dut_module']

        self.cert_module = self.info['cert_module']

        self.rootcanal_running = self.info['rootcanal_running']

        self.rootcanal_logpath = self.info['rootcanal_logpath']

        self.rootcanal_process = self.info['rootcanal_process']

        self.rootcanal_logger = self.info['rootcanal_logger']

        if 'rootcanal' in self.controller_configs:

            asserts.assert_true(self.info['rootcanal_exist'],

                                "Root canal does not exist at %s" % self.info['rootcanal'])

            asserts.assert_true(self.info['make_rootcanal_ports_available'],

                                "Failed to make root canal ports available")

            self.log.debug("Running %s" % " ".join(self.info['rootcanal_cmd']))

            asserts.assert_true(

                self.info['is_rootcanal_process_started'],

                msg="Cannot start root-canal at " + str(self.info['rootcanal']))

            asserts.assert_true(self.info['is_subprocess_alive'], msg="root-canal stopped immediately after running")

            self.controller_configs = self.info['controller_configs']

        # Parse and construct GD device objects

        self.register_controller(importlib.import_module('blueberry.tests.gd.cert.gd_device'), builtin=True)

        self.dut = self.gd_device[1]

        self.cert = self.gd_device[0]

    def teardown_class(self):

        teardown_class_core(

            rootcanal_running=self.rootcanal_running,

            rootcanal_process=self.rootcanal_process,

            rootcanal_logger=self.rootcanal_logger,

            subprocess_wait_timeout_seconds=self.SUBPROCESS_WAIT_TIMEOUT_SECONDS)

    def setup_test(self):

        setup_test_core(dut=self.dut, cert=self.cert, dut_module=self.dut_module, cert_module=self.cert_module)

    def teardown_test(self):

        teardown_test_core(cert=self.cert, dut=self.dut)

    @staticmethod

    def get_module_reference_name(a_module):

        """Returns the module's module's submodule name as reference name.

        Args:

            a_module: Any module. Ideally, a controller module.

        Returns:

            A string corresponding to the module's name.

        """

        return a_module.__name__.split('.')[-1]

    def register_controller(self, controller_module, required=True, builtin=False):

        """Registers an controller module for a test class. Invokes Mobly's

        implementation of register_controller.

        """

        module_ref_name = self.get_module_reference_name(controller_module)

        module_config_name = controller_module.MOBLY_CONTROLLER_CONFIG_NAME

        # Get controller objects from Mobly's register_controller

        controllers = self._controller_manager.register_controller(controller_module, required=required)

        if not controllers:

            return None

        # Log controller information

        # Implementation of "get_info" is optional for a controller module.

        if hasattr(controller_module, "get_info"):

            controller_info = controller_module.get_info(controllers)

            self.log.info("Controller %s: %s", module_config_name, controller_info)

        if builtin:

            setattr(self, module_ref_name, controllers)

        return controllers

    def __getattribute__(self, name):

        attr = super().__getattribute__(name)

        if not callable(attr) or not GdBaseTestClass.__is_entry_function(name):

            return attr

        @wraps(attr)

        def __wrapped(*args, **kwargs):

            try:

                return attr(*args, **kwargs)

            except RpcError as e:

                exception_info = "".join(traceback.format_exception(e.__class__, e, e.__traceback__))

                raise signals.TestFailure(

                    "RpcError during test\n\nRpcError:\n\n%s\n%s" % (exception_info, self.__dump_crashes()))

        return __wrapped

    __ENTRY_METHODS = {"setup_class", "teardown_class", "setup_test", "teardown_test"}

    @staticmethod

    def __is_entry_function(name):

        return name.startswith("test_") or name in GdBaseTestClass.__ENTRY_METHODS

    def __dump_crashes(self):

        """

        return: formatted stack traces if found, or last few lines of log

        """

        crash_detail = dump_crashes_core(

            dut=self.dut,

            cert=self.cert,

            rootcanal_running=self.rootcanal_running,

            rootcanal_process=self.rootcanal_process,

            rootcanal_logpath=self.rootcanal_logpath)

        return crash_detail

#!/usr/bin/env python3

#

#   Copyright 2019 - The Android Open Source Project

#

#   Licensed under the Apache License, Version 2.0 (the "License");

#   you may not use this file except in compliance with the License.

#   You may obtain a copy of the License at

#

#       http://www.apache.org/licenses/LICENSE-2.0

#

#   Unless required by applicable law or agreed to in writing, software

#   distributed under the License is distributed on an "AS IS" BASIS,

#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

#   See the License for the specific language governing permissions and

#   limitations under the License.

import functools

import inspect

from mobly import asserts

from blueberry.tests.gd.cert.test_decorators import test_info

def _fail_decorator(msg):

    def fail_decorator(func):

        @functools.wraps(func)

        def fail(*args, **kwargs):

            asserts.fail(msg)

        return fail

    return fail_decorator

def metadata(_do_not_use=None, pts_test_id=None, pts_test_name=None):

    """

    Record a piece of test metadata in the Extra section of the test Record in

    the test summary file. The metadata will come with a timestamp, but there

    is no guarantee on the order of when the metadata will be written

    Note:

    - Metadata is recorded per test case as key-value pairs.

    - Metadata is only guaranteed to be written when the test result is PASS,

      FAIL or SKIPPED. When there are test infrastructural errors, metadata

      might not be written successfully

    :param _do_not_use: a positional argument with default value. This argument

                        is to ensure that @metadata(key=value) is used in a

                        functional form instead of @metadata or @metadata(a)

    :param pts_test_id: A fully qualified PTS test ID such as

                        L2CAP/COS/IEX/BV-01-C

    :param pts_test_name: A human readable test name such as

                          "Request Connection" for the above example

    :return: decorated test case function object

    """

    if _do_not_use is not None:

        def fail(*args, **kwargs):

            asserts.fail("@metadata must be used in functional form such " "as @metadta(key=value)")

        return fail

    # Create a dictionary of optional parameters

    values = locals()

    args = {arg: values[arg] for arg in inspect.getfullargspec(metadata).args}

    del args["_do_not_use"]

    # Check if at least one optional parameter is valid

    if not any(args.values()):

        return _fail_decorator("at least one optional argument should be valid")

    # Validate pts_test_id and pts_test_name

    if any((pts_test_id, pts_test_name)) and \

            not all((pts_test_id, pts_test_name)):

        return _fail_decorator("pts_test_id and pts_test_name must both " "be valid if one of them is valid")

    return test_info(**args)