lms.backend.testing

  1import glob
  2import os
  3import typing
  4
  5import edq.core.log
  6import edq.net.exchangeserver
  7import edq.testing.cli
  8import edq.testing.unittest
  9import edq.testing.httpserver
 10import edq.util.pyimport
 11
 12import lms.model.backend
 13import lms.model.base
 14import lms.backend.instance
 15
 16THIS_DIR: str = os.path.abspath(os.path.dirname(os.path.realpath(__file__)))
 17TESTDATA_DIR: str = os.path.join(THIS_DIR, 'testdata')
 18
 19BACKEND_TESTS_DIR: str = os.path.join(TESTDATA_DIR, 'backendtests')
 20
 21CLI_TESTDATA_DIR: str = os.path.join(TESTDATA_DIR, 'cli')
 22CLI_TESTS_DIR: str = os.path.join(CLI_TESTDATA_DIR, 'tests')
 23CLI_DATA_DIR: str = os.path.join(CLI_TESTDATA_DIR, 'data')
 24CLI_GLOBAL_CONFG_PATH: str = os.path.join(CLI_DATA_DIR, 'testing-edq-lms.json')
 25
 26TEST_FUNC_NAME_PREFIX: str = 'test_'
 27TEST_FILENAME_GLOB_PATTERN: str = '*_backendtest.py'
 28
 29class BackendTest(edq.testing.httpserver.HTTPServerTest):
 30    """
 31    A special test suite that is common across all LMS backends.
 32
 33    This is an HTTP test that will start a test server with exchanges specific to the target backend.
 34
 35    A common directory (BACKEND_TESTS_DIR) will be searched for any file that starts with TEST_FILENAME_GLOB_PATTERN.
 36    Then, that file will be checked for any function that starts with TEST_FUNC_NAME_PREFIX and matches BackendTestFunction.
 37    """
 38
 39    backend_type: typing.Union[str, None] = None
 40    """
 41    The backend type for this test.
 42    Must be set by the child class.
 43    """
 44
 45    exchanges_dir: typing.Union[str, None] = None
 46    """
 47    The directory to load HTTP exchanges from.
 48    Must be set by the child class.
 49    """
 50
 51    params_to_skip: typing.List[str] = []
 52    """ Parameters to skip while looking up exchanges. """
 53
 54    headers_to_skip: typing.List[str] = []
 55    """ Headers to skip while looking up exchanges. """
 56
 57    backend: typing.Union[lms.model.backend.APIBackend, None] = None
 58    """
 59    The backend for this test.
 60    Will be created during setup_server().
 61    """
 62
 63    backend_args: typing.Dict[str, typing.Any] = {
 64        'testing': True,
 65    }
 66    """ Any additional arguments to send to get_backend(). """
 67
 68    skip_base_request_test: bool = False
 69    """ Skip any base request tests. """
 70
 71    allowed_backend: typing.Union[str, None] = None
 72    """ If set, skip any backend tests that do not match this filter. """
 73
 74    def __init__(self, *args: typing.Any, **kwargs: typing.Any) -> None:
 75        super().__init__(*args, **kwargs)
 76
 77        self._user_email: typing.Union[str, None] = None
 78        """
 79        The email of the current user for this backend.
 80        Setting the user allows child classes to fetch specific information (like authentication information).
 81        """
 82
 83    @classmethod
 84    def setup_server(cls, server: edq.net.exchangeserver.HTTPExchangeServer) -> None:
 85        if (cls.server_key == ''):
 86            raise ValueError("BackendTest subclass did not set server key properly.")
 87
 88        if (cls.backend_type is None):
 89            raise ValueError("BackendTest subclass did not set backend type properly.")
 90
 91        if (cls.exchanges_dir is None):
 92            raise ValueError("BackendTest subclass did not set exchanges dir properly.")
 93
 94        edq.testing.httpserver.HTTPServerTest.setup_server(server)
 95        server.load_exchanges_dir(cls.exchanges_dir)
 96
 97        # Update match options.
 98        for (key, values) in [('params_to_skip', cls.params_to_skip), ('headers_to_skip', cls.headers_to_skip)]:
 99            if (key not in server.match_options):
