edq.testing.cli

  1"""
  2Infrastructure for testing CLI tools using a JSON file which describes a test case,
  3which is essentially an invocation of a CLI tool and the expected output.
  4
  5The test case file must be a `.txt` file that live in the test cases dir.
  6The file contains two parts (separated by a line with just TEST_CASE_SEP):
  7the first part which is a JSON object (see below for available keys),
  8and a second part which is the expected text output (stdout).
  9For the keys of the JSON section, see the defaulted arguments to CLITestInfo.
 10The options JSON will be splatted into CLITestInfo's constructor.
 11
 12If a test class implements a method with the signature `modify_cli_test_info(self, test_info: CLITestInfo) -> None`,
 13then this method will be called with the test info right after the test info is read from disk.
 14
 15If a test class implements a class method with the signature `get_test_basename(cls, path: str) -> str`,
 16then this method will be called to create the base name for the test case at the given path.
 17
 18The expected output or any argument can reference the test's current temp or data dirs with `__TEMP_DIR__()` or `__DATA_DIR__()`, respectively.
 19An optional slash-separated path can be used as an argument to reference a path within those base directories.
 20For example, `__DATA_DIR__(foo/bar.txt)` references `bar.txt` inside the `foo` directory inside the data directory.
 21"""
 22
 23import contextlib
 24import glob
 25import io
 26import os
 27import re
 28import sys
 29import typing
 30
 31import edq.testing.asserts
 32import edq.testing.unittest
 33import edq.util.dirent
 34import edq.util.json
 35import edq.util.pyimport
 36
 37TEST_CASE_SEP: str = '---'
 38OUTPUT_SEP: str = '+++'
 39DATA_DIR_ID: str = '__DATA_DIR__'
 40ABS_DATA_DIR_ID: str = '__ABS_DATA_DIR__'
 41TEMP_DIR_ID: str = '__TEMP_DIR__'
 42BASE_DIR_ID: str = '__BASE_DIR__'
 43
 44REPLACE_LIMIT: int = 10000
 45""" The maximum number of replacements that will be made with a single test replacement. """
 46
 47DEFAULT_ASSERTION_FUNC_NAME: str = 'edq.testing.asserts.content_equals_normalize'
 48
 49BASE_TEMP_DIR_ATTR: str = '_edq_cli_base_test_dir'
 50
 51class CLITestInfo:
 52    """ The required information to run a CLI test. """
 53
 54    def __init__(self,
 55            test_name: str,
 56            base_dir: str,
 57            data_dir: str,
 58            temp_dir: str,
 59            cli: typing.Union[str, None] = None,
 60            arguments: typing.Union[typing.List[str], None] = None,
 61            error: bool = False,
 62            platform_skip: typing.Union[str, None] = None,
 63            stdout_assertion_func: typing.Union[str, None] = DEFAULT_ASSERTION_FUNC_NAME,
 64            stderr_assertion_func: typing.Union[str, None] = None,
 65            expected_stdout: str = '',
 66            expected_stderr: str = '',
 67            split_stdout_stderr: bool = False,
 68            strip_error_output: bool = True,
 69            extra_options: typing.Union[typing.Dict[str, typing.Any], None] = None,
 70            **kwargs: typing.Any) -> None:
 71        self.skip_reasons: typing.List[str] = []
 72        """
 73        Reasons that this test will be skipped.
 74        Any entries in this list indicate that the test should be skipped.
 75        """
 76
 77        self.platform_skip_pattern: typing.Union[str, None] = platform_skip
 78        """
 79        A pattern to check if the test should be skipped on the current platform.
 80        Will be used in `re.search()` against `sys.platform`.
 81        """
 82
 83        if ((platform_skip is not None) and re.search(platform_skip, sys.platform)):
 84            self.skip_reasons.append(f"not available on platform: '{sys.platform}'")
 85
 86        self.test_name: str = test_name
 87        """ The name of this test. """
 88
 89        self.base_dir: str = base_dir
 90        """
 91        The base directory for this test (usually the dir the CLI test file lives.
 92        This is the expansion for `__BASE_DIR__` paths.
 93        """
 94
 95        self.data_dir: str = data_dir
 96        """
 97        A directory that additional testing data lives in.
 98        This is the expansion for `__DATA_DIR__` paths.
 99        """
