#!/usr/bin/env -S python3 -B # Copyright (c) 2021 Project CHIP Authors # # 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 import sys import time import typing from dataclasses import dataclass, field import chiptest import click import coloredlogs from chiptest.accessories import AppsRegister from chiptest.glob_matcher import GlobMatcher from chiptest.test_definition import TestRunTime, TestTag from yaml.paths_finder import PathsFinder DEFAULT_CHIP_ROOT = os.path.abspath( os.path.join(os.path.dirname(__file__), '..', '..')) class ManualHandling(enum.Enum): INCLUDE = enum.auto() SKIP = enum.auto() ONLY = enum.auto() # Supported log levels, mapping string values required for argument # parsing into logging constants __LOG_LEVELS__ = { 'debug': logging.DEBUG, 'info': logging.INFO, 'warn': logging.WARN, 'fatal': logging.FATAL, } @dataclass class RunContext: root: str tests: typing.List[chiptest.TestDefinition] in_unshare: bool chip_tool: str dry_run: bool runtime: TestRunTime # If not empty, include only the specified test tags include_tags: set(TestTag) = field(default_factory={}) # If not empty, exclude tests tagged with these tags exclude_tags: set(TestTag) = field(default_factory={}) @click.group(chain=True) @click.option( '--log-level', default='info', type=click.Choice(__LOG_LEVELS__.keys(), case_sensitive=False), help='Determines the verbosity of script output.') @click.option( '--dry-run', default=False, is_flag=True, help='Only print out shell commands that would be executed') @click.option( '--target', default=['all'], multiple=True, help='Test to run (use "all" to run all tests)' ) @click.option( '--target-glob', default='', help='What targets to accept (glob)' ) @click.option( '--target-skip-glob', default='', help='What targets to skip (glob)' ) @click.option( '--no-log-timestamps', default=False, is_flag=True, help='Skip timestaps in log output') @click.option( '--root', default=DEFAULT_CHIP_ROOT, help='Default directory path for CHIP. Used to copy run configurations') @click.option( '--internal-inside-unshare', hidden=True, is_flag=True, default=False, help='Internal flag for running inside a unshared environment' ) @click.option( '--include-tags', type=click.Choice(TestTag.__members__.keys(), case_sensitive=False), multiple=True, help='What test tags to include when running. Equivalent to "exlcude all except these" for priority purpuses.', ) @click.option( '--exclude-tags', type=click.Choice(TestTag.__members__.keys(), case_sensitive=False), multiple=True, help='What test tags to exclude when running. Exclude options takes precedence over include.', ) @click.option( '--runner', type=click.Choice(['codegen', 'chip_repl_python', 'chip_tool_python'], case_sensitive=False), default='codegen', help='Run YAML tests using the specified runner.') @click.option( '--chip-tool', help='Binary path of chip tool app to use to run the test') @click.pass_context def main(context, dry_run, log_level, target, target_glob, target_skip_glob, no_log_timestamps, root, internal_inside_unshare, include_tags, exclude_tags, runner, chip_tool): # Ensures somewhat pretty logging of what is going on log_fmt = '%(asctime)s.%(msecs)03d %(levelname)-7s %(message)s' if no_log_timestamps: log_fmt = '%(levelname)-7s %(message)s' coloredlogs.install(level=__LOG_LEVELS__[log_level], fmt=log_fmt) runtime = TestRunTime.CHIP_TOOL_BUILTIN if runner == 'chip_repl_python': runtime = TestRunTime.CHIP_REPL_PYTHON elif runner == 'chip_tool_python': runtime = TestRunTime.CHIP_TOOL_PYTHON elif chip_tool is not None and os.path.basename(chip_tool) == "darwin-framework-tool": runtime = TestRunTime.DARWIN_FRAMEWORK_TOOL_BUILTIN if chip_tool is None and not runtime == TestRunTime.CHIP_REPL_PYTHON: # non yaml tests REQUIRE chip-tool. Yaml tests should not require chip-tool paths_finder = PathsFinder() chip_tool = paths_finder.get('chip-tool') if include_tags: include_tags = set([TestTag.__members__[t] for t in include_tags]) if exclude_tags: exclude_tags = set([TestTag.__members__[t] for t in exclude_tags]) # Figures out selected test that match the given name(s) if runtime == TestRunTime.CHIP_REPL_PYTHON: all_tests = [test for test in chiptest.AllReplYamlTests()] elif runtime == TestRunTime.CHIP_TOOL_PYTHON and os.path.basename(chip_tool) != "darwin-framework-tool": all_tests = [test for test in chiptest.AllChipToolYamlTests()] else: all_tests = [test for test in chiptest.AllChipToolTests(chip_tool)] tests = all_tests # If just defaults specified, do not run manual and in development # Specific target basically includes everything if 'all' in target and not include_tags and not exclude_tags: exclude_tags = { TestTag.MANUAL, TestTag.IN_DEVELOPMENT, TestTag.FLAKY, TestTag.EXTRA_SLOW, TestTag.PURPOSEFUL_FAILURE, } if runtime != TestRunTime.CHIP_TOOL_PYTHON: exclude_tags.add(TestTag.CHIP_TOOL_PYTHON_ONLY) if 'all' not in target: tests = [] for name in target: targeted = [test for test in all_tests if test.name.lower() == name.lower()] if len(targeted) == 0: logging.error("Unknown target: %s" % name) tests.extend(targeted) if target_glob: matcher = GlobMatcher(target_glob.lower()) tests = [test for test in tests if matcher.matches(test.name.lower())] if len(tests) == 0: logging.error("No targets match, exiting.") logging.error("Valid targets are (case-insensitive): %s" % (", ".join(test.name for test in all_tests))) exit(1) if target_skip_glob: matcher = GlobMatcher(target_skip_glob.lower()) tests = [test for test in tests if not matcher.matches( test.name.lower())] tests.sort(key=lambda x: x.name) context.obj = RunContext(root=root, tests=tests, in_unshare=internal_inside_unshare, chip_tool=chip_tool, dry_run=dry_run, runtime=runtime, include_tags=include_tags, exclude_tags=exclude_tags) @main.command( 'list', help='List available test suites') @click.pass_context def cmd_list(context): for test in context.obj.tests: tags = test.tags_str() if tags: tags = f" ({tags})" print("%s%s" % (test.name, tags)) @main.command( 'run', help='Execute the tests') @click.option( '--iterations', default=1, help='Number of iterations to run') @click.option( '--all-clusters-app', help='what all clusters app to use') @click.option( '--lock-app', help='what lock app to use') @click.option( '--ota-provider-app', help='what ota provider app to use') @click.option( '--ota-requestor-app', help='what ota requestor app to use') @click.option( '--tv-app', help='what tv app to use') @click.option( '--bridge-app', help='what bridge app to use') @click.option( '--chip-repl-yaml-tester', help='what python script to use for running yaml tests using chip-repl as controller') @click.option( '--chip-tool-with-python', help='what python script to use for running yaml tests using chip-tool as controller') @click.option( '--pics-file', type=click.Path(exists=True), default="src/app/tests/suites/certification/ci-pics-values", show_default=True, help='PICS file to use for test runs.') @click.option( '--keep-going', is_flag=True, default=False, show_default=True, help='Keep running the rest of the tests even if a test fails.') @click.option( '--test-timeout-seconds', default=None, type=int, help='If provided, fail if a test runs for longer than this time') @click.option( '--expected-failures', type=int, default=0, show_default=True, help='Number of tests that are expected to fail in each iteration. Overall test will pass if the number of failures matches this. Nonzero values require --keep-going') @click.pass_context def cmd_run(context, iterations, all_clusters_app, lock_app, ota_provider_app, ota_requestor_app, tv_app, bridge_app, chip_repl_yaml_tester, chip_tool_with_python, pics_file, keep_going, test_timeout_seconds, expected_failures): if expected_failures != 0 and not keep_going: logging.exception(f"'--expected-failures {expected_failures}' used without '--keep-going'") sys.exit(2) runner = chiptest.runner.Runner() paths_finder = PathsFinder() if all_clusters_app is None: all_clusters_app = paths_finder.get('chip-all-clusters-app') if lock_app is None: lock_app = paths_finder.get('chip-lock-app') if ota_provider_app is None: ota_provider_app = paths_finder.get('chip-ota-provider-app') if ota_requestor_app is None: ota_requestor_app = paths_finder.get('chip-ota-requestor-app') if tv_app is None: tv_app = paths_finder.get('chip-tv-app') if bridge_app is None: bridge_app = paths_finder.get('chip-bridge-app') if chip_repl_yaml_tester is None: chip_repl_yaml_tester = paths_finder.get('yamltest_with_chip_repl_tester.py') if chip_tool_with_python is None: if context.obj.chip_tool and os.path.basename(context.obj.chip_tool) == "darwin-framework-tool": chip_tool_with_python = paths_finder.get('darwinframeworktool.py') else: chip_tool_with_python = paths_finder.get('chiptool.py') # Command execution requires an array paths = chiptest.ApplicationPaths( chip_tool=[context.obj.chip_tool], all_clusters_app=[all_clusters_app], lock_app=[lock_app], ota_provider_app=[ota_provider_app], ota_requestor_app=[ota_requestor_app], tv_app=[tv_app], bridge_app=[bridge_app], chip_repl_yaml_tester_cmd=['python3'] + [chip_repl_yaml_tester], chip_tool_with_python_cmd=['python3'] + [chip_tool_with_python], ) if sys.platform == 'linux': chiptest.linux.PrepareNamespacesForTestExecution( context.obj.in_unshare) paths = chiptest.linux.PathsWithNetworkNamespaces(paths) logging.info("Each test will be executed %d times" % iterations) apps_register = AppsRegister() apps_register.init() def cleanup(): apps_register.uninit() if sys.platform == 'linux': chiptest.linux.ShutdownNamespaceForTestExecution() for i in range(iterations): logging.info("Starting iteration %d" % (i+1)) observed_failures = 0 for test in context.obj.tests: if context.obj.include_tags: if not (test.tags & context.obj.include_tags): logging.debug("Test %s not included" % test.name) continue if context.obj.exclude_tags: if test.tags & context.obj.exclude_tags: logging.debug("Test %s excluded" % test.name) continue test_start = time.monotonic() try: if context.obj.dry_run: logging.info("Would run test: %s" % test.name) continue logging.info('%-20s - Starting test' % (test.name)) test.Run( runner, apps_register, paths, pics_file, test_timeout_seconds, context.obj.dry_run, test_runtime=context.obj.runtime) test_end = time.monotonic() logging.info('%-30s - Completed in %0.2f seconds' % (test.name, (test_end - test_start))) except Exception: test_end = time.monotonic() logging.exception('%-30s - FAILED in %0.2f seconds' % (test.name, (test_end - test_start))) observed_failures += 1 if not keep_going: cleanup() sys.exit(2) if observed_failures != expected_failures: logging.exception(f'Iteration {i}: expected failure count {expected_failures}, but got {observed_failures}') cleanup() sys.exit(2) cleanup() # On linux, allow an execution shell to be prepared if sys.platform == 'linux': @main.command( 'shell', help=('Execute a bash shell in the environment (useful to test ' 'network namespaces)')) @click.pass_context def cmd_shell(context): chiptest.linux.PrepareNamespacesForTestExecution( context.obj.in_unshare) os.execvpe("bash", ["bash"], os.environ.copy()) if __name__ == '__main__': main(auto_envvar_prefix='CHIP')