integration_test.py 8.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299
  1. # Copyright 2021 The Chromium Authors. All rights reserved.
  2. # Use of this source code is governed by a BSD-style license that can be
  3. # found in the LICENSE file.
  4. """Integration tests."""
  5. import argparse
  6. import builtins
  7. from collections import defaultdict
  8. import difflib
  9. from functools import wraps
  10. import glob
  11. import json
  12. import os
  13. import unittest
  14. import sys
  15. import tempfile
  16. import pprint
  17. from typing import List, Dict
  18. import disable
  19. import resultdb
  20. def cmd_record(args: argparse.Namespace):
  21. record_testcase(args.name, ['./disable'] + args.args)
  22. def record_testcase(name: str, testcase_args: List[str]):
  23. # While running the test, point CANNED_RESPONSE_FILE to a temporary that we
  24. # can recover data from afterwards.
  25. fd, temp_canned_response_file = tempfile.mkstemp()
  26. os.fdopen(fd).close()
  27. resultdb.CANNED_RESPONSE_FILE = temp_canned_response_file
  28. original_open = builtins.open
  29. builtins.open = opener(original_open)
  30. try:
  31. disable.main(testcase_args)
  32. # TODO: We probably want to test failure cases as well. We can add an
  33. # "exception" field to the testcase JSON and test that the same exception is
  34. # raised.
  35. finally:
  36. builtins.open = original_open
  37. with open(temp_canned_response_file) as f:
  38. recorded_requests = f.read()
  39. os.remove(temp_canned_response_file)
  40. testcase = {
  41. 'args': testcase_args,
  42. 'requests': recorded_requests,
  43. 'read_data': TrackingFile.read_data,
  44. 'written_data': TrackingFile.written_data,
  45. }
  46. print(f'Recorded testcase {name}.\nDiff from this testcase is:\n')
  47. print_diffs(TrackingFile.read_data, TrackingFile.written_data)
  48. with open(os.path.join('tests', f'{name}.json'), 'w') as f:
  49. json.dump(testcase, f, indent=2)
  50. TrackingFile.read_data.clear()
  51. TrackingFile.written_data.clear()
  52. def print_diffs(read_data: Dict[str, str], written_data: Dict[str, str]):
  53. def lines(s: str) -> List[str]:
  54. return [line + '\n' for line in s.split('\n')]
  55. for filename in read_data:
  56. if filename in written_data:
  57. before = lines(read_data[filename])
  58. after = lines(written_data[filename])
  59. sys.stdout.writelines(
  60. difflib.unified_diff(before,
  61. after,
  62. fromfile=f'a/{filename}',
  63. tofile=f'b/{filename}'))
  64. def opener(old_open):
  65. @wraps(old_open)
  66. def tracking_open(path, mode='r', **kwargs):
  67. if os.path.abspath(path).startswith(disable.SRC_ROOT):
  68. return TrackingFile(old_open, path, mode, **kwargs)
  69. return old_open(path, mode, **kwargs)
  70. return tracking_open
  71. class TrackingFile:
  72. """A file-like class that records what data was read/written."""
  73. read_data = {}
  74. written_data = defaultdict(str)
  75. def __init__(self, old_open, path, mode, **kwargs):
  76. self.path = path
  77. if mode != 'w':
  78. self.file = old_open(path, mode, **kwargs)
  79. else:
  80. self.file = None
  81. def read(self, n_bytes=-1):
  82. # It's easier to stash all the results if we only deal with the case where
  83. # all the data is read at once. Right now we can get away with this as the
  84. # tool only does this, but if that changes we'll need to support it here.
  85. assert n_bytes == -1
  86. data = self.file.read(n_bytes)
  87. TrackingFile.read_data[src_root_relative(self.path)] = data
  88. return data
  89. def write(self, data):
  90. # Don't actually write the data, since we're just recording a testcase.
  91. TrackingFile.written_data[src_root_relative(self.path)] += data
  92. def __enter__(self):
  93. return self
  94. def __exit__(self, e_type, e_val, e_tb):
  95. if self.file is not None:
  96. self.file.close()
  97. self.file = None
  98. def src_root_relative(path: str) -> str:
  99. if os.path.abspath(path).startswith(disable.SRC_ROOT):
  100. return os.path.relpath(path, disable.SRC_ROOT)
  101. return path
  102. class IntegrationTest(unittest.TestCase):
  103. """This class represents a data-driven integration test.
  104. Given a list of arguments to pass to the test disabler, a set of ResultDB
  105. requests and responses to replay, and the data read/written to the filesystem,
  106. run the test disabler in a hermetic test environment and check that the output
  107. is the same.
  108. """
  109. def __init__(self, name, args, requests, read_data, written_data):
  110. unittest.TestCase.__init__(self, methodName='test_one_testcase')
  111. self.name = name
  112. self.args = args
  113. self.requests = requests
  114. self.read_data = read_data
  115. self.written_data = written_data
  116. def test_one_testcase(self):
  117. fd, temp_canned_response_file = tempfile.mkstemp(text=True)
  118. os.fdopen(fd).close()
  119. with open(temp_canned_response_file, 'w') as f:
  120. f.write(self.requests)
  121. resultdb.CANNED_RESPONSE_FILE = temp_canned_response_file
  122. TrackingFile.read_data.clear()
  123. TrackingFile.written_data.clear()
  124. with tempfile.TemporaryDirectory() as temp_dir:
  125. disable.SRC_ROOT = temp_dir
  126. for filename, contents in self.read_data.items():
  127. in_temp = os.path.join(temp_dir, filename)
  128. os.makedirs(os.path.dirname(in_temp))
  129. with open(in_temp, 'w') as f:
  130. f.write(contents)
  131. original_open = builtins.open
  132. builtins.open = opener(original_open)
  133. try:
  134. disable.main(self.args)
  135. finally:
  136. os.remove(temp_canned_response_file)
  137. builtins.open = original_open
  138. for path, data in TrackingFile.written_data.items():
  139. if path == temp_canned_response_file:
  140. continue
  141. relpath = src_root_relative(path)
  142. self.assertIn(relpath, self.written_data)
  143. self.assertEqual(data, self.written_data[relpath])
  144. def shortDescription(self):
  145. return self.name
  146. def cmd_show(args: argparse.Namespace):
  147. try:
  148. with open(os.path.join('tests', f'{args.name}.json'), 'r') as f:
  149. testcase = json.load(f)
  150. except FileNotFoundError:
  151. print(f"No such testcase '{args.name}'", file=sys.stderr)
  152. sys.exit(1)
  153. command_line = ' '.join(testcase['args'])
  154. print(f'Testcase {args.name}, invokes disabler with:\n{command_line}\n\n')
  155. # Pretty-print ResultDB RPC requests and corresponding responses.
  156. requests = json.loads(testcase['requests'])
  157. if len(requests) != 0:
  158. print(f'Makes {len(requests)} request(s) to ResultDB:')
  159. for request, response in requests.items():
  160. n = request.index('/')
  161. name = request[:n]
  162. payload = json.loads(request[n + 1:])
  163. print(f'\n{name}')
  164. pprint.pprint(payload)
  165. print('->')
  166. pprint.pprint(json.loads(response))
  167. print('\n')
  168. # List all files read.
  169. read_data = testcase['read_data']
  170. if len(read_data) > 0:
  171. print(f'Reads {len(read_data)} file(s):')
  172. print('\n'.join(read_data))
  173. print('\n')
  174. # Show diff between read and written for all written files.
  175. written_data = testcase['written_data']
  176. if len(written_data) > 0:
  177. print('Produces the following diffs:')
  178. print_diffs(read_data, written_data)
  179. def all_testcase_jsons():
  180. for testcase in glob.glob('tests/*.json'):
  181. with open(testcase, 'r') as f:
  182. yield os.path.basename(testcase)[:-5], json.load(f)
  183. def cmd_run(_args: argparse.Namespace):
  184. testcases = []
  185. for name, testcase_json in all_testcase_jsons():
  186. testcases.append(
  187. IntegrationTest(
  188. name,
  189. testcase_json['args'],
  190. testcase_json['requests'],
  191. testcase_json['read_data'],
  192. testcase_json['written_data'],
  193. ))
  194. test_runner = unittest.TextTestRunner()
  195. test_runner.run(unittest.TestSuite(testcases))
  196. def cmd_rerecord(_args: argparse.Namespace):
  197. for name, testcase_json in all_testcase_jsons():
  198. record_testcase(name, testcase_json['args'])
  199. print('')
  200. def main():
  201. parser = argparse.ArgumentParser(
  202. description='Record / replay integration tests.', )
  203. subparsers = parser.add_subparsers()
  204. record_parser = subparsers.add_parser('record', help='Record a testcase')
  205. record_parser.add_argument('name',
  206. type=str,
  207. help='The name to give the testcase')
  208. record_parser.add_argument(
  209. 'args',
  210. type=str,
  211. nargs='+',
  212. help='The arguments to use for running the testcase.')
  213. record_parser.set_defaults(func=cmd_record)
  214. run_parser = subparsers.add_parser('run', help='Run all testcases')
  215. run_parser.set_defaults(func=cmd_run)
  216. show_parser = subparsers.add_parser('show', help='Describe a testcase')
  217. show_parser.add_argument('name', type=str, help='The testcase to describe')
  218. show_parser.set_defaults(func=cmd_show)
  219. rerecord_parser = subparsers.add_parser(
  220. 'rerecord', help='Re-record all existing testcases')
  221. rerecord_parser.set_defaults(func=cmd_rerecord)
  222. args = parser.parse_args()
  223. args.func(args)
  224. if __name__ == '__main__':
  225. main()