blob: d7329c678c1c2da4f921d901b15213d83143c972 [file] [log] [blame]
#!/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())