python/generators/diff_tests/runner.py - third_party/perfetto - Git at Google


            
            #!/usr/bin/env python3

            
            # Copyright (C) 2023 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 concurrent.futures

            
            import datetime

            
            import os

            
            import subprocess

            
            import sys

            
            import tempfile

            
            from dataclasses import dataclass

            
            from typing import Dict, List, Set, Tuple


            from python.generators.diff_tests.models import (TestCase, TestResult, TestType,

            
                                                             PerfResult, Config)

            
            from python.generators.diff_tests.utils import (ColorFormatter, ProtoManager,

            
                                                            get_trace_descriptor_path,

            
                                                            write_diff)

            
            from python.generators.diff_tests.trace_generator import generate_trace_file

            
            from python.generators.diff_tests.test_executor import (

            
                QueryTestExecutor, MetricTestExecutor, MetricV2TestExecutor,

            
                StructuredQueryTestExecutor)

            
            from python.generators.diff_tests.test_loader import TestLoader


            @dataclass

            
            class TestResults:

            
              """Results of running the test suite.


              Mostly used for printing aggregated results.

            
              """

            
              test_failures: List[str]

            
              perf_data: List[PerfResult]

            
              test_time_ms: int

            
              skipped_tests_filter: List[Tuple[str, str]]

            
              total_tests_to_run: int

            
              tests_skipped_by_name: int


              def str(self, no_colors: bool) -> str:

            
                c = ColorFormatter(no_colors)

            
                total_tests = self.total_tests_to_run + self.tests_skipped_by_name + len(

            
                    self.skipped_tests_filter)

            
                tests_in_filter = self.total_tests_to_run + len(self.skipped_tests_filter)

            
                passed_tests = self.total_tests_to_run - len(self.test_failures)

            
                res = (

            
                    f"[==========] Name filter selected {tests_in_filter} tests out of {total_tests}.\n"

            
                    f"[==========] {self.total_tests_to_run} tests ran out of {tests_in_filter} total. ({self.test_time_ms} ms total)\n"

            
                    f"{c.green('[  PASSED  ]')} "

            
                    f"{passed_tests} tests.\n")

            
                if len(self.skipped_tests_filter) > 0:

            
                  res += (f"{c.yellow('[ SKIPPED  ]')} "

            
                          f"{len(self.skipped_tests_filter)} tests.\n")

            
                  for name, reason in self.skipped_tests_filter:

            
                    res += f"{c.yellow('[ SKIPPED  ]')} {name} (reason: {reason})\n"

            
                if len(self.test_failures) > 0:

            
                  res += (f"{c.red('[  FAILED  ]')} "

            
                          f"{len(self.test_failures)} tests.\n")

            
                  for failure in self.test_failures:

            
                    res += f"{c.red('[  FAILED  ]')} {failure}\n"

            
                return res


            @dataclass

            
            class DiffTestsRunner:

            
              """Fetches and executes all diff tests."""


              def __init__(self, config: Config):

            
                self.config = config

            
                self.test_loader = TestLoader(os.path.abspath(self.config.test_dir))

            
                self.enabled_modules = self._get_build_config()


              def run(self) -> TestResults:

            
                # Discover and filter tests.

            
                db = self.test_loader.discover_and_load_tests(self.config.name_filter,

            
                                                              self.enabled_modules)

            
                tests = db.runnable

            
                total_tests_to_run = len(tests)

            
                tests_skipped_by_name = len(db.skipped_name_filter)


                sys.stderr.write(f'[==========] Running {total_tests_to_run} tests.\n')


                trace_descriptor_path = get_trace_descriptor_path(

            
                    os.path.dirname(self.config.trace_processor_path),

            
                    self.config.trace_descriptor)


                if not self.config.metrics_descriptor_paths:

            
                  out_path = os.path.dirname(self.config.trace_processor_path)

            
                  metrics_protos_path = os.path.join(out_path, 'gen', 'protos', 'perfetto',

            
                                                     'metrics')

            
                  self.config.metrics_descriptor_paths = [

            
                      os.path.join(metrics_protos_path, 'metrics.descriptor'),

            
                      os.path.join(metrics_protos_path, 'chrome',

            
                                   'all_chrome_metrics.descriptor'),

            
                      os.path.join(metrics_protos_path, 'webview',

            
                                   'all_webview_metrics.descriptor')

            
                  ]


                perf_results = []

            
                failures = []

            
                test_run_start = datetime.datetime.now()

            
                completed_tests = 0


                max_workers = self.config.jobs if self.config.jobs > 0 else None

            
                with concurrent.futures.ProcessPoolExecutor(max_workers=max_workers) as e:

            
                  fut = [

            
                      e.submit(self._run_test, test, trace_descriptor_path)

            
                      for test in tests

            
                  ]

            
                  for res in concurrent.futures.as_completed(fut):

            
                    test_name, res_str, result = res.result()


                    if self.config.quiet:

            
                      completed_tests += 1

            
                      sys.stderr.write(f"\rRan {completed_tests} tests")

            
                      if not result.passed:

            
                        sys.stderr.write(f"\r")

            
                        sys.stderr.write(res_str)

            
                    else:

            
                      sys.stderr.write(res_str)


                    if not result or not result.passed:

            
                      failures.append(test_name)

            
                    elif result.perf_result:

            
                      # Passed expected-error tests have no perf result: trace_processor

            
                      # exits before the query runs.

            
                      perf_results.append(result.perf_result)

            
                test_time_ms = int(

            
                    (datetime.datetime.now() - test_run_start).total_seconds() * 1000)

            
                if self.config.quiet:

            
                  sys.stderr.write(f"\r")

            
                return TestResults(failures, perf_results, test_time_ms,

            
                                   db.skipped_module_missing, total_tests_to_run,

            
                                   tests_skipped_by_name)


              def _run_test(self, test: TestCase,

            
                            trace_descriptor_path: str) -> Tuple[str, str, TestResult]:

            
                # Simpleperf is not an extension of Perfetto's trace proto, but a separate

            
                # format. Only pass it for simpleperf-specific tests.

            
                extension_descriptor_paths = [

            
                    self.config.chrome_extensions, self.config.test_extensions,

            
                    self.config.winscope_extensions, self.config.gpu_extensions,

            
                    self.config.gpu_interned_data_extensions

            
                ]

            
                gen_trace_file = generate_trace_file(test, trace_descriptor_path,

            
                                                     extension_descriptor_paths,

            
                                                     self.config.simpleperf_descriptor)


                if gen_trace_file:

            
                  trace_path = os.path.realpath(gen_trace_file.name)

            
                else:

            
                  trace_path = test.trace_path

            
                assert trace_path


                if test.type == TestType.QUERY:

            
                  executor = QueryTestExecutor(self.config.trace_processor_path,

            
                                               self.config.override_sql_package_paths)

            
                  result = executor.run(test, trace_path)

            
                elif test.type == TestType.METRIC:

            
                  executor = MetricTestExecutor(

            
                      self.config.trace_processor_path,

            
                      self.config.override_sql_package_paths,

            
                      ProtoManager(self.config.metrics_descriptor_paths).create_message)

            
                  result = executor.run(test, trace_path)

            
                elif test.type == TestType.METRIC_V2:

            
                  executor = MetricV2TestExecutor(

            
                      self.config.trace_processor_path,

            
                      self.config.override_sql_package_paths, self.config.keep_input,

            
                      ProtoManager([self.config.summary_descriptor]).create_message,

            
                      ProtoManager([self.config.summary_descriptor]).create_message)

            
                  result = executor.run(test, trace_path)

            
                elif test.type == TestType.STRUCTURED_QUERY:

            
                  executor = StructuredQueryTestExecutor(

            
                      self.config.trace_processor_path,

            
                      self.config.override_sql_package_paths, self.config.keep_input)

            
                  result = executor.run(test, trace_path)

            
                else:

            
                  assert False


                if gen_trace_file:

            
                  if not self.config.keep_input:

            
                    gen_trace_file.close()

            
                    os.remove(trace_path)


                print_trace_path = gen_trace_file and self.config.keep_input

            
                run_str = self._process_test_result(result, trace_path,

            
                                                    extension_descriptor_paths,

            
                                                    trace_descriptor_path, print_trace_path)

            
                return test.name, run_str, result


              def _process_test_result(

            
                  self,

            
                  result: TestResult,

            
                  trace_path: str,

            
                  extension_descriptor_paths: List[str],

            
                  trace_descriptor_path: str,

            
                  print_trace_path: bool,

            
              ) -> str:

            
                colors = ColorFormatter(self.config.no_colors)


                def write_cmdlines() -> str:

            
                  res = ""

            
                  if result.test.trace_path and (

            
                      result.test.trace_path.endswith('.textproto') or

            
                      result.test.trace_path.endswith('.py')):

            
                    res += 'Command to generate trace:\n'

            
                    res += 'tools/serialize_test_trace.py '

            
                    assert result.test.trace_path

            
                    res += '--descriptor {} {} {} > {}\n'.format(

            
                        os.path.relpath(trace_descriptor_path,

            
                                        self.config.test_dir), " ".join([

            
                                            "--extension-descriptor {}".format(

            
                                                os.path.relpath(p, self.config.test_dir))

            
                                            for p in extension_descriptor_paths

            
                                        ]),

            
                        os.path.relpath(result.test.trace_path, self.config.test_dir),

            
                        os.path.relpath(trace_path, self.config.test_dir),

            
                        extension_descriptor_paths)

            
                  res += f"Command line:\n{' '.join(result.cmd)}\n"

            
                  return res


                run_str = f"{colors.yellow('[ RUN      ]')} {result.test.name}\n"

            
                run_diagnostics = [f"trace_path: {trace_path}"] if print_trace_path else []

            
                if result.test.blueprint.is_out_expected_error():

            
                  # Expected-error tests pass when trace_processor fails: a non-zero exit

            
                  # code is part of the expectation, not a test harness failure.

            
                  if result.passed:

            
                    run_str += f"{colors.green('[       OK ]')} {result.test.name}"

            
                  else:

            
                    run_str += result.stderr

            
                    if result.exit_code == 0:

            
                      run_str += (f"Expected trace_processor to fail with an error "

            
                                  f"containing '{result.expected}' but it exited "

            
                                  f"successfully.\n")

            
                    else:

            
                      run_str += (f"trace_processor failed as expected but stderr did not "

            
                                  f"contain '{result.expected}'.\n")

            
                    run_str += write_cmdlines()

            
                    run_str += (f"{colors.red('[  FAILED  ]')} {result.test.name}")

            
                elif result.exit_code != 0 or not result.passed:

            
                  result.passed = False

            
                  run_str += result.stderr


                  if result.exit_code == 0:

            
                    run_str += f"Expected did not match actual for test {result.test.name}.\n"

            
                    run_str += write_cmdlines()

            
                    run_str += write_diff(result.expected, result.actual)

            
                  else:

            
                    run_str += write_cmdlines()


                  run_str += (f"{colors.red('[  FAILED  ]')} {result.test.name}")

            
                else:

            
                  assert result.perf_result

            
                  run_str += f"{colors.green('[       OK ]')} {result.test.name}"

            
                  run_diagnostics.append(

            
                      f"ingest: {result.perf_result.ingest_time_ns / 1000000:.2f} ms")

            
                  run_diagnostics.append(

            
                      f"query: {result.perf_result.real_time_ns / 1000000:.2f} ms")

            
                run_str += f" ({' '.join(run_diagnostics)})\n" if run_diagnostics else "\n"

            
                return run_str


              def _get_build_config(self) -> Set[str]:

            
                """Returns the modules from trace processor."""

            
                with tempfile.NamedTemporaryFile(

            
                    mode='w', suffix='.textproto') as empty_trace:

            
                  empty_trace.write('')

            
                  empty_trace.flush()

            
                  args = [

            
                      self.config.trace_processor_path,

            
                      empty_trace.name,

            
                      '--query-string',

            
                      'select name from __intrinsic_modules',

            
                  ]

            
                  modules_str = subprocess.check_output(args, stderr=subprocess.PIPE)

            
                  modules = set(

            
                      line.strip('"')

            
                      for line in modules_str.decode('utf-8').splitlines()

            
                      if line and not line.startswith('name'))

            
                  return modules
	#!/usr/bin/env python3
	# Copyright (C) 2023 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 concurrent.futures
	import datetime
	import os
	import subprocess
	import sys
	import tempfile
	from dataclasses import dataclass
	from typing import Dict, List, Set, Tuple

	from python.generators.diff_tests.models import (TestCase, TestResult, TestType,
	PerfResult, Config)
	from python.generators.diff_tests.utils import (ColorFormatter, ProtoManager,
	get_trace_descriptor_path,
	write_diff)
	from python.generators.diff_tests.trace_generator import generate_trace_file
	from python.generators.diff_tests.test_executor import (
	QueryTestExecutor, MetricTestExecutor, MetricV2TestExecutor,
	StructuredQueryTestExecutor)
	from python.generators.diff_tests.test_loader import TestLoader


	@dataclass
	class TestResults:
	"""Results of running the test suite.

	Mostly used for printing aggregated results.
	"""
	test_failures: List[str]
	perf_data: List[PerfResult]
	test_time_ms: int
	skipped_tests_filter: List[Tuple[str, str]]
	total_tests_to_run: int
	tests_skipped_by_name: int

	def str(self, no_colors: bool) -> str:
	c = ColorFormatter(no_colors)
	total_tests = self.total_tests_to_run + self.tests_skipped_by_name + len(
	self.skipped_tests_filter)
	tests_in_filter = self.total_tests_to_run + len(self.skipped_tests_filter)
	passed_tests = self.total_tests_to_run - len(self.test_failures)
	res = (
	f"[==========] Name filter selected {tests_in_filter} tests out of {total_tests}.\n"
	f"[==========] {self.total_tests_to_run} tests ran out of {tests_in_filter} total. ({self.test_time_ms} ms total)\n"
	f"{c.green('[ PASSED ]')} "
	f"{passed_tests} tests.\n")
	if len(self.skipped_tests_filter) > 0:
	res += (f"{c.yellow('[ SKIPPED ]')} "
	f"{len(self.skipped_tests_filter)} tests.\n")
	for name, reason in self.skipped_tests_filter:
	res += f"{c.yellow('[ SKIPPED ]')} {name} (reason: {reason})\n"
	if len(self.test_failures) > 0:
	res += (f"{c.red('[ FAILED ]')} "
	f"{len(self.test_failures)} tests.\n")
	for failure in self.test_failures:
	res += f"{c.red('[ FAILED ]')} {failure}\n"
	return res


	@dataclass
	class DiffTestsRunner:
	"""Fetches and executes all diff tests."""

	def __init__(self, config: Config):
	self.config = config
	self.test_loader = TestLoader(os.path.abspath(self.config.test_dir))
	self.enabled_modules = self._get_build_config()

	def run(self) -> TestResults:
	# Discover and filter tests.
	db = self.test_loader.discover_and_load_tests(self.config.name_filter,
	self.enabled_modules)
	tests = db.runnable
	total_tests_to_run = len(tests)
	tests_skipped_by_name = len(db.skipped_name_filter)

	sys.stderr.write(f'[==========] Running {total_tests_to_run} tests.\n')

	trace_descriptor_path = get_trace_descriptor_path(
	os.path.dirname(self.config.trace_processor_path),
	self.config.trace_descriptor)

	if not self.config.metrics_descriptor_paths:
	out_path = os.path.dirname(self.config.trace_processor_path)
	metrics_protos_path = os.path.join(out_path, 'gen', 'protos', 'perfetto',
	'metrics')
	self.config.metrics_descriptor_paths = [
	os.path.join(metrics_protos_path, 'metrics.descriptor'),
	os.path.join(metrics_protos_path, 'chrome',
	'all_chrome_metrics.descriptor'),
	os.path.join(metrics_protos_path, 'webview',
	'all_webview_metrics.descriptor')
	]

	perf_results = []
	failures = []
	test_run_start = datetime.datetime.now()
	completed_tests = 0

	max_workers = self.config.jobs if self.config.jobs > 0 else None
	with concurrent.futures.ProcessPoolExecutor(max_workers=max_workers) as e:
	fut = [
	e.submit(self._run_test, test, trace_descriptor_path)
	for test in tests
	]
	for res in concurrent.futures.as_completed(fut):
	test_name, res_str, result = res.result()

	if self.config.quiet:
	completed_tests += 1
	sys.stderr.write(f"\rRan {completed_tests} tests")
	if not result.passed:
	sys.stderr.write(f"\r")
	sys.stderr.write(res_str)
	else:
	sys.stderr.write(res_str)

	if not result or not result.passed:
	failures.append(test_name)
	elif result.perf_result:
	# Passed expected-error tests have no perf result: trace_processor
	# exits before the query runs.
	perf_results.append(result.perf_result)
	test_time_ms = int(
	(datetime.datetime.now() - test_run_start).total_seconds() * 1000)
	if self.config.quiet:
	sys.stderr.write(f"\r")
	return TestResults(failures, perf_results, test_time_ms,
	db.skipped_module_missing, total_tests_to_run,
	tests_skipped_by_name)

	def _run_test(self, test: TestCase,
	trace_descriptor_path: str) -> Tuple[str, str, TestResult]:
	# Simpleperf is not an extension of Perfetto's trace proto, but a separate
	# format. Only pass it for simpleperf-specific tests.
	extension_descriptor_paths = [
	self.config.chrome_extensions, self.config.test_extensions,
	self.config.winscope_extensions, self.config.gpu_extensions,
	self.config.gpu_interned_data_extensions
	]
	gen_trace_file = generate_trace_file(test, trace_descriptor_path,
	extension_descriptor_paths,
	self.config.simpleperf_descriptor)

	if gen_trace_file:
	trace_path = os.path.realpath(gen_trace_file.name)
	else:
	trace_path = test.trace_path
	assert trace_path

	if test.type == TestType.QUERY:
	executor = QueryTestExecutor(self.config.trace_processor_path,
	self.config.override_sql_package_paths)
	result = executor.run(test, trace_path)
	elif test.type == TestType.METRIC:
	executor = MetricTestExecutor(
	self.config.trace_processor_path,
	self.config.override_sql_package_paths,
	ProtoManager(self.config.metrics_descriptor_paths).create_message)
	result = executor.run(test, trace_path)
	elif test.type == TestType.METRIC_V2:
	executor = MetricV2TestExecutor(
	self.config.trace_processor_path,
	self.config.override_sql_package_paths, self.config.keep_input,
	ProtoManager([self.config.summary_descriptor]).create_message,
	ProtoManager([self.config.summary_descriptor]).create_message)
	result = executor.run(test, trace_path)
	elif test.type == TestType.STRUCTURED_QUERY:
	executor = StructuredQueryTestExecutor(
	self.config.trace_processor_path,
	self.config.override_sql_package_paths, self.config.keep_input)
	result = executor.run(test, trace_path)
	else:
	assert False

	if gen_trace_file:
	if not self.config.keep_input:
	gen_trace_file.close()
	os.remove(trace_path)

	print_trace_path = gen_trace_file and self.config.keep_input
	run_str = self._process_test_result(result, trace_path,
	extension_descriptor_paths,
	trace_descriptor_path, print_trace_path)
	return test.name, run_str, result

	def _process_test_result(
	self,
	result: TestResult,
	trace_path: str,
	extension_descriptor_paths: List[str],
	trace_descriptor_path: str,
	print_trace_path: bool,
	) -> str:
	colors = ColorFormatter(self.config.no_colors)

	def write_cmdlines() -> str:
	res = ""
	if result.test.trace_path and (
	result.test.trace_path.endswith('.textproto') or
	result.test.trace_path.endswith('.py')):
	res += 'Command to generate trace:\n'
	res += 'tools/serialize_test_trace.py '
	assert result.test.trace_path
	res += '--descriptor {} {} {} > {}\n'.format(
	os.path.relpath(trace_descriptor_path,
	self.config.test_dir), " ".join([
	"--extension-descriptor {}".format(
	os.path.relpath(p, self.config.test_dir))
	for p in extension_descriptor_paths
	]),
	os.path.relpath(result.test.trace_path, self.config.test_dir),
	os.path.relpath(trace_path, self.config.test_dir),
	extension_descriptor_paths)
	res += f"Command line:\n{' '.join(result.cmd)}\n"
	return res

	run_str = f"{colors.yellow('[ RUN ]')} {result.test.name}\n"
	run_diagnostics = [f"trace_path: {trace_path}"] if print_trace_path else []
	if result.test.blueprint.is_out_expected_error():
	# Expected-error tests pass when trace_processor fails: a non-zero exit
	# code is part of the expectation, not a test harness failure.
	if result.passed:
	run_str += f"{colors.green('[ OK ]')} {result.test.name}"
	else:
	run_str += result.stderr
	if result.exit_code == 0:
	run_str += (f"Expected trace_processor to fail with an error "
	f"containing '{result.expected}' but it exited "
	f"successfully.\n")
	else:
	run_str += (f"trace_processor failed as expected but stderr did not "
	f"contain '{result.expected}'.\n")
	run_str += write_cmdlines()
	run_str += (f"{colors.red('[ FAILED ]')} {result.test.name}")
	elif result.exit_code != 0 or not result.passed:
	result.passed = False
	run_str += result.stderr

	if result.exit_code == 0:
	run_str += f"Expected did not match actual for test {result.test.name}.\n"
	run_str += write_cmdlines()
	run_str += write_diff(result.expected, result.actual)
	else:
	run_str += write_cmdlines()

	run_str += (f"{colors.red('[ FAILED ]')} {result.test.name}")
	else:
	assert result.perf_result
	run_str += f"{colors.green('[ OK ]')} {result.test.name}"
	run_diagnostics.append(
	f"ingest: {result.perf_result.ingest_time_ns / 1000000:.2f} ms")
	run_diagnostics.append(
	f"query: {result.perf_result.real_time_ns / 1000000:.2f} ms")
	run_str += f" ({' '.join(run_diagnostics)})\n" if run_diagnostics else "\n"
	return run_str

	def _get_build_config(self) -> Set[str]:
	"""Returns the modules from trace processor."""
	with tempfile.NamedTemporaryFile(
	mode='w', suffix='.textproto') as empty_trace:
	empty_trace.write('')
	empty_trace.flush()
	args = [
	self.config.trace_processor_path,
	empty_trace.name,
	'--query-string',
	'select name from __intrinsic_modules',
	]
	modules_str = subprocess.check_output(args, stderr=subprocess.PIPE)
	modules = set(
	line.strip('"')
	for line in modules_str.decode('utf-8').splitlines()
	if line and not line.startswith('name'))
	return modules