123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299 |
- # Copyright 2021 The Chromium Authors. All rights reserved.
- # Use of this source code is governed by a BSD-style license that can be
- # found in the LICENSE file.
- """Integration tests."""
- import argparse
- import builtins
- from collections import defaultdict
- import difflib
- from functools import wraps
- import glob
- import json
- import os
- import unittest
- import sys
- import tempfile
- import pprint
- from typing import List, Dict
- import disable
- import resultdb
- def cmd_record(args: argparse.Namespace):
- record_testcase(args.name, ['./disable'] + args.args)
- def record_testcase(name: str, testcase_args: List[str]):
- # While running the test, point CANNED_RESPONSE_FILE to a temporary that we
- # can recover data from afterwards.
- fd, temp_canned_response_file = tempfile.mkstemp()
- os.fdopen(fd).close()
- resultdb.CANNED_RESPONSE_FILE = temp_canned_response_file
- original_open = builtins.open
- builtins.open = opener(original_open)
- try:
- disable.main(testcase_args)
- # TODO: We probably want to test failure cases as well. We can add an
- # "exception" field to the testcase JSON and test that the same exception is
- # raised.
- finally:
- builtins.open = original_open
- with open(temp_canned_response_file) as f:
- recorded_requests = f.read()
- os.remove(temp_canned_response_file)
- testcase = {
- 'args': testcase_args,
- 'requests': recorded_requests,
- 'read_data': TrackingFile.read_data,
- 'written_data': TrackingFile.written_data,
- }
- print(f'Recorded testcase {name}.\nDiff from this testcase is:\n')
- print_diffs(TrackingFile.read_data, TrackingFile.written_data)
- with open(os.path.join('tests', f'{name}.json'), 'w') as f:
- json.dump(testcase, f, indent=2)
- TrackingFile.read_data.clear()
- TrackingFile.written_data.clear()
- def print_diffs(read_data: Dict[str, str], written_data: Dict[str, str]):
- def lines(s: str) -> List[str]:
- return [line + '\n' for line in s.split('\n')]
- for filename in read_data:
- if filename in written_data:
- before = lines(read_data[filename])
- after = lines(written_data[filename])
- sys.stdout.writelines(
- difflib.unified_diff(before,
- after,
- fromfile=f'a/{filename}',
- tofile=f'b/{filename}'))
- def opener(old_open):
- @wraps(old_open)
- def tracking_open(path, mode='r', **kwargs):
- if os.path.abspath(path).startswith(disable.SRC_ROOT):
- return TrackingFile(old_open, path, mode, **kwargs)
- return old_open(path, mode, **kwargs)
- return tracking_open
- class TrackingFile:
- """A file-like class that records what data was read/written."""
- read_data = {}
- written_data = defaultdict(str)
- def __init__(self, old_open, path, mode, **kwargs):
- self.path = path
- if mode != 'w':
- self.file = old_open(path, mode, **kwargs)
- else:
- self.file = None
- def read(self, n_bytes=-1):
- # It's easier to stash all the results if we only deal with the case where
- # all the data is read at once. Right now we can get away with this as the
- # tool only does this, but if that changes we'll need to support it here.
- assert n_bytes == -1
- data = self.file.read(n_bytes)
- TrackingFile.read_data[src_root_relative(self.path)] = data
- return data
- def write(self, data):
- # Don't actually write the data, since we're just recording a testcase.
- TrackingFile.written_data[src_root_relative(self.path)] += data
- def __enter__(self):
- return self
- def __exit__(self, e_type, e_val, e_tb):
- if self.file is not None:
- self.file.close()
- self.file = None
- def src_root_relative(path: str) -> str:
- if os.path.abspath(path).startswith(disable.SRC_ROOT):
- return os.path.relpath(path, disable.SRC_ROOT)
- return path
- class IntegrationTest(unittest.TestCase):
- """This class represents a data-driven integration test.
- Given a list of arguments to pass to the test disabler, a set of ResultDB
- requests and responses to replay, and the data read/written to the filesystem,
- run the test disabler in a hermetic test environment and check that the output
- is the same.
- """
- def __init__(self, name, args, requests, read_data, written_data):
- unittest.TestCase.__init__(self, methodName='test_one_testcase')
- self.name = name
- self.args = args
- self.requests = requests
- self.read_data = read_data
- self.written_data = written_data
- def test_one_testcase(self):
- fd, temp_canned_response_file = tempfile.mkstemp(text=True)
- os.fdopen(fd).close()
- with open(temp_canned_response_file, 'w') as f:
- f.write(self.requests)
- resultdb.CANNED_RESPONSE_FILE = temp_canned_response_file
- TrackingFile.read_data.clear()
- TrackingFile.written_data.clear()
- with tempfile.TemporaryDirectory() as temp_dir:
- disable.SRC_ROOT = temp_dir
- for filename, contents in self.read_data.items():
- in_temp = os.path.join(temp_dir, filename)
- os.makedirs(os.path.dirname(in_temp))
- with open(in_temp, 'w') as f:
- f.write(contents)
- original_open = builtins.open
- builtins.open = opener(original_open)
- try:
- disable.main(self.args)
- finally:
- os.remove(temp_canned_response_file)
- builtins.open = original_open
- for path, data in TrackingFile.written_data.items():
- if path == temp_canned_response_file:
- continue
- relpath = src_root_relative(path)
- self.assertIn(relpath, self.written_data)
- self.assertEqual(data, self.written_data[relpath])
- def shortDescription(self):
- return self.name
- def cmd_show(args: argparse.Namespace):
- try:
- with open(os.path.join('tests', f'{args.name}.json'), 'r') as f:
- testcase = json.load(f)
- except FileNotFoundError:
- print(f"No such testcase '{args.name}'", file=sys.stderr)
- sys.exit(1)
- command_line = ' '.join(testcase['args'])
- print(f'Testcase {args.name}, invokes disabler with:\n{command_line}\n\n')
- # Pretty-print ResultDB RPC requests and corresponding responses.
- requests = json.loads(testcase['requests'])
- if len(requests) != 0:
- print(f'Makes {len(requests)} request(s) to ResultDB:')
- for request, response in requests.items():
- n = request.index('/')
- name = request[:n]
- payload = json.loads(request[n + 1:])
- print(f'\n{name}')
- pprint.pprint(payload)
- print('->')
- pprint.pprint(json.loads(response))
- print('\n')
- # List all files read.
- read_data = testcase['read_data']
- if len(read_data) > 0:
- print(f'Reads {len(read_data)} file(s):')
- print('\n'.join(read_data))
- print('\n')
- # Show diff between read and written for all written files.
- written_data = testcase['written_data']
- if len(written_data) > 0:
- print('Produces the following diffs:')
- print_diffs(read_data, written_data)
- def all_testcase_jsons():
- for testcase in glob.glob('tests/*.json'):
- with open(testcase, 'r') as f:
- yield os.path.basename(testcase)[:-5], json.load(f)
- def cmd_run(_args: argparse.Namespace):
- testcases = []
- for name, testcase_json in all_testcase_jsons():
- testcases.append(
- IntegrationTest(
- name,
- testcase_json['args'],
- testcase_json['requests'],
- testcase_json['read_data'],
- testcase_json['written_data'],
- ))
- test_runner = unittest.TextTestRunner()
- test_runner.run(unittest.TestSuite(testcases))
- def cmd_rerecord(_args: argparse.Namespace):
- for name, testcase_json in all_testcase_jsons():
- record_testcase(name, testcase_json['args'])
- print('')
- def main():
- parser = argparse.ArgumentParser(
- description='Record / replay integration tests.', )
- subparsers = parser.add_subparsers()
- record_parser = subparsers.add_parser('record', help='Record a testcase')
- record_parser.add_argument('name',
- type=str,
- help='The name to give the testcase')
- record_parser.add_argument(
- 'args',
- type=str,
- nargs='+',
- help='The arguments to use for running the testcase.')
- record_parser.set_defaults(func=cmd_record)
- run_parser = subparsers.add_parser('run', help='Run all testcases')
- run_parser.set_defaults(func=cmd_run)
- show_parser = subparsers.add_parser('show', help='Describe a testcase')
- show_parser.add_argument('name', type=str, help='The testcase to describe')
- show_parser.set_defaults(func=cmd_show)
- rerecord_parser = subparsers.add_parser(
- 'rerecord', help='Re-record all existing testcases')
- rerecord_parser.set_defaults(func=cmd_rerecord)
- args = parser.parse_args()
- args.func(args)
- if __name__ == '__main__':
- main()
|