#!/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())
|
