| """Step class to run tests against a local x86 cast shell.""" |
| |
| from __future__ import absolute_import |
| import datetime |
| import glob |
| import json |
| import logging |
| import os |
| import time |
| import traceback |
| |
| from helpers import cast_shell_utils |
| from helpers import py_env_utils |
| from slave import base_step |
| |
| _FRAGGLE_ROCK_CONFIG_FILES_ENV_VAR = 'FRAGGLE_ROCK_CONFIG_FILES' |
| _LOG_DIR = 'integration_test_logs' |
| |
| |
| # pylint: disable=method-hidden |
| |
| |
| class RunIntegrationTestStep(base_step.BaseStep): |
| """Step for running integration test against a local x86 cast shell.""" |
| |
| def __init__(self, test_case_class, test_case_module, test_sequence_class, |
| test_sequence_module, test_config_file_path, |
| device_class='LocalX86CastShellDevice', |
| device_module='eurtest.device.local_x86_cast_shell_device', |
| assistant_enabled=False, |
| headless=False, |
| debug_output=True, |
| **kwargs): |
| """Creates a RunIntegrationTestStep instance. |
| |
| Args: |
| test_case_class: A list of TestCase implementations to run. Each TestCase |
| class is expected to be defined in the test_case_module. |
| test_case_module: Module containing TestCase implementation to run. |
| test_sequence_class: TestSequence implementation to run. |
| test_sequence_module: Module containing TestSequence implementation to |
| run. |
| test_config_file_path: Dot-notation path to the test's config file |
| relative to config_file_root_dir. |
| device_class: A string with the name of Python class that represents the |
| device. |
| device_module: Dot-notation path to the module containing |device_class|. |
| assistant_enabled: Whether the x86 cast_shell against which tests will |
| be run is assistant-enabled (in which case, some special flags are |
| specified with the cast_shell is started). |
| headless: Whether the x86 cast_shell uses headless ozone platform. |
| debug_output: Log debug output from the test. |
| **kwargs: Any additional args to pass to BaseStep. |
| """ |
| base_step.BaseStep.__init__(self, name='Run integration tests', **kwargs) |
| self._test_case_class = test_case_class |
| self._test_case_module = test_case_module |
| self._test_sequence_class = test_sequence_class |
| self._test_sequence_module = test_sequence_module |
| self._test_config_file_path = test_config_file_path |
| self._device_class = device_class |
| self._device_module = device_module |
| self._device_ip = '127.0.0.1' |
| self._config_file_root_dir = os.path.join( |
| self.get_project_path('continuous-tests'), 'lab_system', 'master', |
| 'config') |
| self._assistant_enabled = assistant_enabled |
| self._headless = headless |
| self._debug_output = debug_output |
| |
| def _remove_existing_results_json(self): |
| """Remove any results JSON files from previous runs.""" |
| previous_results = glob.glob(os.path.join(os.getcwd(), '*results.json')) |
| for result_found in previous_results: |
| os.remove(result_found) |
| |
| def _execute_tests(self, venv_root): |
| """Execute the integration tests using eurtest.framework.test_runner. |
| |
| Args: |
| venv_root: Path to virtual environment root directory. |
| |
| Returns: |
| A Tuple of (a dict representing a review object, |
| a boolean indicating if tests passed) |
| """ |
| self._remove_existing_results_json() |
| |
| short_wait_secs = 5 |
| env = { |
| _FRAGGLE_ROCK_CONFIG_FILES_ENV_VAR: self._config_file_root_dir, |
| 'PYTHONPATH': self._get_python_path() |
| } |
| runner = os.path.abspath( |
| os.path.join(self.get_project_path('test'), |
| 'eurtest', 'framework', 'test_runner.py')) |
| run_info = {} |
| run_info['start_timestamp'] = datetime.datetime.now().isoformat() |
| run_info['branch_name'] = self.manifest_branch |
| run_info['platform'] = 'clang_x86' |
| cmd = [ |
| runner, |
| '--test_case_class', ','.join(self._test_case_class), |
| '--test_case_module', self._test_case_module, |
| '--test_sequence_class', self._test_sequence_class, |
| '--test_sequence_module', self._test_sequence_module, |
| '--target_device_ethernet_ip', self._device_ip, |
| '--target_device_wifi_ip', self._device_ip, |
| '--device_class', self._device_class, |
| '--device_module', self._device_module, |
| '--run_information_json', json.dumps(run_info), |
| '--test_config_file_path', self._test_config_file_path, |
| '--nouse_tunnel' |
| ] |
| if self._debug_output: |
| cmd.append('--debug') |
| returncode, stdout, stderr = py_env_utils.exec_within_virtual_env( |
| self, venv_root, cmd, env=env) |
| if returncode != 0: |
| logging.error('Failed to run tests: %s\n%s', stdout, stderr) |
| |
| review = ({'message': 'no result found'}, False) |
| # A JSON file will be saved to results_path directory within a few seconds |
| time.sleep(short_wait_secs) |
| results = glob.glob(os.path.join(os.getcwd(), '*results.json')) |
| if results: |
| # Assumption: there should be only one results JSON file |
| review = self.generate_review(results[0]) |
| return review |
| |
| def generate_review(self, results): |
| """Generate a review object from integration test results. |
| |
| Args: |
| results: File path to result JSON file. |
| |
| Returns: |
| A Tuple of (a dict representing a review object, |
| a boolean indicating if tests passed) |
| """ |
| passing = True |
| result_messages = [] |
| results_data = [] |
| summary = 'Suites: {}, Run: {}, Errors: {}, Failures: {}' |
| with open(results) as results_json: |
| json_output = json.load(results_json) |
| if isinstance(json_output, list): |
| results_data.extend(json_output) |
| else: |
| results_data.append(json_output) |
| for result in results_data: |
| suite_count = 0 |
| test_count = 0 |
| failure_count = 0 |
| error_count = 0 |
| for suite in result['test_suites']: |
| suite_count += 1 |
| test_count += suite['test_count'] |
| failure_count += suite['failure_count'] |
| error_count += suite['error_count'] |
| if failure_count + error_count > 0: |
| message = ('Test: {}, {}, Time Taken(sec): {:.2f}').format( |
| result['test_name'], |
| summary.format(suite_count, test_count, error_count, failure_count), |
| result['elapsed_secs']) |
| result_messages.append(message) |
| passing = False |
| if passing: |
| status = 'Looks Good. All tests passed.' |
| else: |
| status = 'Cast compliance tests failed. See the build logs for details.' |
| review = { |
| 'message': ('{} \n{}'.format(status, '\n'.join(result_messages))) |
| } |
| return review, passing |
| |
| def _get_python_path(self): |
| """Returns the PYTHONPATH to execute tests with.""" |
| projects = ['test', 'continuous-tests'] |
| python_paths = [ |
| os.path.abspath(self.get_project_path(project)) for project in projects] |
| return os.pathsep.join(python_paths) |
| |
| def run(self): |
| """Starts a local x86 cast shell and run integration tests against it. |
| |
| Returns: |
| True iff all tests pass. |
| """ |
| log_fd = None |
| log_file_path = None |
| try: |
| # Launch cast_shell |
| _, log_fd, log_file_path = cast_shell_utils.start_cast_shell( |
| self, |
| self.get_project_path('test'), |
| self.get_step_data(cast_shell_utils.CAST_SHELL_EXE_PATH), |
| self.get_tmp_dir(), |
| enable_cma_media_pipeline=True, |
| use_headless_ozone_platform=( |
| self._headless or self._assistant_enabled), |
| launch_utility_process=self._assistant_enabled) |
| venv_root = self.get_step_data('virtual_env_root') |
| review_message, is_passing = self._execute_tests(venv_root) |
| except: # pylint: disable=bare-except |
| review_message = { |
| 'message': ('Cast compliance test setup failed.\n{}'.format( |
| traceback.format_exc())) |
| } |
| is_passing = False |
| |
| if log_fd and log_file_path: |
| cast_shell_utils.upload_cast_shell_logs(log_fd, |
| log_file_path, |
| self.get_gcs_dir(), _LOG_DIR, |
| 'cast_shell_console.txt') |
| if not is_passing: |
| self.add_review(review_message) |
| else: |
| logging.info(review_message.get('message')) |
| return is_passing |
