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 difflib
 import os
 import subprocess
 import sys
 import tempfile
 from binascii import unhexlify
 from dataclasses import dataclass
 from typing import List, Tuple, Optional

 from google.protobuf import text_format, message_factory, descriptor_pool
 from python.generators.diff_tests.testing import TestCase, TestType, BinaryProto
 from python.generators.diff_tests.utils import (
     ColorFormatter, create_message_factory, get_env, get_trace_descriptor_path,
     read_all_tests, serialize_python_trace, serialize_textproto_trace)

 ROOT_DIR = os.path.dirname(
     os.path.dirname(
         os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))


 # Performance result of running the test.
 @dataclass
 class PerfResult:
   test: TestCase
   ingest_time_ns: int
   real_time_ns: int

   def __init__(self, test: TestCase, perf_lines: List[str]):
     self.test = test

     assert len(perf_lines) == 1
     perf_numbers = perf_lines[0].split(',')

     assert len(perf_numbers) == 2
     self.ingest_time_ns = int(perf_numbers[0])
     self.real_time_ns = int(perf_numbers[1])


 # Data gathered from running the test.
 @dataclass
 class TestResult:
   test: TestCase
   trace: str
   cmd: List[str]
   expected: str
   actual: str
   passed: bool
   stderr: str
   exit_code: int
   perf_result: Optional[PerfResult]

   def __init__(self, test: TestCase, gen_trace_path: str, cmd: List[str],
                expected_text: str, actual_text: str, stderr: str,
                exit_code: int, perf_lines: List[str]) -> None:
     self.test = test
     self.trace = gen_trace_path
     self.cmd = cmd
     self.stderr = stderr
     self.exit_code = exit_code

     # For better string formatting we often add whitespaces, which has to now
     # be removed.
     def strip_whitespaces(text: str):
       no_front_new_line_text = text.lstrip('\n')
       return '\n'.join(s.strip() for s in no_front_new_line_text.split('\n'))

     self.expected = strip_whitespaces(expected_text)
     self.actual = strip_whitespaces(actual_text)

     expected_content = self.expected.replace('\r\n', '\n')

     actual_content = self.actual.replace('\r\n', '\n')
     self.passed = (expected_content == actual_content)

     if self.exit_code == 0:
       self.perf_result = PerfResult(self.test, perf_lines)
     else:
       self.perf_result = None

   def write_diff(self):
     expected_lines = self.expected.splitlines(True)
     actual_lines = self.actual.splitlines(True)
     diff = difflib.unified_diff(
         expected_lines, actual_lines, fromfile='expected', tofile='actual')
     return "".join(list(diff))

   def rebase(self, rebase) -> str:
     if not rebase or self.passed:
       return ""
     if not self.test.blueprint.is_out_file():
       return f"Can't rebase expected results passed as strings.\n"
     if self.exit_code != 0:
       return f"Rebase failed for {self.test.name} as query failed\n"

     with open(self.test.expected_path, 'w') as f:
       f.write(self.actual)
     return f"Rebasing {self.test.name}\n"


 # Results of running the test suite. Mostly used for printing aggregated
 # results.
 @dataclass
 class TestResults:
   test_failures: List[str]
   perf_data: List[PerfResult]
   rebased: List[str]
   test_time_ms: int

   def str(self, no_colors: bool, tests_no: int):
     c = ColorFormatter(no_colors)
     res = (
         f"[==========] {tests_no} tests ran. ({self.test_time_ms} ms total)\n"
         f"{c.green('[  PASSED  ]')} "
         f"{tests_no - len(self.test_failures)} tests.\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

   def rebase_str(self):
     res = f"\n[  REBASED  ] {len(self.rebased)} tests.\n"
     for name in self.rebased:
       res += f"[  REBASED  ] {name}\n"
     return res


 # Responsible for executing singular diff test.
 @dataclass
 class TestCaseRunner:
   test: TestCase
   trace_processor_path: str
   trace_descriptor_path: str
   colors: ColorFormatter

   def __output_to_text_proto(self, actual: str, out: BinaryProto) -> str:
     """Deserializes a binary proto and returns its text representation.

   Args:
     actual: (string) HEX encoded serialized proto message
     message_type: (string) Message type

   Returns:
     Text proto
   """
     try:
       raw_data = unhexlify(actual.splitlines()[-1][1:-1])
       out_path = os.path.dirname(self.trace_processor_path)
       descriptor_paths = [
           f.path
           for f in os.scandir(
               os.path.join(ROOT_DIR, out_path, 'gen', 'protos', 'perfetto',
                            'trace_processor'))
           if f.is_file() and os.path.splitext(f.name)[1] == '.descriptor'
       ]
       descriptor_paths.append(
           os.path.join(ROOT_DIR, out_path, 'gen', 'protos', 'third_party',
                        'pprof', 'profile.descriptor'))
       proto = create_message_factory(descriptor_paths, out.message_type)()
       proto.ParseFromString(raw_data)
       try:
         return out.post_processing(proto)
       except:
         return '<Proto post processing failed>'
       return text_format.MessageToString(proto)
     except:
       return '<Invalid input for proto deserializaiton>'

   def __run_metrics_test(self, trace_path: str,
                          metrics_message_factory) -> TestResult:

     if self.test.blueprint.is_out_file():
       with open(self.test.expected_path, 'r') as expected_file:
         expected = expected_file.read()
     else:
       expected = self.test.blueprint.out.contents

     tmp_perf_file = tempfile.NamedTemporaryFile(delete=False)
     is_json_output_file = self.test.blueprint.is_out_file(
     ) and os.path.basename(self.test.expected_path).endswith('.json.out')
     is_json_output = is_json_output_file or self.test.blueprint.is_out_json()
     cmd = [
         self.trace_processor_path,
         '--analyze-trace-proto-content',
         '--crop-track-events',
         '--run-metrics',
         self.test.blueprint.query.name,
         '--metrics-output=%s' % ('json' if is_json_output else 'binary'),
         '--perf-file',
         tmp_perf_file.name,
         trace_path,
     ]
     tp = subprocess.Popen(
         cmd,
         stdout=subprocess.PIPE,
         stderr=subprocess.PIPE,
         env=get_env(ROOT_DIR))
     (stdout, stderr) = tp.communicate()

     if is_json_output:
       expected_text = expected
       actual_text = stdout.decode('utf8')
     else:
       # Expected will be in text proto format and we'll need to parse it to
       # a real proto.
       expected_message = metrics_message_factory()
       text_format.Merge(expected, expected_message)

       # Actual will be the raw bytes of the proto and we'll need to parse it
       # into a message.
       actual_message = metrics_message_factory()
       actual_message.ParseFromString(stdout)

       # Convert both back to text format.
       expected_text = text_format.MessageToString(expected_message)
       actual_text = text_format.MessageToString(actual_message)

     perf_lines = [line.decode('utf8') for line in tmp_perf_file.readlines()]
     tmp_perf_file.close()
     os.remove(tmp_perf_file.name)
     return TestResult(self.test, trace_path, cmd, expected_text, actual_text,
                       stderr.decode('utf8'), tp.returncode, perf_lines)

   # Run a query based Diff Test.
   def __run_query_test(self, trace_path: str) -> TestResult:
     # Fetch expected text.
     if self.test.expected_path:
       with open(self.test.expected_path, 'r') as expected_file:
         expected = expected_file.read()
     else:
       expected = self.test.blueprint.out.contents

     # Fetch query.
     if self.test.blueprint.is_query_file():
       query = self.test.query_path
     else:
       tmp_query_file = tempfile.NamedTemporaryFile(delete=False)
       with open(tmp_query_file.name, 'w') as query_file:
         query_file.write(self.test.blueprint.query)
       query = tmp_query_file.name

     tmp_perf_file = tempfile.NamedTemporaryFile(delete=False)
     cmd = [
         self.trace_processor_path,
         '--analyze-trace-proto-content',
         '--crop-track-events',
         '-q',
         query,
         '--perf-file',
         tmp_perf_file.name,
         trace_path,
     ]
     tp = subprocess.Popen(
         cmd,
         stdout=subprocess.PIPE,
         stderr=subprocess.PIPE,
         env=get_env(ROOT_DIR))
     (stdout, stderr) = tp.communicate()

     if not self.test.blueprint.is_query_file():
       tmp_query_file.close()
       os.remove(tmp_query_file.name)
     perf_lines = [line.decode('utf8') for line in tmp_perf_file.readlines()]
     tmp_perf_file.close()
     os.remove(tmp_perf_file.name)

     actual = stdout.decode('utf8')
     if self.test.blueprint.is_out_binaryproto():
       actual = self.__output_to_text_proto(actual, self.test.blueprint.out)

     return TestResult(self.test, trace_path, cmd, expected, actual,
                       stderr.decode('utf8'), tp.returncode, perf_lines)

   def __run(self, metrics_descriptor_paths: List[str],
             extension_descriptor_paths: List[str], keep_input,
             rebase) -> Tuple[TestResult, str]:
     # We can't use delete=True here. When using that on Windows, the
     # resulting file is opened in exclusive mode (in turn that's a subtle
     # side-effect of the underlying CreateFile(FILE_ATTRIBUTE_TEMPORARY))
     # and TP fails to open the passed path.
     gen_trace_file = None
     if self.test.blueprint.is_trace_file():
       if self.test.trace_path.endswith('.py'):
         gen_trace_file = tempfile.NamedTemporaryFile(delete=False)
         serialize_python_trace(ROOT_DIR, self.trace_descriptor_path,
                                self.test.trace_path, gen_trace_file)

       elif self.test.trace_path.endswith('.textproto'):
         gen_trace_file = tempfile.NamedTemporaryFile(delete=False)
         serialize_textproto_trace(self.trace_descriptor_path,
                                   extension_descriptor_paths,
                                   self.test.trace_path, gen_trace_file)

     elif self.test.blueprint.is_trace_textproto():
       gen_trace_file = tempfile.NamedTemporaryFile(delete=False)
       proto = create_message_factory([self.trace_descriptor_path] +
                                      extension_descriptor_paths,
                                      'perfetto.protos.Trace')()
       text_format.Merge(self.test.blueprint.trace.contents, proto)
       gen_trace_file.write(proto.SerializeToString())
       gen_trace_file.flush()

     else:
       gen_trace_file = tempfile.NamedTemporaryFile(delete=False)
       with open(gen_trace_file.name, 'w') as trace_file:
         trace_file.write(self.test.blueprint.trace.contents)

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

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

     if self.test.type == TestType.QUERY:
       result = self.__run_query_test(trace_path)
     elif self.test.type == TestType.METRIC:
       result = self.__run_metrics_test(
           trace_path,
           create_message_factory(metrics_descriptor_paths,
                                  'perfetto.protos.TraceMetrics'))
     else:
       assert False

     if gen_trace_file:
       if keep_input:
         str += f"Saving generated input trace: {trace_path}\n"
       else:
         gen_trace_file.close()
         os.remove(trace_path)

     def write_cmdlines():
       res = ""
       if not gen_trace_file:
         res += 'Command to generate trace:\n'
         res += 'tools/serialize_test_trace.py '
         res += '--descriptor {} {} > {}\n'.format(
             os.path.relpath(self.trace_descriptor_path, ROOT_DIR),
             os.path.relpath(self.test.trace_path, ROOT_DIR),
             os.path.relpath(trace_path, ROOT_DIR))
       res += f"Command line:\n{' '.join(result.cmd)}\n"
       return res

     if result.exit_code != 0 or not result.passed:
       str += result.stderr

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

       str += (f"{self.colors.red('[  FAILED  ]')} {self.test.name}\n")
       str += result.rebase(rebase)

       return result, str
     else:
       str += (f"{self.colors.green('[       OK ]')} {self.test.name} "
               f"(ingest: {result.perf_result.ingest_time_ns / 1000000:.2f} ms "
               f"query: {result.perf_result.real_time_ns / 1000000:.2f} ms)\n")
     return result, str

   # Run a TestCase.
   def execute(self, extension_descriptor_paths: List[str],
               metrics_descriptor: str, keep_input: bool,
               rebase: bool) -> Tuple[str, str, TestResult]:
     if metrics_descriptor:
       metrics_descriptor_paths = [metrics_descriptor]
     else:
       out_path = os.path.dirname(self.trace_processor_path)
       metrics_protos_path = os.path.join(out_path, 'gen', 'protos', 'perfetto',
                                          'metrics')
       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')
       ]
     result_str = ""

     result, run_str = self.__run(metrics_descriptor_paths,
                                  extension_descriptor_paths, keep_input, rebase)
     result_str += run_str
     if not result:
       return self.test.name, result_str, None

     return self.test.name, result_str, result


 # Fetches and executes all diff viable tests.
 @dataclass
 class DiffTestsRunner:
   tests: List[TestCase]
   trace_processor_path: str
   trace_descriptor_path: str
   test_runners: List[TestCaseRunner]

   def __init__(self, name_filter: str, trace_processor_path: str,
                trace_descriptor: str, no_colors: bool):
     self.tests = read_all_tests(name_filter, ROOT_DIR)
     self.trace_processor_path = trace_processor_path

     out_path = os.path.dirname(self.trace_processor_path)
     self.trace_descriptor_path = get_trace_descriptor_path(
         out_path, trace_descriptor)
     self.test_runners = []
     color_formatter = ColorFormatter(no_colors)
     for test in self.tests:
       self.test_runners.append(
           TestCaseRunner(test, self.trace_processor_path,
                          self.trace_descriptor_path, color_formatter))

   def run_all_tests(self, metrics_descriptor: str, keep_input: bool,
                     rebase: bool) -> TestResults:
     perf_results = []
     failures = []
     rebased = []
     test_run_start = datetime.datetime.now()

     out_path = os.path.dirname(self.trace_processor_path)
     chrome_extensions = os.path.join(out_path, 'gen', 'protos', 'third_party',
                                      'chromium',
                                      'chrome_track_event.descriptor')
     test_extensions = os.path.join(out_path, 'gen', 'protos', 'perfetto',
                                    'trace', 'test_extensions.descriptor')

     with concurrent.futures.ProcessPoolExecutor() as e:
       fut = [
           e.submit(test.execute, [chrome_extensions, test_extensions],
                    metrics_descriptor, keep_input, rebase)
           for test in self.test_runners
       ]
       for res in concurrent.futures.as_completed(fut):
         test_name, res_str, result = res.result()
         sys.stderr.write(res_str)
         if not result or not result.passed:
           if rebase:
             rebased.append(test_name)
           failures.append(test_name)
         else:
           perf_results.append(result.perf_result)
     test_time_ms = int(
         (datetime.datetime.now() - test_run_start).total_seconds() * 1000)
     return TestResults(failures, perf_results, rebased, test_time_ms)
	#!/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 difflib
	import os
	import subprocess
	import sys
	import tempfile
	from binascii import unhexlify
	from dataclasses import dataclass
	from typing import List, Tuple, Optional

	from google.protobuf import text_format, message_factory, descriptor_pool
	from python.generators.diff_tests.testing import TestCase, TestType, BinaryProto
	from python.generators.diff_tests.utils import (
	ColorFormatter, create_message_factory, get_env, get_trace_descriptor_path,
	read_all_tests, serialize_python_trace, serialize_textproto_trace)

	ROOT_DIR = os.path.dirname(
	os.path.dirname(
	os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))


	# Performance result of running the test.
	@dataclass
	class PerfResult:
	test: TestCase
	ingest_time_ns: int
	real_time_ns: int

	def __init__(self, test: TestCase, perf_lines: List[str]):
	self.test = test

	assert len(perf_lines) == 1
	perf_numbers = perf_lines[0].split(',')

	assert len(perf_numbers) == 2
	self.ingest_time_ns = int(perf_numbers[0])
	self.real_time_ns = int(perf_numbers[1])


	# Data gathered from running the test.
	@dataclass
	class TestResult:
	test: TestCase
	trace: str
	cmd: List[str]
	expected: str
	actual: str
	passed: bool
	stderr: str
	exit_code: int
	perf_result: Optional[PerfResult]

	def __init__(self, test: TestCase, gen_trace_path: str, cmd: List[str],
	expected_text: str, actual_text: str, stderr: str,
	exit_code: int, perf_lines: List[str]) -> None:
	self.test = test
	self.trace = gen_trace_path
	self.cmd = cmd
	self.stderr = stderr
	self.exit_code = exit_code

	# For better string formatting we often add whitespaces, which has to now
	# be removed.
	def strip_whitespaces(text: str):
	no_front_new_line_text = text.lstrip('\n')
	return '\n'.join(s.strip() for s in no_front_new_line_text.split('\n'))

	self.expected = strip_whitespaces(expected_text)
	self.actual = strip_whitespaces(actual_text)

	expected_content = self.expected.replace('\r\n', '\n')

	actual_content = self.actual.replace('\r\n', '\n')
	self.passed = (expected_content == actual_content)

	if self.exit_code == 0:
	self.perf_result = PerfResult(self.test, perf_lines)
	else:
	self.perf_result = None

	def write_diff(self):
	expected_lines = self.expected.splitlines(True)
	actual_lines = self.actual.splitlines(True)
	diff = difflib.unified_diff(
	expected_lines, actual_lines, fromfile='expected', tofile='actual')
	return "".join(list(diff))

	def rebase(self, rebase) -> str:
	if not rebase or self.passed:
	return ""
	if not self.test.blueprint.is_out_file():
	return f"Can't rebase expected results passed as strings.\n"
	if self.exit_code != 0:
	return f"Rebase failed for {self.test.name} as query failed\n"

	with open(self.test.expected_path, 'w') as f:
	f.write(self.actual)
	return f"Rebasing {self.test.name}\n"


	# Results of running the test suite. Mostly used for printing aggregated
	# results.
	@dataclass
	class TestResults:
	test_failures: List[str]
	perf_data: List[PerfResult]
	rebased: List[str]
	test_time_ms: int

	def str(self, no_colors: bool, tests_no: int):
	c = ColorFormatter(no_colors)
	res = (
	f"[==========] {tests_no} tests ran. ({self.test_time_ms} ms total)\n"
	f"{c.green('[ PASSED ]')} "
	f"{tests_no - len(self.test_failures)} tests.\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

	def rebase_str(self):
	res = f"\n[ REBASED ] {len(self.rebased)} tests.\n"
	for name in self.rebased:
	res += f"[ REBASED ] {name}\n"
	return res


	# Responsible for executing singular diff test.
	@dataclass
	class TestCaseRunner:
	test: TestCase
	trace_processor_path: str
	trace_descriptor_path: str
	colors: ColorFormatter

	def __output_to_text_proto(self, actual: str, out: BinaryProto) -> str:
	"""Deserializes a binary proto and returns its text representation.

	Args:
	actual: (string) HEX encoded serialized proto message
	message_type: (string) Message type

	Returns:
	Text proto
	"""
	try:
	raw_data = unhexlify(actual.splitlines()[-1][1:-1])
	out_path = os.path.dirname(self.trace_processor_path)
	descriptor_paths = [
	f.path
	for f in os.scandir(
	os.path.join(ROOT_DIR, out_path, 'gen', 'protos', 'perfetto',
	'trace_processor'))
	if f.is_file() and os.path.splitext(f.name)[1] == '.descriptor'
	]
	descriptor_paths.append(
	os.path.join(ROOT_DIR, out_path, 'gen', 'protos', 'third_party',
	'pprof', 'profile.descriptor'))
	proto = create_message_factory(descriptor_paths, out.message_type)()
	proto.ParseFromString(raw_data)
	try:
	return out.post_processing(proto)
	except:
	return '<Proto post processing failed>'
	return text_format.MessageToString(proto)
	except:
	return '<Invalid input for proto deserializaiton>'

	def __run_metrics_test(self, trace_path: str,
	metrics_message_factory) -> TestResult:

	if self.test.blueprint.is_out_file():
	with open(self.test.expected_path, 'r') as expected_file:
	expected = expected_file.read()
	else:
	expected = self.test.blueprint.out.contents

	tmp_perf_file = tempfile.NamedTemporaryFile(delete=False)
	is_json_output_file = self.test.blueprint.is_out_file(
	) and os.path.basename(self.test.expected_path).endswith('.json.out')
	is_json_output = is_json_output_file or self.test.blueprint.is_out_json()
	cmd = [
	self.trace_processor_path,
	'--analyze-trace-proto-content',
	'--crop-track-events',
	'--run-metrics',
	self.test.blueprint.query.name,
	'--metrics-output=%s' % ('json' if is_json_output else 'binary'),
	'--perf-file',
	tmp_perf_file.name,
	trace_path,
	]
	tp = subprocess.Popen(
	cmd,
	stdout=subprocess.PIPE,
	stderr=subprocess.PIPE,
	env=get_env(ROOT_DIR))
	(stdout, stderr) = tp.communicate()

	if is_json_output:
	expected_text = expected
	actual_text = stdout.decode('utf8')
	else:
	# Expected will be in text proto format and we'll need to parse it to
	# a real proto.
	expected_message = metrics_message_factory()
	text_format.Merge(expected, expected_message)

	# Actual will be the raw bytes of the proto and we'll need to parse it
	# into a message.
	actual_message = metrics_message_factory()
	actual_message.ParseFromString(stdout)

	# Convert both back to text format.
	expected_text = text_format.MessageToString(expected_message)
	actual_text = text_format.MessageToString(actual_message)

	perf_lines = [line.decode('utf8') for line in tmp_perf_file.readlines()]
	tmp_perf_file.close()
	os.remove(tmp_perf_file.name)
	return TestResult(self.test, trace_path, cmd, expected_text, actual_text,
	stderr.decode('utf8'), tp.returncode, perf_lines)

	# Run a query based Diff Test.
	def __run_query_test(self, trace_path: str) -> TestResult:
	# Fetch expected text.
	if self.test.expected_path:
	with open(self.test.expected_path, 'r') as expected_file:
	expected = expected_file.read()
	else:
	expected = self.test.blueprint.out.contents

	# Fetch query.
	if self.test.blueprint.is_query_file():
	query = self.test.query_path
	else:
	tmp_query_file = tempfile.NamedTemporaryFile(delete=False)
	with open(tmp_query_file.name, 'w') as query_file:
	query_file.write(self.test.blueprint.query)
	query = tmp_query_file.name

	tmp_perf_file = tempfile.NamedTemporaryFile(delete=False)
	cmd = [
	self.trace_processor_path,
	'--analyze-trace-proto-content',
	'--crop-track-events',
	'-q',
	query,
	'--perf-file',
	tmp_perf_file.name,
	trace_path,
	]
	tp = subprocess.Popen(
	cmd,
	stdout=subprocess.PIPE,
	stderr=subprocess.PIPE,
	env=get_env(ROOT_DIR))
	(stdout, stderr) = tp.communicate()

	if not self.test.blueprint.is_query_file():
	tmp_query_file.close()
	os.remove(tmp_query_file.name)
	perf_lines = [line.decode('utf8') for line in tmp_perf_file.readlines()]
	tmp_perf_file.close()
	os.remove(tmp_perf_file.name)

	actual = stdout.decode('utf8')
	if self.test.blueprint.is_out_binaryproto():
	actual = self.__output_to_text_proto(actual, self.test.blueprint.out)

	return TestResult(self.test, trace_path, cmd, expected, actual,
	stderr.decode('utf8'), tp.returncode, perf_lines)

	def __run(self, metrics_descriptor_paths: List[str],
	extension_descriptor_paths: List[str], keep_input,
	rebase) -> Tuple[TestResult, str]:
	# We can't use delete=True here. When using that on Windows, the
	# resulting file is opened in exclusive mode (in turn that's a subtle
	# side-effect of the underlying CreateFile(FILE_ATTRIBUTE_TEMPORARY))
	# and TP fails to open the passed path.
	gen_trace_file = None
	if self.test.blueprint.is_trace_file():
	if self.test.trace_path.endswith('.py'):
	gen_trace_file = tempfile.NamedTemporaryFile(delete=False)
	serialize_python_trace(ROOT_DIR, self.trace_descriptor_path,
	self.test.trace_path, gen_trace_file)

	elif self.test.trace_path.endswith('.textproto'):
	gen_trace_file = tempfile.NamedTemporaryFile(delete=False)
	serialize_textproto_trace(self.trace_descriptor_path,
	extension_descriptor_paths,
	self.test.trace_path, gen_trace_file)

	elif self.test.blueprint.is_trace_textproto():
	gen_trace_file = tempfile.NamedTemporaryFile(delete=False)
	proto = create_message_factory([self.trace_descriptor_path] +
	extension_descriptor_paths,
	'perfetto.protos.Trace')()
	text_format.Merge(self.test.blueprint.trace.contents, proto)
	gen_trace_file.write(proto.SerializeToString())
	gen_trace_file.flush()

	else:
	gen_trace_file = tempfile.NamedTemporaryFile(delete=False)
	with open(gen_trace_file.name, 'w') as trace_file:
	trace_file.write(self.test.blueprint.trace.contents)

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

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

	if self.test.type == TestType.QUERY:
	result = self.__run_query_test(trace_path)
	elif self.test.type == TestType.METRIC:
	result = self.__run_metrics_test(
	trace_path,
	create_message_factory(metrics_descriptor_paths,
	'perfetto.protos.TraceMetrics'))
	else:
	assert False

	if gen_trace_file:
	if keep_input:
	str += f"Saving generated input trace: {trace_path}\n"
	else:
	gen_trace_file.close()
	os.remove(trace_path)

	def write_cmdlines():
	res = ""
	if not gen_trace_file:
	res += 'Command to generate trace:\n'
	res += 'tools/serialize_test_trace.py '
	res += '--descriptor {} {} > {}\n'.format(
	os.path.relpath(self.trace_descriptor_path, ROOT_DIR),
	os.path.relpath(self.test.trace_path, ROOT_DIR),
	os.path.relpath(trace_path, ROOT_DIR))
	res += f"Command line:\n{' '.join(result.cmd)}\n"
	return res

	if result.exit_code != 0 or not result.passed:
	str += result.stderr

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

	str += (f"{self.colors.red('[ FAILED ]')} {self.test.name}\n")
	str += result.rebase(rebase)

	return result, str
	else:
	str += (f"{self.colors.green('[ OK ]')} {self.test.name} "
	f"(ingest: {result.perf_result.ingest_time_ns / 1000000:.2f} ms "
	f"query: {result.perf_result.real_time_ns / 1000000:.2f} ms)\n")
	return result, str

	# Run a TestCase.
	def execute(self, extension_descriptor_paths: List[str],
	metrics_descriptor: str, keep_input: bool,
	rebase: bool) -> Tuple[str, str, TestResult]:
	if metrics_descriptor:
	metrics_descriptor_paths = [metrics_descriptor]
	else:
	out_path = os.path.dirname(self.trace_processor_path)
	metrics_protos_path = os.path.join(out_path, 'gen', 'protos', 'perfetto',
	'metrics')
	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')
	]
	result_str = ""

	result, run_str = self.__run(metrics_descriptor_paths,
	extension_descriptor_paths, keep_input, rebase)
	result_str += run_str
	if not result:
	return self.test.name, result_str, None

	return self.test.name, result_str, result


	# Fetches and executes all diff viable tests.
	@dataclass
	class DiffTestsRunner:
	tests: List[TestCase]
	trace_processor_path: str
	trace_descriptor_path: str
	test_runners: List[TestCaseRunner]

	def __init__(self, name_filter: str, trace_processor_path: str,
	trace_descriptor: str, no_colors: bool):
	self.tests = read_all_tests(name_filter, ROOT_DIR)
	self.trace_processor_path = trace_processor_path

	out_path = os.path.dirname(self.trace_processor_path)
	self.trace_descriptor_path = get_trace_descriptor_path(
	out_path, trace_descriptor)
	self.test_runners = []
	color_formatter = ColorFormatter(no_colors)
	for test in self.tests:
	self.test_runners.append(
	TestCaseRunner(test, self.trace_processor_path,
	self.trace_descriptor_path, color_formatter))

	def run_all_tests(self, metrics_descriptor: str, keep_input: bool,
	rebase: bool) -> TestResults:
	perf_results = []
	failures = []
	rebased = []
	test_run_start = datetime.datetime.now()

	out_path = os.path.dirname(self.trace_processor_path)
	chrome_extensions = os.path.join(out_path, 'gen', 'protos', 'third_party',
	'chromium',
	'chrome_track_event.descriptor')
	test_extensions = os.path.join(out_path, 'gen', 'protos', 'perfetto',
	'trace', 'test_extensions.descriptor')

	with concurrent.futures.ProcessPoolExecutor() as e:
	fut = [
	e.submit(test.execute, [chrome_extensions, test_extensions],
	metrics_descriptor, keep_input, rebase)
	for test in self.test_runners
	]
	for res in concurrent.futures.as_completed(fut):
	test_name, res_str, result = res.result()
	sys.stderr.write(res_str)
	if not result or not result.passed:
	if rebase:
	rebased.append(test_name)
	failures.append(test_name)
	else:
	perf_results.append(result.perf_result)
	test_time_ms = int(
	(datetime.datetime.now() - test_run_start).total_seconds() * 1000)
	return TestResults(failures, perf_results, rebased, test_time_ms)