100                server.match_options[key] = []
101
102            server.match_options[key] += values
103
104    @classmethod
105    def post_start_server(cls, server: edq.net.exchangeserver.HTTPExchangeServer) -> None:
106        cls.backend = lms.backend.instance.get_backend(cls.get_server_url(), backend_type = cls.backend_type, **cls.backend_args)
107
108    @classmethod
109    def get_base_args(cls) -> typing.Dict[str, typing.Any]:
110        """ Get a copy of the base arguments for a request (function). """
111
112        return {}
113
114    def setUp(self) -> None:
115        edq.core.log.init('ERROR')
116
117        self.clear_user()
118
119    def set_user(self, email: str) -> None:
120        """
121        Set the current user for this test.
122        This can be especially useful for child classes that need to set information based on the user
123        (like authentication headers).
124        """
125
126        self._user_email = email
127
128    def clear_user(self) -> None:
129        """
130        Clear the current user for this test.
131        This is automatically called before each test method.
132        """
133
134        self._user_email = None
135
136    def base_request_test(self,
137            request_function: typing.Callable,
138            test_cases: typing.List[typing.Tuple[typing.Dict[str, typing.Any], typing.Any, typing.Union[str, None]]],
139            stop_on_notimplemented: bool = True,
140            actual_clean_func: typing.Union[typing.Callable, None] = None,
141            expected_clean_func: typing.Union[typing.Callable, None] = None,
142            assertion_func: typing.Union[typing.Callable, None] = None,
143            ) -> None:
144        """
145        A common test for the base request functionality.
146        Test cases are passed in as: `[(kwargs (and overrides), expected, error substring), ...]`.
147        """
148
149        if ((self.allowed_backend is not None) and (self.allowed_backend != self.backend_type)):
150            self.skipTest(f"Backend {self.backend_type} has been filtered.")
151
152        skip_reason = None
153
154        for (i, test_case) in enumerate(test_cases):
155            (extra_kwargs, expected, error_substring) = test_case
156
157            with self.subTest(msg = f"Case {i}:"):
158                kwargs = self.get_base_args()
159                kwargs.update(extra_kwargs)
160
161                try:
162                    actual = request_function(**kwargs)
163                except NotImplementedError as ex:
164                    # We must handle this directly since we are in a subtest.
165                    if (stop_on_notimplemented):
166                        skip_reason = str(ex)
167                        break
168
169                    self.skipTest(f"Backend component not implemented: {str(ex)}.")
170                except Exception as ex:
171                    error_string = self.format_error_string(ex)
172                    if (error_substring is None):
173                        self.fail(f"Unexpected error: '{error_string}'.")
174
175                    self.assertIn(error_substring, error_string, 'Error is not as expected.')
176
177                    continue
178
179                if (error_substring is not None):
180                    self.fail(f"Did not get expected error: '{error_substring}'.")
181
182                if (actual_clean_func is not None):
183                    actual = actual_clean_func(actual)
184
185                if (expected_clean_func is not None):
186                    expected = expected_clean_func(expected)
187
188                # If we expect a tuple, compare the tuple contents instead of the tuple itself.
189                if (isinstance(expected, tuple)):
190                    if (not isinstance(actual, tuple)):
191                        raise ValueError(f"Expected results to be a tuple, found '{type(actual)}'.")
192
193                    if (len(expected) != len(actual)):
194                        raise ValueError(f"Result size mismatch. Expected: {len(expected)}, Actual: {len(actual)}.")
195                else:
196                    # Wrap the results in a tuple.
197                    expected = (expected, )
198                    actual = (actual, )
199
200                for i in range(len(expected)):  # pylint: disable=consider-using-enumerate
201                    expected_value = expected[i]
202                    actual_value = actual[i]
203
204                    if (assertion_func is not None):
205                        assertion_func(expected_value, actual_value)
206                    elif (isinstance(expected_value, lms.model.base.BaseType)):
207                        self.assertJSONEqual(expected_value, actual_value)
208                    elif (isinstance(expected_value, dict)):
209                        self.assertJSONDictEqual(expected_value, actual_value)
210                    elif (isinstance(expected_value, list)):
211                        self.assertJSONListEqual(expected_value, actual_value)
212                    else:
213                        self.assertEqual(expected_value, actual_value)
214
215        if (skip_reason is not None):
216            self.skipTest(f"Backend component not implemented: {skip_reason}.")
217
218    def modify_cli_test_info(self, test_info: edq.testing.cli.CLITestInfo) -> None:
219        """ Adjust the CLI test info to include core info (like server information). """
220
221        test_info.arguments += [
222            '--config-global', CLI_GLOBAL_CONFG_PATH,
223            '--server', self.get_server_url(),
224            '--server-type', str(self.backend_type),
225            '--config', 'testing=true',
226        ]
227
228        # Mark this CLI test for skipping based on the backend filter.
229        if ((self.allowed_backend is not None) and (self.allowed_backend != self.backend_type)):
230            test_info.skip_reasons.append(f"Backend {self.backend_type} has been filtered.")
231
232    @classmethod
233    def get_test_basename(cls, path: str) -> str:
234        """ Get the test's name based off of its filename and location. """
235
236        path = os.path.abspath(path)
237
238        name = os.path.splitext(os.path.basename(path))[0]
239
240        ancestors = os.path.dirname(path).replace(CLI_TESTS_DIR, '')
241        prefix = ancestors.replace(os.sep, '_')
242
243        if (prefix.startswith('_')):
244            prefix = prefix.replace('_', '', 1)
245
246        if (len(prefix) > 0):
247            name =  f"{prefix}_{name}"
248
249        return name
250
251@typing.runtime_checkable
252class BackendTestFunction(typing.Protocol):
253    """
254    A test function for backend tests.
255    A copy of this function will be attached to a test class created for each backend.
256    Therefore, `self` will be an instance of BackendTest.
257    """
258
259    def __call__(self, test: BackendTest) -> None:
260        """
261        A unit test for a BackendTest.
262        """
263
264def _wrap_test_function(test_function: BackendTestFunction) -> typing.Callable:
265    """ Wrap the backend test function in some common code for backend tests. """
266
267    def __method(self: BackendTest) -> None:
268        try:
269            test_function(self)
270        except NotImplementedError as ex:
271            # Skip tests for backend component that do not have implementations.
272            self.skipTest(f"Backend component not implemented: {str(ex)}.")
273
274    return __method
275
276def add_test_path(target_class: type, path: str) -> None:
277    """ Add tests from the given test files. """
278
279    test_module = edq.util.pyimport.import_path(path)
280
281    for attr_name in sorted(dir(test_module)):
282        if (not attr_name.startswith(TEST_FUNC_NAME_PREFIX)):
283            continue
284
285        test_function = getattr(test_module, attr_name)
286        setattr(target_class, attr_name, _wrap_test_function(test_function))
287
288def discover_test_cases(target_class: type) -> None:
289    """ Look in the text cases directory for any test cases and add them as test methods to the test class. """
290
291    paths = list(sorted(glob.glob(os.path.join(BACKEND_TESTS_DIR, "**", TEST_FILENAME_GLOB_PATTERN), recursive = True)))
292    for path in sorted(paths):
293        add_test_path(target_class, path)
294
295def attach_test_cases(target_class: type) -> None:
296    """ Attach all the standard test cases to the given class. """
297
298    # Attach backend tests.
299    discover_test_cases(target_class)
300
301    # Attach CLI tests.
302    edq.testing.cli.discover_test_cases(target_class, CLI_TESTS_DIR, CLI_DATA_DIR, test_method_wrapper = _wrap_cli_test_method)
303
304def _wrap_cli_test_method(test_method: typing.Callable, test_info_path: str) -> typing.Callable:
305    """ Wrap the CLI tests to ignore NotImplemented errors. """
306
307    def __method(self: edq.testing.unittest.BaseTest) -> None:
308        try:
309            test_method(self, reraise_exception_types = (NotImplementedError,))
310        except NotImplementedError as ex:
311            # Skip tests for backend component that do not have implementations.
312            self.skipTest(f"Backend component not implemented: {str(ex)}.")
313
314    return __method