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            ) -> typing.Callable:
 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_func: typing.Union[str, None] = None,
 78            teardown_func: typing.Union[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_func: typing.Union[SetupTeardownFunction, None] = None
139        """ The function to run before the test to setup. """
140
141        if (setup_func is not None):
142            self.setup_func = edq.util.pyimport.fetch(setup_func)
143
144        self.teardown_func: typing.Union[SetupTeardownFunction, None] = None
145        """ The function to run after the test to cleanup. """
146
147        if (teardown_func is not None):
148            self.teardown_func = edq.util.pyimport.fetch(teardown_func)
149
150        if (cli is None):
151            raise ValueError("Missing CLI module.")
152
153        self.module_name: str = cli
154        """ The name of the module to invoke. """
155
156        self.module: typing.Any = None
157        """ The module to invoke. """
158
159        if (not self.should_skip()):
160            self.module = edq.util.pyimport.import_name(self.module_name)
161
162        if (arguments is None):
163            arguments = []
164
165        self.arguments: typing.List[str] = arguments
166        """ The CLI arguments. """
167
168        self.error: bool = error
169        """ Whether or not this test is expected to be an error (raise an exception). """
170
171        self.stdout_assertion_func: typing.Union[edq.testing.asserts.StringComparisonAssertion, None] = None
172        """ The assertion func to compare the expected and actual stdout of the CLI. """
173
174        if ((stdout_assertion_func is not None) and (not self.should_skip())):
175            self.stdout_assertion_func = edq.util.pyimport.fetch(stdout_assertion_func)
176
177        self.stderr_assertion_func: typing.Union[edq.testing.asserts.StringComparisonAssertion, None] = None
178        """ The assertion func to compare the expected and actual stderr of the CLI. """
179
180        if ((stderr_assertion_func is not None) and (not self.should_skip())):
181            self.stderr_assertion_func = edq.util.pyimport.fetch(stderr_assertion_func)
182
183        self.expected_stdout: str = expected_stdout
184        """ The expected stdout. """
185
186        self.expected_stderr: str = expected_stderr
187        """ The expected stderr. """
188
189        if (error and strip_error_output):
190            self.expected_stdout = self.expected_stdout.strip()
191            self.expected_stderr = self.expected_stderr.strip()
192
193        self.split_stdout_stderr: bool = split_stdout_stderr
194        """
195        Split stdout and stderr into different strings for testing.
196        By default, these two will be combined.
197        If both are non-empty, then they will be joined like: f"{stdout}\n{OUTPUT_SEP}\n{stderr}".
198        Otherwise, only the non-empty one will be present with no separator.
199        Any stdout assertions will be applied to the combined text.
200        """
201
202        # Make any path normalizations over the arguments and expected output.
203        self.expected_stdout = self._process_text(self.expected_stdout)
204        self.expected_stderr = self._process_text(self.expected_stderr)
205        for (i, argument) in enumerate(self.arguments):
206            self.arguments[i] = self._process_text(argument)
207
208        if (extra_options is None):
209            extra_options = {}
210
211        self.extra_options: typing.Union[typing.Dict[str, typing.Any], None] = extra_options
212        """
213        A place to store additional options.
214        Extra top-level options will cause tests to error.
215        """
216
217        if (len(kwargs) > 0):
218            raise ValueError(f"Found unknown CLI test options: '{kwargs}'.")
219
220    def _process_text(self, text: str) -> str:
221        """
222        Process text with and desired replacements.
223
224        This will expand path replacements in testing text.
225        This allows for consistent paths (even absolute paths) in the test text.
226        """
227
228        text_replacements = [
229            (OS_PATH_SEP, os.sep),
230        ]
231
232        for (target, replacement) in text_replacements:
233            text = text.replace(target, replacement)
234
235        path_replacements = [
236            (DATA_DIR_ID, self.data_dir, False),
237            (TEMP_DIR_ID, self.temp_dir, False),
238            (BASE_DIR_ID, self.base_dir, False),
239            (ABS_DATA_DIR_ID, self.data_dir, True),
240        ]
241
242        for (key, target_dir, normalize) in path_replacements:
243            text = replace_path_pattern(text, key, target_dir, normalize_path = normalize)
244
245        return text
246
247    def should_skip(self) -> bool:
248        """ Check if this test should be skipped. """
249
250        return (len(self.skip_reasons) > 0)
251
252    def skip_message(self) -> str:
253        """ Get a message displaying the reasons this test should be skipped. """
254
255        return f"This test has been skipped because of the following: {self.skip_reasons}."
256
257    @staticmethod
258    def load_path(path: str, test_name: str, base_temp_dir: str, data_dir: str) -> 'CLITestInfo':
259        """ Load a CLI test file and extract the test info. """
260
261        options, expected_stdout = read_test_file(path)
262
263        options['expected_stdout'] = expected_stdout
264
265        base_dir = os.path.dirname(os.path.abspath(path))
266        temp_dir = os.path.join(base_temp_dir, test_name)
267
268        return CLITestInfo(test_name, base_dir, data_dir, temp_dir, **options)
269
270@typing.runtime_checkable
271class TestMethodWrapperFunction(typing.Protocol):
272    """
273    A function that can be used to wrap/modify a CLI test method before it is attached to the test class.
274    """
275
276    def __call__(self,
277            test_method: typing.Callable,
278            test_info_path: str,
279            ) -> typing.Callable:
280        """
281        Wrap and/or modify the CLI test method before it is attached to the test class.
282        See _get_test_method() for the input method.
283        The returned method will be used in-place of the input one.
284        """
285
286def read_test_file(path: str) -> typing.Tuple[typing.Dict[str, typing.Any], str]:
287    """ Read a test case file and split the output into JSON data and text. """
288
289    json_lines: typing.List[str] = []
290    output_lines: typing.List[str] = []
291
292    text = edq.util.dirent.read_file(path, strip = False)
293
294    accumulator = json_lines
295    switched_accumulator = False
296
297    for line in text.split("\n"):
298        if ((not switched_accumulator) and (line.strip() == TEST_CASE_SEP)):
299            accumulator = output_lines
300            switched_accumulator = True
301            continue
302
303        accumulator.append(line)
304
305    options = edq.util.json.loads(''.join(json_lines))
306    output = "\n".join(output_lines)
307
308    return options, output
309
310def replace_path_pattern(text: str, key: str, target_dir: str, normalize_path: bool = False) -> str:
311    """ Make any test replacement inside the given string. """
312
313    for _ in range(REPLACE_LIMIT):
314        match = re.search(rf'{key}\(([^)]*)\)', text)
315        if (match is None):
316            break
317
318        filename = match.group(1)
319
320        # Normalize any path separators.
321        filename = os.path.join(*filename.split('/'))
322
323        if (filename == ''):
324            path = target_dir
325        else:
326            path = os.path.join(target_dir, filename)
327
328        if (normalize_path):
329            path = os.path.abspath(path)
330
331        text = text.replace(match.group(0), path)
332
333    return text
334
335def compute_ancestor_basename(path: str, cli_tests_dir: str) -> str:
336    """
337    Get the test's name based off of its filename and location.
338    A useful function to use in get_test_basename().
339    """
340
341    path = os.path.abspath(path)
342
343    name = os.path.splitext(os.path.basename(path))[0]
344
345    # Clean drive identifiers (for Windows).
346    cli_tests_dir_path = os.path.splitdrive(os.path.abspath(cli_tests_dir))[1]
347    path = os.path.splitdrive(path)[1]
348
349    ancestors = os.path.dirname(path).replace(cli_tests_dir_path, '')
350    prefix = ancestors.replace(os.sep, '_')
351
352    if (prefix.startswith('_')):
353        prefix = prefix.replace('_', '', 1)
354
355    if (len(prefix) > 0):
356        name = f"{prefix}_{name}"
357
358    return name
359
360
361def _get_test_method(test_name: str, path: str, data_dir: str) -> typing.Callable:
362    """ Get a test method that represents the test case at the given path. """
363
364    def __method(self: edq.testing.unittest.BaseTest,
365            reraise_exception_types: typing.Union[typing.Tuple[typing.Type], None] = None,
366            **kwargs: typing.Any,
367            ) -> None:
368        test_info = CLITestInfo.load_path(path, test_name, getattr(self, BASE_TEMP_DIR_ATTR), data_dir)
369
370        # Allow the test class a chance to modify the test info before the test runs.
371        if (hasattr(self, 'modify_cli_test_info')):
372            self.modify_cli_test_info(test_info)
373
374        if (test_info.should_skip()):
375            self.skipTest(test_info.skip_message())
376
377        if (test_info.setup_func is not None):
378            test_info.setup_func(self, test_info)
379
380        old_args = sys.argv
381        sys.argv = [test_info.module.__file__] + test_info.arguments
382
383        previous_work_directory = os.getcwd()
384        os.chdir(test_info.work_dir)
385
386        try:
387            with contextlib.redirect_stdout(io.StringIO()) as stdout_output:
388                with contextlib.redirect_stderr(io.StringIO()) as stderr_output:
389                    test_info.module.main()
390
391            stdout_text = stdout_output.getvalue()
392            stderr_text = stderr_output.getvalue()
393
394            if (test_info.error):
395                self.fail(f"No error was not raised when one was expected ('{str(test_info.expected_stdout)}').")
396        except BaseException as ex:
397            if ((reraise_exception_types is not None) and isinstance(ex, reraise_exception_types)):
398                raise ex
399
400            if (not test_info.error):
401                raise ex
402
403            stdout_text = self.format_error_string(ex)
404
405            stderr_text = ''
406            if (isinstance(ex, SystemExit) and (ex.__context__ is not None)):
407                stderr_text = self.format_error_string(ex.__context__)
408        finally:
409            os.chdir(previous_work_directory)
410            sys.argv = old_args
411
412            if (test_info.teardown_func is not None):
413                test_info.teardown_func(self, test_info)
414
415        if (not test_info.split_stdout_stderr):
416            if ((len(stdout_text) > 0) and (len(stderr_text) > 0)):
417                stdout_text = f"{stdout_text}\n{OUTPUT_SEP}\n{stderr_text}"
418            elif (len(stderr_text) > 0):
419                stdout_text = stderr_text
420
421        if (test_info.stdout_assertion_func is not None):
422            test_info.stdout_assertion_func(self, test_info.expected_stdout, stdout_text)
423
424        if (test_info.stderr_assertion_func is not None):
425            test_info.stderr_assertion_func(self, test_info.expected_stderr, stderr_text)
426
427    return __method
428
429def add_test_paths(target_class: type, data_dir: str, paths: typing.List[str],
430        test_method_wrapper: typing.Union[TestMethodWrapperFunction, None] = None) -> None:
431    """ Add tests from the given test files. """
432
433    # Attach a temp directory to the testing class so all tests can share a common base temp dir.
434    if (not hasattr(target_class, BASE_TEMP_DIR_ATTR)):
435        setattr(target_class, BASE_TEMP_DIR_ATTR, edq.util.dirent.get_temp_path('edq_cli_test_'))
436
437    for path in sorted(paths):
438        basename = os.path.splitext(os.path.basename(path))[0]
439        if (hasattr(target_class, 'get_test_basename')):
440            basename = getattr(target_class, 'get_test_basename')(path)
441
442        test_name = 'test_cli__' + basename
443
444        try:
445            test_method = _get_test_method(test_name, path, data_dir)
446        except Exception as ex:
447            raise ValueError(f"Failed to parse test case '{path}'.") from ex
448
449        if (test_method_wrapper is not None):
450            test_method = test_method_wrapper(test_method, path)
451
452        setattr(target_class, test_name, test_method)
453
454def discover_test_cases(target_class: type, test_cases_dir: str, data_dir: str,
455        test_method_wrapper: typing.Union[TestMethodWrapperFunction, None] = None) -> None:
456    """ Look in the text cases directory for any test cases and add them as test methods to the test class. """
457
458    paths = list(sorted(glob.glob(os.path.join(test_cases_dir, "**", "*.txt"), recursive = True)))
459    add_test_paths(target_class, data_dir, paths, test_method_wrapper = test_method_wrapper)