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