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.
 17Otherwise, the file's basename will be used.
 18
 19The expected output or any argument can reference the test's current temp or data dirs with `__TEMP_DIR__()` or `__DATA_DIR__()`, respectively.
 20An optional slash-separated path can be used as an argument to reference a path within those base directories.
 21For example, `__DATA_DIR__(foo/bar.txt)` references `bar.txt` inside the `foo` directory inside the data directory.
 22"""
 23
 24import contextlib
 25import glob
 26import io
 27import os
 28import re
 29import sys
 30import typing
 31
 32import edq.testing.asserts
 33import edq.testing.unittest
 34import edq.util.dirent
 35import edq.util.json
 36import edq.util.pyimport
 37
 38TEST_CASE_SEP: str = '---'
 39OUTPUT_SEP: str = '+++'
 40DATA_DIR_ID: str = '__DATA_DIR__'
 41ABS_DATA_DIR_ID: str = '__ABS_DATA_DIR__'
 42TEMP_DIR_ID: str = '__TEMP_DIR__'
 43BASE_DIR_ID: str = '__BASE_DIR__'
 44
 45OS_PATH_SEP: str = '__OS_PATH_SEP__'
 46
 47REPLACE_LIMIT: int = 10000
 48""" The maximum number of replacements that will be made with a single test replacement. """
 49
 50DEFAULT_ASSERTION_FUNC_NAME: str = 'edq.testing.asserts.content_equals_normalize'
 51
 52BASE_TEMP_DIR_ATTR: str = '_edq_cli_base_test_dir'
 53
 54@typing.runtime_checkable
 55class SetupTeardownFunction(typing.Protocol):
 56    """
 57    A function used to perform setup or teardown around a CLI test.
 58    """
 59
 60    def __call__(self,
 61            test: edq.testing.unittest.BaseTest,
 62            test_info: 'CLITestInfo',
 63            ) -> None:
 64        """
 65        Setup or teardown function to run before/after a CLI test.
 66        """
 67
 68class CLITestInfo:
 69    """ The required information to run a CLI test. """
 70
 71    def __init__(self,
 72            test_name: str,
 73            base_dir: str,
 74            data_dir: str,
 75            temp_dir: str,
 76            work_dir: typing.Union[str, None] = None,
 77            setup_funcs: typing.Union[typing.List[str], str, None] = None,
 78            teardown_funcs: typing.Union[typing.List[str], str, None] = None,
 79            cli: typing.Union[str, None] = None,
 80            arguments: typing.Union[typing.List[str], None] = None,
 81            error: bool = False,
 82            platform_skip: typing.Union[str, None] = None,
 83            stdout_assertion_func: typing.Union[str, None] = DEFAULT_ASSERTION_FUNC_NAME,
 84            stderr_assertion_func: typing.Union[str, None] = None,
 85            expected_stdout: str = '',
 86            expected_stderr: str = '',
 87            split_stdout_stderr: bool = False,
 88            strip_error_output: bool = True,
 89            extra_options: typing.Union[typing.Dict[str, typing.Any], None] = None,
 90            **kwargs: typing.Any,
 91            ) -> None:
 92        self.skip_reasons: typing.List[str] = []
 93        """
 94        Reasons that this test will be skipped.
 95        Any entries in this list indicate that the test should be skipped.
 96        """
 97
 98        self.platform_skip_pattern: typing.Union[str, None] = platform_skip
 99        """
