| """Step for running Sprockets protocol conformance tests. |
| |
| See go/sprockets-eng and go/stl-manual for more information. |
| """ |
| |
| from __future__ import absolute_import |
| import collections |
| import json |
| import logging |
| import os |
| |
| from helpers import cast_shell_utils |
| from slave import base_step |
| |
| _SprocketsTestConfig = collections.namedtuple('_SprocketsTestConfig', |
| 'project, config') |
| _SPROCKETS_PROJECT = 'google/sprockets' |
| _SPROCKETS_TEST_DRIVER = 'test_driver.py' |
| _SPROCKETS_TEST_CONFIGS = [ |
| _SprocketsTestConfig(project='test', config='sprockets_tests/tests.json') |
| ] |
| |
| _CAST_SHELL_FAILURE_MSG = 'Failed to start up and communicate with cast_shell.' |
| _SPROCKETS_TEST_FAILURE_MSG = ( |
| 'Sprockets test failed. See the build log for details.') |
| |
| |
| class SprocketsConformanceTestStep(base_step.BaseStep): |
| """Step to run Sprockets tests.""" |
| |
| def __init__(self, **kwargs): |
| """Constructor.""" |
| base_step.BaseStep.__init__( |
| self, name='sprockets conformance tests', directory='', **kwargs) |
| |
| def read_config(self, config_file_path): |
| if not os.path.exists(config_file_path): |
| logging.info('%s does not exist. Skipping.', config_file_path) |
| return [] |
| with open(config_file_path) as config_json: |
| return json.load(config_json) |
| |
| def run(self): |
| """Returns which commands to be run. Can be overridden by a subclass. |
| |
| Returns: |
| List of ShellCommand objects to execute. |
| """ |
| try: |
| _, log_descriptor, log_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(), |
| use_headless_ozone_platform=True, |
| enable_ccs_transport_v2=True) |
| except (cast_shell_utils.CastShellError, |
| cast_shell_utils.CastShellTimeoutError): |
| logging.exception('Failed to start cast_shell') |
| self.add_review({'message': _CAST_SHELL_FAILURE_MSG}) |
| return False |
| |
| sprockets_path = self.get_project_path(_SPROCKETS_PROJECT) |
| sprockets_test_runner = os.path.join(sprockets_path, _SPROCKETS_TEST_DRIVER) |
| |
| success = True |
| for config in _SPROCKETS_TEST_CONFIGS: |
| config_path_root = self.get_project_path(config.project) |
| config_file_path = os.path.join(config_path_root, config.config) |
| for test in self.read_config(config_file_path): |
| test_path_root = self.get_project_path(test['project']) |
| test_manifest = os.path.join(test_path_root, test['manifest']) |
| test_command = ['python3', sprockets_test_runner, test_manifest] |
| test_command.extend(test['args']) |
| returncode, _, _ = self.exec_subprocess(test_command) |
| success = (returncode == 0) and success |
| |
| if log_descriptor and log_path: |
| cast_shell_utils.upload_cast_shell_logs(log_descriptor, log_path, |
| self.get_gcs_dir(), |
| 'sprockets_conformance_tests', |
| 'cast_shell.logcat') |
| if not success: |
| self.add_review({ |
| 'message': _SPROCKETS_TEST_FAILURE_MSG |
| }) |
| |
| return success |
