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