100        A pattern to check if the test should be skipped on the current platform.
101        Will be used in `re.search()` against `sys.platform`.
102        """
103
104        if ((platform_skip is not None) and re.search(platform_skip, sys.platform)):
105            self.skip_reasons.append(f"not available on platform: '{sys.platform}'")
106
107        self.test_name: str = test_name
108        """ The name of this test. """
109
110        self.base_dir: str = base_dir
111        """
112        The base directory for this test (usually the dir the CLI test file lives.
113        This is the expansion for `__BASE_DIR__` paths.
114        """
115
116        self.data_dir: str = data_dir
117        """
118        A directory that additional testing data lives in.
119        This is the expansion for `__DATA_DIR__` paths.
120        """
121
122        self.temp_dir: str = temp_dir
123        """
124        A temp directory that this test has access to.
125        This is the expansion for `__TEMP_DIR__` paths.
126        """
127
128        edq.util.dirent.mkdir(temp_dir)
129
130        if (work_dir is None):
131            work_dir = os.getcwd()
132        else:
133            work_dir = self._process_text(work_dir)
134
135        self.work_dir: str = work_dir
136        """ The directory the test runs from. """
137
138        self.setup_funcs: typing.List[SetupTeardownFunction] = []
139        """ The functions to run before the test to setup. """
140
141        if (setup_funcs is not None):
142            if (isinstance(setup_funcs, str)):
143                setup_funcs = [setup_funcs]
144
145            for setup_func in setup_funcs:
146                self.setup_funcs.append(edq.util.pyimport.fetch(setup_func))
147
148        self.teardown_funcs: typing.List[SetupTeardownFunction] = []
149        """ The functions to run after the test to cleanup. """
150
151        if (teardown_funcs is not None):
152            if (isinstance(teardown_funcs, str)):
153                teardown_funcs = [teardown_funcs]
154
155            for teardown_func in teardown_funcs:
156                self.teardown_funcs.append(edq.util.pyimport.fetch(teardown_func))
157
158        if (cli is None):
159            raise ValueError("Missing CLI module.")
160
161        self.module_name: str = cli
162        """ The name of the module to invoke. """
163
164        self.module: typing.Any = None
165        """ The module to invoke. """
166
167        if (not self.should_skip()):
168            self.module = edq.util.pyimport.import_name(self.module_name)
169
170        if (arguments is None):
171            arguments = []
172
173        self.arguments: typing.List[str] = arguments
174        """ The CLI arguments. """
175
176        self.error: bool = error
177        """ Whether or not this test is expected to be an error (raise an exception). """
178
179        self.stdout_assertion_func: typing.Union[edq.testing.asserts.StringComparisonAssertion, None] = None
180        """ The assertion func to compare the expected and actual stdout of the CLI. """
181
182        if ((stdout_assertion_func is not None) and (not self.should_skip())):
183            self.stdout_assertion_func = edq.util.pyimport.fetch(stdout_assertion_func)
184
185        self.stderr_assertion_func: typing.Union[edq.testing.asserts.StringComparisonAssertion, None] = None
186        """ The assertion func to compare the expected and actual stderr of the CLI. """
187
188        if ((stderr_assertion_func is not None) and (not self.should_skip())):
189            self.stderr_assertion_func = edq.util.pyimport.fetch(stderr_assertion_func)
190
191        self.expected_stdout: str = expected_stdout
192        """ The expected stdout. """
193
194        self.expected_stderr: str = expected_stderr
195        """ The expected stderr. """
196
197        if (error and strip_error_output):
198            self.expected_stdout = self.expected_stdout.strip()
199            self.expected_stderr = self.expected_stderr.strip()
200
201        self.split_stdout_stderr: bool = split_stdout_stderr
202        """
203        Split stdout and stderr into different strings for testing.
204        By default, these two will be combined.
205        If both are non-empty, then they will be joined like: f"{stdout}\n{OUTPUT_SEP}\n{stderr}".
206        Otherwise, only the non-empty one will be present with no separator.
207        Any stdout assertions will be applied to the combined text.
208        """
209
210        # Make any path normalizations over the arguments and expected output.
211        self.expected_stdout = self._process_text(self.expected_stdout)
212        self.expected_stderr = self._process_text(self.expected_stderr)
213        for (i, argument) in enumerate(self.arguments):
214            self.arguments[i] = self._process_text(argument)
215
216        if (extra_options is None):
217            extra_options = {}
218
219        self.extra_options: typing.Dict[str, typing.Any] = extra_options
220        """
221        A place to store additional options.
222        Extra top-level options will cause tests to error.
223        """
224
225        if (len(kwargs) > 0):
226            raise ValueError(f"Found unknown CLI test options: '{kwargs}'.")
227
228    def _process_text(self, text: str) -> str:
229        """
230        Process text with and desired replacements.
231
232        This will expand path replacements in testing text.
233        This allows for consistent paths (even absolute paths) in the test text.
234        """
235
236        text_replacements = [
237            (OS_PATH_SEP, os.sep),
238        ]
239
240        for (target, replacement) in text_replacements:
241            text = text.replace(target, replacement)
242
243        path_replacements = [
244            (DATA_DIR_ID, self.data_dir, False),
245            (TEMP_DIR_ID, self.temp_dir, False),
246            (BASE_DIR_ID, self.base_dir, False),
247            (ABS_DATA_DIR_ID, self.data_dir, True),
248        ]
249
250        for (key, target_dir, normalize) in path_replacements:
251            text = replace_path_pattern(text, key, target_dir, normalize_path = normalize)
252
253        return text
254
255    def should_skip(self) -> bool:
256        """ Check if this test should be skipped. """
257
258        return (len(self.skip_reasons) > 0)
259
260    def skip_message(self) -> str:
261        """ Get a message displaying the reasons this test should be skipped. """
262
263        return f"This test has been skipped because of the following: {self.skip_reasons}."
264
265    @staticmethod
266    def load_path(path: str, test_name: str, base_temp_dir: str, data_dir: str) -> 'CLITestInfo':
267        """ Load a CLI test file and extract the test info. """
268
269        options, expected_stdout = read_test_file(path)
270
271        options['expected_stdout'] = expected_stdout
272
273        base_dir = os.path.dirname(os.path.abspath(path))
274        temp_dir = os.path.join(base_temp_dir, test_name)
275
276        return CLITestInfo(test_name, base_dir, data_dir, temp_dir, **options)
277
278@typing.runtime_checkable
279class TestMethodWrapperFunction(typing.Protocol):
280    """
281    A function that can be used to wrap/modify a CLI test method before it is attached to the test class.
282    """
283
284    def __call__(self,
285            test_method: typing.Callable,
286            test_info_path: str,
287            ) -> typing.Callable:
288        """
289        Wrap and/or modify the CLI test method before it is attached to the test class.
290        See _get_test_method() for the input method.
291        The returned method will be used in-place of the input one.
292        """
293
294def read_test_file(path: str) -> typing.Tuple[typing.Dict[str, typing.Any], str]:
295    """ Read a test case file and split the output into JSON data and text. """
296
297    json_lines: typing.List[str] = []
298    output_lines: typing.List[str] = []
299
300    text = edq.util.dirent.read_file(path, strip = False)
301
302    accumulator = json_lines
303    switched_accumulator = False
304
305    for line in text.split("\n"):
306        if ((not switched_accumulator) and (line.strip() == TEST_CASE_SEP)):
307            accumulator = output_lines
308            switched_accumulator = True
309            continue
310
311        accumulator.append(line)
312
313    options = edq.util.json.loads(''.join(json_lines))
314    output = "\n".join(output_lines)
315
316    return options, output
317
318def replace_path_pattern(text: str, key: str, target_dir: str, normalize_path: bool = False) -> str:
319    """ Make any test replacement inside the given string. """
320
321    for _ in range(REPLACE_LIMIT):
322        match = re.search(rf'{key}\(([^)]*)\)', text)
323        if (match is None):
324            break
325
326        filename = match.group(1)
327
328        # Normalize any path separators.
329        filename = os.path.join(*filename.split('/'))
330
331        if (filename == ''):
332            path = target_dir
333        else:
334            path = os.path.join(target_dir, filename)
335
336        if (normalize_path):
337            path = os.path.abspath(path)
338
339        text = text.replace(match.group(0), path)
340
341    return text
342
343def compute_ancestor_basename(path: str, cli_tests_dir: str) -> str:
344    """
345    Get the test's name based off of its filename and location.
346    A useful function to use in get_test_basename().
347    """
348
349    path = os.path.abspath(path)
350
351    name = os.path.splitext(os.path.basename(path))[0]
352
353    # Clean drive identifiers (for Windows).
354    cli_tests_dir_path = os.path.splitdrive(os.path.abspath(cli_tests_dir))[1]
355    path = os.path.splitdrive(path)[1]
356
357    ancestors = os.path.dirname(path).replace(cli_tests_dir_path, '')
358    prefix = ancestors.replace(os.sep, '_')
359
360    if (prefix.startswith('_')):
361        prefix = prefix.replace('_', '', 1)
362
363    if (len(prefix) > 0):
364        name = f"{prefix}_{name}"
365
366    return name
367
368
369def _get_test_method(test_name: str, path: str, data_dir: str) -> typing.Callable:
370    """ Get a test method that represents the test case at the given path. """
371
372    def __method(self: edq.testing.unittest.BaseTest,
373            reraise_exception_types: typing.Union[typing.Tuple[typing.Type], None] = None,
374            **kwargs: typing.Any,
375            ) -> None:
376        test_info = CLITestInfo.load_path(path, test_name, getattr(self, BASE_TEMP_DIR_ATTR), data_dir)
377
378        # Allow the test class a chance to modify the test info before the test runs.
379        if (hasattr(self, 'modify_cli_test_info')):
380            self.modify_cli_test_info(test_info)
381
382        if (test_info.should_skip()):
383            self.skipTest(test_info.skip_message())
384
385        for setup_func in test_info.setup_funcs:
386            setup_func(self, test_info)
387
388        # Check for skipping once more after any setup funcs are called.
389        if (test_info.should_skip()):
390            for teardown_func in test_info.teardown_funcs:
391                teardown_func(self, test_info)
392
393            self.skipTest(test_info.skip_message())
394
395        old_args = sys.argv
396        sys.argv = [test_info.module.__file__] + test_info.arguments
397
398        previous_work_directory = os.getcwd()
399        os.chdir(test_info.work_dir)
400
401        try:
402            with contextlib.redirect_stdout(io.StringIO()) as stdout_output:
403                with contextlib.redirect_stderr(io.StringIO()) as stderr_output:
404                    test_info.module.main()
405
406            stdout_text = stdout_output.getvalue()
407            stderr_text = stderr_output.getvalue()
408
409            if (test_info.error):
410                self.fail(f"No error was not raised when one was expected ('{str(test_info.expected_stdout)}').")
411        except BaseException as ex:
412            if ((reraise_exception_types is not None) and isinstance(ex, reraise_exception_types)):
413                raise ex
414
415            if (not test_info.error):
416                raise ex
417
418            stdout_text = self.format_error_string(ex)
419
420            stderr_text = ''
421            if (isinstance(ex, SystemExit) and (ex.__context__ is not None)):
422                stderr_text = self.format_error_string(ex.__context__)
423        finally:
424            os.chdir(previous_work_directory)
425            sys.argv = old_args
426
427            for teardown_func in test_info.teardown_funcs:
428                teardown_func(self, test_info)
429
430        if (not test_info.split_stdout_stderr):
431            if ((len(stdout_text) > 0) and (len(stderr_text) > 0)):
432                stdout_text = f"{stdout_text}\n{OUTPUT_SEP}\n{stderr_text}"
433            elif (len(stderr_text) > 0):
434                stdout_text = stderr_text
435
436        if (test_info.stdout_assertion_func is not None):
437            test_info.stdout_assertion_func(self, test_info.expected_stdout, stdout_text)
438
439        if (test_info.stderr_assertion_func is not None):
440            test_info.stderr_assertion_func(self, test_info.expected_stderr, stderr_text)
441
442    return __method
443
444def add_test_paths(target_class: type, data_dir: str, paths: typing.List[str],
445        test_method_wrapper: typing.Union[TestMethodWrapperFunction, None] = None) -> None:
446    """ Add tests from the given test files. """
447
448    # Attach a temp directory to the testing class so all tests can share a common base temp dir.
449    if (not hasattr(target_class, BASE_TEMP_DIR_ATTR)):
450        setattr(target_class, BASE_TEMP_DIR_ATTR, edq.util.dirent.get_temp_path('edq_cli_test_'))
451
452    for path in sorted(paths):
453        basename = os.path.splitext(os.path.basename(path))[0]
454        if (hasattr(target_class, 'get_test_basename')):
455            basename = getattr(target_class, 'get_test_basename')(path)
456
457        test_name = 'test_cli__' + basename
458
459        try:
460            test_method = _get_test_method(test_name, path, data_dir)
461        except Exception as ex:
462            raise ValueError(f"Failed to parse test case '{path}'.") from ex
463
464        if (test_method_wrapper is not None):
465            test_method = test_method_wrapper(test_method, path)
466
467        setattr(target_class, test_name, test_method)
468
469def discover_test_cases(target_class: type, test_cases_dir: str, data_dir: str,
470        test_method_wrapper: typing.Union[TestMethodWrapperFunction, None] = None) -> None:
471    """ Look in the text cases directory for any test cases and add them as test methods to the test class. """
472
473    paths = list(sorted(glob.glob(os.path.join(test_cases_dir, "**", "*.txt"), recursive = True)))
474    add_test_paths(target_class, data_dir, paths, test_method_wrapper = test_method_wrapper)
475
476def setup_teardown_copy(
477    test: edq.testing.unittest.BaseTest,
478    test_info: CLITestInfo,
479    ) -> None:
480    """
481    A setup/teardown function for copying dirents.
482
483    Will read copy operands from the list `test_info.extra_options['copy']` as two-item tuples (source, dest).
484    Paths should be absolute or relative to the test's work dir and use POSIX-style path separators.
485    """
486
487    for (source, dest) in test_info.extra_options.get('copy', []):
488        source = test_info._process_text(source)
489        dest = test_info._process_text(dest)
490
491        source = os.sep.join(source.split('/'))
492        if (not os.path.isabs(source)):
493            source = os.path.join(test_info.work_dir, source)
494
495        dest = os.sep.join(dest.split('/'))
496        if (not os.path.isabs(dest)):
497            dest = os.path.join(test_info.work_dir, dest)
498
499        edq.util.dirent.copy(source, dest)
500
501def create_directory_structure(
502        test: edq.testing.unittest.BaseTest,
503        test_info: CLITestInfo,
504        ) -> None:
505    """
506    A version of edq.testing.unittest.create_directory_structure() callable from a CLI test.
507    The test's temp dir will be used as the base dir,
508    and 'directory_structure' on the extra options will be used as the spec.
509    """
510
511    edq.testing.unittest.create_directory_structure(test_info.extra_options.get('directory_structure', []), test_info.temp_dir)
512
513def check_paths_or_skip(
514        test: edq.testing.unittest.BaseTest,
515        test_info: CLITestInfo,
516        ) -> None:
517    """
518    Check for any required paths listed in `test_info.extra_options['paths_or_skip']`.
519    If a path does not exist, skip this test.
520    """
521
522    for path in test_info.extra_options.get('paths_or_skip', []):
523        path = test_info._process_text(path)
524        if (not os.path.exists(path)):
525            test_info.skip_reasons.append(f"Path does not exist: '{path}'.")