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
 54class CLITestInfo:
 55    """ The required information to run a CLI test. """
 56
 57    def __init__(self,
 58            test_name: str,
 59            base_dir: str,
 60            data_dir: str,
 61            temp_dir: str,
 62            cli: typing.Union[str, None] = None,
 63            arguments: typing.Union[typing.List[str], None] = None,
 64            error: bool = False,
 65            platform_skip: typing.Union[str, None] = None,
 66            stdout_assertion_func: typing.Union[str, None] = DEFAULT_ASSERTION_FUNC_NAME,
 67            stderr_assertion_func: typing.Union[str, None] = None,
 68            expected_stdout: str = '',
 69            expected_stderr: str = '',
 70            split_stdout_stderr: bool = False,
 71            strip_error_output: bool = True,
 72            extra_options: typing.Union[typing.Dict[str, typing.Any], None] = None,
 73            **kwargs: typing.Any) -> None:
 74        self.skip_reasons: typing.List[str] = []
 75        """
 76        Reasons that this test will be skipped.
 77        Any entries in this list indicate that the test should be skipped.
 78        """
 79
 80        self.platform_skip_pattern: typing.Union[str, None] = platform_skip
 81        """
 82        A pattern to check if the test should be skipped on the current platform.
 83        Will be used in `re.search()` against `sys.platform`.
 84        """
 85
 86        if ((platform_skip is not None) and re.search(platform_skip, sys.platform)):
 87            self.skip_reasons.append(f"not available on platform: '{sys.platform}'")
 88
 89        self.test_name: str = test_name
 90        """ The name of this test. """
 91
 92        self.base_dir: str = base_dir
 93        """
 94        The base directory for this test (usually the dir the CLI test file lives.
 95        This is the expansion for `__BASE_DIR__` paths.
 96        """
 97
 98        self.data_dir: str = data_dir
 99        """
100        A directory that additional testing data lives in.
101        This is the expansion for `__DATA_DIR__` paths.
102        """
103
104        self.temp_dir: str = temp_dir
105        """
106        A temp directory that this test has access to.
107        This is the expansion for `__TEMP_DIR__` paths.
108        """
109
110        edq.util.dirent.mkdir(temp_dir)
111
112        if (cli is None):
113            raise ValueError("Missing CLI module.")
114
115        self.module_name: str = cli
116        """ The name of the module to invoke. """
117
118        self.module: typing.Any = None
119        """ The module to invoke. """
120
121        if (not self.should_skip()):
122            self.module = edq.util.pyimport.import_name(self.module_name)
123
124        if (arguments is None):
125            arguments = []
126
127        self.arguments: typing.List[str] = arguments
128        """ The CLI arguments. """
129
130        self.error: bool = error
131        """ Whether or not this test is expected to be an error (raise an exception). """
132
133        self.stdout_assertion_func: typing.Union[edq.testing.asserts.StringComparisonAssertion, None] = None
134        """ The assertion func to compare the expected and actual stdout of the CLI. """
135
136        if ((stdout_assertion_func is not None) and (not self.should_skip())):
137            self.stdout_assertion_func = edq.util.pyimport.fetch(stdout_assertion_func)
138
139        self.stderr_assertion_func: typing.Union[edq.testing.asserts.StringComparisonAssertion, None] = None
140        """ The assertion func to compare the expected and actual stderr of the CLI. """
141
142        if ((stderr_assertion_func is not None) and (not self.should_skip())):
143            self.stderr_assertion_func = edq.util.pyimport.fetch(stderr_assertion_func)
144
145        self.expected_stdout: str = expected_stdout
146        """ The expected stdout. """
147
148        self.expected_stderr: str = expected_stderr
149        """ The expected stderr. """
150
151        if (error and strip_error_output):
152            self.expected_stdout = self.expected_stdout.strip()
153            self.expected_stderr = self.expected_stderr.strip()
154
155        self.split_stdout_stderr: bool = split_stdout_stderr
156        """
157        Split stdout and stderr into different strings for testing.
158        By default, these two will be combined.
159        If both are non-empty, then they will be joined like: f"{stdout}\n{OUTPUT_SEP}\n{stderr}".
160        Otherwise, only the non-empty one will be present with no separator.
161        Any stdout assertions will be applied to the combined text.
162        """
163
164        # Make any path normalizations over the arguments and expected output.
165        self.expected_stdout = self._process_text(self.expected_stdout)
166        self.expected_stderr = self._process_text(self.expected_stderr)
167        for (i, argument) in enumerate(self.arguments):
168            self.arguments[i] = self._process_text(argument)
169
170        if (extra_options is None):
171            extra_options = {}
172
173        self.extra_options: typing.Union[typing.Dict[str, typing.Any], None] = extra_options
174        """
175        A place to store additional options.
176        Extra top-level options will cause tests to error.
177        """
178
179        if (len(kwargs) > 0):
180            raise ValueError(f"Found unknown CLI test options: '{kwargs}'.")
181
182    def _process_text(self, text: str) -> str:
183        """
184        Process text with and desired replacements.
185
186        This will expand path replacements in testing text.
187        This allows for consistent paths (even absolute paths) in the test text.
188        """
189
190        text_replacements = [
191            (OS_PATH_SEP, os.sep),
192        ]
193
194        for (target, replacement) in text_replacements:
195            text = text.replace(target, replacement)
196
197        path_replacements = [
198            (DATA_DIR_ID, self.data_dir, False),
199            (TEMP_DIR_ID, self.temp_dir, False),
200            (BASE_DIR_ID, self.base_dir, False),
201            (ABS_DATA_DIR_ID, self.data_dir, True),
202        ]
203
204        for (key, target_dir, normalize) in path_replacements:
205            text = replace_path_pattern(text, key, target_dir, normalize_path = normalize)
206
207        return text
208
209    def should_skip(self) -> bool:
210        """ Check if this test should be skipped. """
211
212        return (len(self.skip_reasons) > 0)
213
214    def skip_message(self) -> str:
215        """ Get a message displaying the reasons this test should be skipped. """
216
217        return f"This test has been skipped because of the following: {self.skip_reasons}."
218
219    @staticmethod
220    def load_path(path: str, test_name: str, base_temp_dir: str, data_dir: str) -> 'CLITestInfo':
221        """ Load a CLI test file and extract the test info. """
222
223        options, expected_stdout = read_test_file(path)
224
225        options['expected_stdout'] = expected_stdout
226
227        base_dir = os.path.dirname(os.path.abspath(path))
228        temp_dir = os.path.join(base_temp_dir, test_name)
229
230        return CLITestInfo(test_name, base_dir, data_dir, temp_dir, **options)
231
232@typing.runtime_checkable
233class TestMethodWrapperFunction(typing.Protocol):
234    """
235    A function that can be used to wrap/modify a CLI test method before it is attached to the test class.
236    """
237
238    def __call__(self,
239            test_method: typing.Callable,
240            test_info_path: str,
241            ) -> typing.Callable:
242        """
243        Wrap and/or modify the CLI test method before it is attached to the test class.
244        See _get_test_method() for the input method.
245        The returned method will be used in-place of the input one.
246        """
247
248def read_test_file(path: str) -> typing.Tuple[typing.Dict[str, typing.Any], str]:
249    """ Read a test case file and split the output into JSON data and text. """
250
251    json_lines: typing.List[str] = []
252    output_lines: typing.List[str] = []
253
254    text = edq.util.dirent.read_file(path, strip = False)
255
256    accumulator = json_lines
257    switched_accumulator = False
258
259    for line in text.split("\n"):
260        if ((not switched_accumulator) and (line.strip() == TEST_CASE_SEP)):
261            accumulator = output_lines
262            switched_accumulator = True
263            continue
264
265        accumulator.append(line)
266
267    options = edq.util.json.loads(''.join(json_lines))
268    output = "\n".join(output_lines)
269
270    return options, output
271
272def replace_path_pattern(text: str, key: str, target_dir: str, normalize_path: bool = False) -> str:
273    """ Make any test replacement inside the given string. """
274
275    for _ in range(REPLACE_LIMIT):
276        match = re.search(rf'{key}\(([^)]*)\)', text)
277        if (match is None):
278            break
279
280        filename = match.group(1)
281
282        # Normalize any path separators.
283        filename = os.path.join(*filename.split('/'))
284
285        if (filename == ''):
286            path = target_dir
287        else:
288            path = os.path.join(target_dir, filename)
289
290        if (normalize_path):
291            path = os.path.abspath(path)
292
293        text = text.replace(match.group(0), path)
294
295    return text
296
297def compute_ancestor_basename(path: str, cli_tests_dir: str) -> str:
298    """
299    Get the test's name based off of its filename and location.
300    A useful fuction to use in get_test_basename().
301    """
302
303    path = os.path.abspath(path)
304
305    name = os.path.splitext(os.path.basename(path))[0]
306
307    # Clean drive identifiers (for Windows).
308    cli_tests_dir_path = os.path.splitdrive(os.path.abspath(cli_tests_dir))[1]
309    path = os.path.splitdrive(path)[1]
310
311    ancestors = os.path.dirname(path).replace(cli_tests_dir_path, '')
312    prefix = ancestors.replace(os.sep, '_')
313
314    if (prefix.startswith('_')):
315        prefix = prefix.replace('_', '', 1)
316
317    if (len(prefix) > 0):
318        name = f"{prefix}_{name}"
319
320    return name
321
322
323def _get_test_method(test_name: str, path: str, data_dir: str) -> typing.Callable:
324    """ Get a test method that represents the test case at the given path. """
325
326    def __method(self: edq.testing.unittest.BaseTest,
327            reraise_exception_types: typing.Union[typing.Tuple[typing.Type], None] = None,
328            **kwargs: typing.Any) -> None:
329        test_info = CLITestInfo.load_path(path, test_name, getattr(self, BASE_TEMP_DIR_ATTR), data_dir)
330
331        # Allow the test class a chance to modify the test info before the test runs.
332        if (hasattr(self, 'modify_cli_test_info')):
333            self.modify_cli_test_info(test_info)
334
335        if (test_info.should_skip()):
336            self.skipTest(test_info.skip_message())
337
338        old_args = sys.argv
339        sys.argv = [test_info.module.__file__] + test_info.arguments
340
341        try:
342            with contextlib.redirect_stdout(io.StringIO()) as stdout_output:
343                with contextlib.redirect_stderr(io.StringIO()) as stderr_output:
344                    test_info.module.main()
345
346            stdout_text = stdout_output.getvalue()
347            stderr_text = stderr_output.getvalue()
348
349            if (test_info.error):
350                self.fail(f"No error was not raised when one was expected ('{str(test_info.expected_stdout)}').")
351        except BaseException as ex:
352            if ((reraise_exception_types is not None) and isinstance(ex, reraise_exception_types)):
353                raise ex
354
355            if (not test_info.error):
356                raise ex
357
358            stdout_text = self.format_error_string(ex)
359
360            stderr_text = ''
361            if (isinstance(ex, SystemExit) and (ex.__context__ is not None)):
362                stderr_text = self.format_error_string(ex.__context__)
363        finally:
364            sys.argv = old_args
365
366        if (not test_info.split_stdout_stderr):
367            if ((len(stdout_text) > 0) and (len(stderr_text) > 0)):
368                stdout_text = f"{stdout_text}\n{OUTPUT_SEP}\n{stderr_text}"
369            elif (len(stderr_text) > 0):
370                stdout_text = stderr_text
371
372        if (test_info.stdout_assertion_func is not None):
373            test_info.stdout_assertion_func(self, test_info.expected_stdout, stdout_text)
374
375        if (test_info.stderr_assertion_func is not None):
376            test_info.stderr_assertion_func(self, test_info.expected_stderr, stderr_text)
377
378    return __method
379
380def add_test_paths(target_class: type, data_dir: str, paths: typing.List[str],
381        test_method_wrapper: typing.Union[TestMethodWrapperFunction, None] = None) -> None:
382    """ Add tests from the given test files. """
383
384    # Attach a temp directory to the testing class so all tests can share a common base temp dir.
385    if (not hasattr(target_class, BASE_TEMP_DIR_ATTR)):
386        setattr(target_class, BASE_TEMP_DIR_ATTR, edq.util.dirent.get_temp_path('edq_cli_test_'))
387
388    for path in sorted(paths):
389        basename = os.path.splitext(os.path.basename(path))[0]
390        if (hasattr(target_class, 'get_test_basename')):
391            basename = getattr(target_class, 'get_test_basename')(path)
392
393        test_name = 'test_cli__' + basename
394
395        try:
396            test_method = _get_test_method(test_name, path, data_dir)
397        except Exception as ex:
398            raise ValueError(f"Failed to parse test case '{path}'.") from ex
399
400        if (test_method_wrapper is not None):
401            test_method = test_method_wrapper(test_method, path)
402
403        setattr(target_class, test_name, test_method)
404
405def discover_test_cases(target_class: type, test_cases_dir: str, data_dir: str,
406        test_method_wrapper: typing.Union[TestMethodWrapperFunction, None] = None) -> None:
407    """ Look in the text cases directory for any test cases and add them as test methods to the test class. """
408
409    paths = list(sorted(glob.glob(os.path.join(test_cases_dir, "**", "*.txt"), recursive = True)))
410    add_test_paths(target_class, data_dir, paths, test_method_wrapper = test_method_wrapper)