100
101        self.temp_dir: str = temp_dir
102        """
103        A temp directory that this test has access to.
104        This is the expansion for `__TEMP_DIR__` paths.
105        """
106
107        edq.util.dirent.mkdir(temp_dir)
108
109        if (cli is None):
110            raise ValueError("Missing CLI module.")
111
112        self.module_name: str = cli
113        """ The name of the module to invoke. """
114
115        self.module: typing.Any = None
116        """ The module to invoke. """
117
118        if (not self.should_skip()):
119            self.module = edq.util.pyimport.import_name(self.module_name)
120
121        if (arguments is None):
122            arguments = []
123
124        self.arguments: typing.List[str] = arguments
125        """ The CLI arguments. """
126
127        self.error: bool = error
128        """ Whether or not this test is expected to be an error (raise an exception). """
129
130        self.stdout_assertion_func: typing.Union[edq.testing.asserts.StringComparisonAssertion, None] = None
131        """ The assertion func to compare the expected and actual stdout of the CLI. """
132
133        if ((stdout_assertion_func is not None) and (not self.should_skip())):
134            self.stdout_assertion_func = edq.util.pyimport.fetch(stdout_assertion_func)
135
136        self.stderr_assertion_func: typing.Union[edq.testing.asserts.StringComparisonAssertion, None] = None
137        """ The assertion func to compare the expected and actual stderr of the CLI. """
138
139        if ((stderr_assertion_func is not None) and (not self.should_skip())):
140            self.stderr_assertion_func = edq.util.pyimport.fetch(stderr_assertion_func)
141
142        self.expected_stdout: str = expected_stdout
143        """ The expected stdout. """
144
145        self.expected_stderr: str = expected_stderr
146        """ The expected stderr. """
147
148        if (error and strip_error_output):
149            self.expected_stdout = self.expected_stdout.strip()
150            self.expected_stderr = self.expected_stderr.strip()
151
152        self.split_stdout_stderr: bool = split_stdout_stderr
153        """
154        Split stdout and stderr into different strings for testing.
155        By default, these two will be combined.
156        If both are non-empty, then they will be joined like: f"{stdout}\n{OUTPUT_SEP}\n{stderr}".
157        Otherwise, only the non-empty one will be present with no separator.
158        Any stdout assertions will be applied to the combined text.
159        """
160
161        # Make any path normalizations over the arguments and expected output.
162        self.expected_stdout = self._expand_paths(self.expected_stdout)
163        self.expected_stderr = self._expand_paths(self.expected_stderr)
164        for (i, argument) in enumerate(self.arguments):
165            self.arguments[i] = self._expand_paths(argument)
166
167        if (extra_options is None):
168            extra_options = {}
169
170        self.extra_options: typing.Union[typing.Dict[str, typing.Any], None] = extra_options
171        """
172        A place to store additional options.
173        Extra top-level options will cause tests to error.
174        """
175
176        if (len(kwargs) > 0):
177            raise ValueError(f"Found unknown CLI test options: '{kwargs}'.")
178
179    def _expand_paths(self, text: str) -> str:
180        """
181        Expand path replacements in testing text.
182        This allows for consistent paths (even absolute paths) in the test text.
183        """
184
185        replacements = [
186            (DATA_DIR_ID, self.data_dir, False),
187            (TEMP_DIR_ID, self.temp_dir, False),
188            (BASE_DIR_ID, self.base_dir, False),
189            (ABS_DATA_DIR_ID, self.data_dir, True),
190        ]
191
192        for (key, target_dir, normalize) in replacements:
193            text = replace_path_pattern(text, key, target_dir, normalize_path = normalize)
194
195        return text
196
197    def should_skip(self) -> bool:
198        """ Check if this test should be skipped. """
199
200        return (len(self.skip_reasons) > 0)
201
202    def skip_message(self) -> str:
203        """ Get a message displaying the reasons this test should be skipped. """
204
205        return f"This test has been skipped because of the following: {self.skip_reasons}."
206
207    @staticmethod
208    def load_path(path: str, test_name: str, base_temp_dir: str, data_dir: str) -> 'CLITestInfo':
209        """ Load a CLI test file and extract the test info. """
210
211        options, expected_stdout = read_test_file(path)
212
213        options['expected_stdout'] = expected_stdout
214
215        base_dir = os.path.dirname(os.path.abspath(path))
216        temp_dir = os.path.join(base_temp_dir, test_name)
217
218        return CLITestInfo(test_name, base_dir, data_dir, temp_dir, **options)
219
220@typing.runtime_checkable
221class TestMethodWrapperFunction(typing.Protocol):
222    """
223    A function that can be used to wrap/modify a CLI test method before it is attached to the test class.
224    """
225
226    def __call__(self,
227            test_method: typing.Callable,
228            test_info_path: str,
229            ) -> typing.Callable:
230        """
231        Wrap and/or modify the CLI test method before it is attached to the test class.
232        See _get_test_method() for the input method.
233        The returned method will be used in-place of the input one.
234        """
235
236def read_test_file(path: str) -> typing.Tuple[typing.Dict[str, typing.Any], str]:
237    """ Read a test case file and split the output into JSON data and text. """
238
239    json_lines: typing.List[str] = []
240    output_lines: typing.List[str] = []
241
242    text = edq.util.dirent.read_file(path, strip = False)
243
244    accumulator = json_lines
245    switched_accumulator = False
246
247    for line in text.split("\n"):
248        if ((not switched_accumulator) and (line.strip() == TEST_CASE_SEP)):
249            accumulator = output_lines
250            switched_accumulator = True
251            continue
252
253        accumulator.append(line)
254
255    options = edq.util.json.loads(''.join(json_lines))
256    output = "\n".join(output_lines)
257
258    return options, output
259
260def replace_path_pattern(text: str, key: str, target_dir: str, normalize_path: bool = False) -> str:
261    """ Make any test replacement inside the given string. """
262
263    for _ in range(REPLACE_LIMIT):
264        match = re.search(rf'{key}\(([^)]*)\)', text)
265        if (match is None):
266            break
267
268        filename = match.group(1)
269
270        # Normalize any path separators.
271        filename = os.path.join(*filename.split('/'))
272
273        if (filename == ''):
274            path = target_dir
275        else:
276            path = os.path.join(target_dir, filename)
277
278        if (normalize_path):
279            path = os.path.abspath(path)
280
281        text = text.replace(match.group(0), path)
282
283    return text
284
285def _get_test_method(test_name: str, path: str, data_dir: str) -> typing.Callable:
286    """ Get a test method that represents the test case at the given path. """
287
288    def __method(self: edq.testing.unittest.BaseTest,
289            reraise_exception_types: typing.Union[typing.Tuple[typing.Type], None] = None,
290            **kwargs: typing.Any) -> None:
291        test_info = CLITestInfo.load_path(path, test_name, getattr(self, BASE_TEMP_DIR_ATTR), data_dir)
292
293        # Allow the test class a chance to modify the test info before the test runs.
294        if (hasattr(self, 'modify_cli_test_info')):
295            self.modify_cli_test_info(test_info)
296
297        if (test_info.should_skip()):
298            self.skipTest(test_info.skip_message())
299
300        old_args = sys.argv
301        sys.argv = [test_info.module.__file__] + test_info.arguments
302
303        try:
304            with contextlib.redirect_stdout(io.StringIO()) as stdout_output:
305                with contextlib.redirect_stderr(io.StringIO()) as stderr_output:
306                    test_info.module.main()
307
308            stdout_text = stdout_output.getvalue()
309            stderr_text = stderr_output.getvalue()
310
311            if (test_info.error):
312                self.fail(f"No error was not raised when one was expected ('{str(test_info.expected_stdout)}').")
313        except BaseException as ex:
314            if ((reraise_exception_types is not None) and isinstance(ex, reraise_exception_types)):
315                raise ex
316
317            if (not test_info.error):
318                raise ex
319
320            stdout_text = self.format_error_string(ex)
321
322            stderr_text = ''
323            if (isinstance(ex, SystemExit) and (ex.__context__ is not None)):
324                stderr_text = self.format_error_string(ex.__context__)
325        finally:
326            sys.argv = old_args
327
328        if (not test_info.split_stdout_stderr):
329            if ((len(stdout_text) > 0) and (len(stderr_text) > 0)):
330                stdout_text = f"{stdout_text}\n{OUTPUT_SEP}\n{stderr_text}"
331            elif (len(stderr_text) > 0):
332                stdout_text = stderr_text
333
334        if (test_info.stdout_assertion_func is not None):
335            test_info.stdout_assertion_func(self, test_info.expected_stdout, stdout_text)
336
337        if (test_info.stderr_assertion_func is not None):
338            test_info.stderr_assertion_func(self, test_info.expected_stderr, stderr_text)
339
340    return __method
341
342def add_test_paths(target_class: type, data_dir: str, paths: typing.List[str],
343        test_method_wrapper: typing.Union[TestMethodWrapperFunction, None] = None) -> None:
344    """ Add tests from the given test files. """
345
346    # Attach a temp directory to the testing class so all tests can share a common base temp dir.
347    if (not hasattr(target_class, BASE_TEMP_DIR_ATTR)):
348        setattr(target_class, BASE_TEMP_DIR_ATTR, edq.util.dirent.get_temp_path('edq_cli_test_'))
349
350    for path in sorted(paths):
351        basename = os.path.splitext(os.path.basename(path))[0]
352        if (hasattr(target_class, 'get_test_basename')):
353            basename = getattr(target_class, 'get_test_basename')(path)
354
355        test_name = 'test_cli__' + basename
356
357        try:
358            test_method = _get_test_method(test_name, path, data_dir)
359        except Exception as ex:
360            raise ValueError(f"Failed to parse test case '{path}'.") from ex
361
362        if (test_method_wrapper is not None):
363            test_method = test_method_wrapper(test_method, path)
364
365        setattr(target_class, test_name, test_method)
366
367def discover_test_cases(target_class: type, test_cases_dir: str, data_dir: str,
368        test_method_wrapper: typing.Union[TestMethodWrapperFunction, None] = None) -> None:
369    """ Look in the text cases directory for any test cases and add them as test methods to the test class. """
370
371    paths = list(sorted(glob.glob(os.path.join(test_cases_dir, "**", "*.txt"), recursive = True)))
372    add_test_paths(target_class, data_dir, paths, test_method_wrapper = test_method_wrapper)