tools/diff_test_trace_processor.py

#!/usr/bin/env python
# Copyright (C) 2018 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 argparse
import difflib
import glob
import importlib
import os
import subprocess
import sys
import tempfile

ROOT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
TEST_DATA_DIR = os.path.join(ROOT_DIR, "test", "trace_processor")

def trace_processor_command(trace_processor_path, trace_path, query_path):
  return [trace_processor_path, '-q', query_path, trace_path]

def main():
  parser = argparse.ArgumentParser()
  parser.add_argument('--index', type=str, help='location of index file',
                      default=os.path.join(TEST_DATA_DIR, "index"))
  parser.add_argument('--trace-descriptor', type=str,
                      help='$(dirname trace_processor)/gen/protos/trace/trace.descriptor')
  parser.add_argument('trace_processor', type=str,
                      help='location of trace processor binary')
  args = parser.parse_args()

  with open(args.index, 'r') as file:
    index_lines = file.readlines()

  if args.trace_descriptor:
    trace_descriptor_path = args.trace_descriptor
  else:
    out_path = os.path.dirname(args.trace_processor)
    trace_protos_path = os.path.join(out_path, "gen", "protos", "trace")
    trace_descriptor_path = os.path.join(trace_protos_path, "trace.descriptor")

  test_failure = 0
  index_dir = os.path.dirname(args.index)
  for line in index_lines:
    stripped = line.strip()
    if stripped.startswith('#'):
      continue
    elif not stripped:
      continue

    [trace_fname, query_fname, expected_fname] = stripped.split(' ')

    trace_path = os.path.abspath(os.path.join(index_dir, trace_fname))
    query_path = os.path.abspath(os.path.join(index_dir, query_fname))
    expected_path = os.path.abspath(os.path.join(index_dir, expected_fname))
    if not os.path.exists(trace_path):
      print("Trace file not found {}".format(trace_path))
      return 1
    elif not os.path.exists(query_path):
      print("Query file not found {}".format(query_path))
      return 1
    elif not os.path.exists(expected_path):
      print("Expected file not found {}".format(expected_path))
      return 1

    if trace_path.endswith(".py"):
      with tempfile.NamedTemporaryFile() as out:
        python_cmd = [
          "python",
          trace_path,
          trace_descriptor_path
        ]
        subprocess.check_call(
          python_cmd,
          stdout=out
        )
        cmd = trace_processor_command(
            args.trace_processor, out.name, query_path)
        actual_raw = subprocess.check_output(cmd)
    else:
      cmd = trace_processor_command(
          args.trace_processor, trace_path, query_path)
      actual_raw = subprocess.check_output(cmd)

    actual = actual_raw.decode("utf-8")
    actual_lines = actual_raw.splitlines(True)

    with open(expected_path, "r") as expected_file:
      expected = expected_file.read()
      if expected != actual:
        sys.stderr.write(
          "Expected did not match actual for trace {} and query {}\n"
          .format(trace_path, query_path))
        sys.stderr.write("Expected file: {}\n".format(expected_path))
        sys.stderr.write("Commandline: {}\n".format(' '.join(cmd)))

        expected_lines = expected.splitlines(True)
        diff = difflib.unified_diff(expected_lines, actual_lines,
                                    fromfile="expected", tofile="actual")
        for line in diff:
          sys.stderr.write(line)
        test_failure += 1

  if test_failure == 0:
    print("All tests passed successfully")
    return 0
  else:
    print("Total failures: {}".format(test_failure))
    return 1

if __name__ == '__main__':
  sys.exit(main())