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
 11import edq.util.serial
 12import quizcomp.parser.math
 13
 14import lms.model.backend
 15import lms.model.base
 16import lms.model.constants
 17import lms.backend.instance
 18import lms.testing.serverrunner
 19
 20THIS_DIR: str = os.path.abspath(os.path.dirname(os.path.realpath(__file__)))
 21TESTDATA_DIR: str = os.path.join(THIS_DIR, 'testdata')
 22
 23BACKEND_TESTS_DIR: str = os.path.join(TESTDATA_DIR, 'backendtests')
 24
 25CLI_TESTDATA_DIR: str = os.path.join(TESTDATA_DIR, 'cli')
 26CLI_TESTS_DIR: str = os.path.join(CLI_TESTDATA_DIR, 'tests')
 27CLI_DATA_DIR: str = os.path.join(CLI_TESTDATA_DIR, 'data')
 28CLI_GLOBAL_CONFG_PATH: str = os.path.join(CLI_DATA_DIR, 'testing-edq-lms.json')
 29
 30TEST_FUNC_NAME_PREFIX: str = 'test_'
 31TEST_FILENAME_GLOB_PATTERN: str = '*_backendtest.py'
 32
 33class BackendTest(edq.testing.httpserver.HTTPServerTest):
 34    """
 35    A special test suite that is common across all LMS backends.
 36
 37    This is an HTTP test that will start a test server with exchanges specific to the target backend.
 38
 39    A common directory (BACKEND_TESTS_DIR) will be searched for any file that starts with TEST_FILENAME_GLOB_PATTERN.
 40    Then, that file will be checked for any function that starts with TEST_FUNC_NAME_PREFIX and matches BackendTestFunction.
 41    """
 42
 43    backend_type: typing.Union[lms.model.constants.BackendType, None] = None
 44    """
 45    The backend type for this test.
 46    Must be set by the child class.
 47    """
 48
 49    server_runner: typing.Union[lms.testing.serverrunner.LMSServerRunner, None] = None
 50    """ If a current server runner for this test (if there is one). """
 51
 52    exchanges_dir: typing.Union[str, None] = None
 53    """
 54    The directory to load HTTP exchanges from.
 55    Must be set by the child class.
 56    """
 57
 58    params_to_skip: typing.List[str] = []
 59    """ Parameters to skip while looking up exchanges. """
 60
 61    headers_to_skip: typing.List[str] = []
 62    """ Headers to skip while looking up exchanges. """
 63
 64    backend: typing.Union[lms.model.backend.APIBackend, None] = None
 65    """
 66    The backend for this test.
 67    Will be created during setup_server().
 68    """
 69
 70    backend_args: typing.Dict[str, typing.Any] = {
 71        'testing': True,
 72    }
 73    """ Any additional arguments to send to get_backend(). """
 74
 75    skip_base_request_test: bool = False
 76    """ Skip any base request tests. """
 77
 78    allowed_backend: typing.Union[lms.model.constants.BackendType, None] = None
 79    """ If set, skip any backend tests that do not match this filter. """
 80
 81    def __init__(self, *args: typing.Any, **kwargs: typing.Any) -> None:
 82        super().__init__(*args, **kwargs)
 83
 84        self._user_email: typing.Union[str, None] = None
 85        """
 86        The email of the current user for this backend.
 87        Setting the user allows child classes to fetch specific information (like authentication information).
 88        """
 89
 90        # Most backends have to modify exchanges in some way that makes these tests unreliable.
 91        # Instead, exchanges are tested enough through normal testing usage.
 92        self.skip_test_exchanges_base = True
 93
 94    @classmethod
 95    def setUpClass(cls) -> None:
 96        super().setUpClass()
 97
 98        # Disable KaTeX for testing.
 99        quizcomp.parser.math._katex_available = False
100
101    @classmethod
102    def tearDownClass(cls) -> None:
103        super().tearDownClass()
104
105        quizcomp.parser.math._katex_available = None
106
107    @classmethod
108    def setup_server(cls, server: edq.net.exchangeserver.HTTPExchangeServer) -> None:
109        if (cls.server_key == ''):
110            raise ValueError("BackendTest subclass did not set server key properly.")
111
112        edq.testing.httpserver.HTTPServerTest.setup_server(server)
113
114    @classmethod
115    def create_server(cls) -> edq.net.exchangeserver.HTTPExchangeServer:
116        return LMSHTTPExchangeServer()
117
118    @classmethod
119    def post_start_server(cls, server: edq.net.exchangeserver.HTTPExchangeServer) -> None:
120        if (cls.backend_type is None):
121            raise ValueError("BackendTest subclass did not set backend type properly.")
122
123        if (cls.exchanges_dir is None):
124            raise ValueError("BackendTest subclass did not set exchanges dir properly.")
125
126        context = edq.util.serial.SerializationContext(json_options = {
127            'strict': True,
128        })
129        server.load_exchanges_dir(cls.exchanges_dir, context = context, finalize_func = cls._finalize_exchange)
130
131        # Update match options.
132        for (key, values) in [('params_to_skip', cls.params_to_skip), ('headers_to_skip', cls.headers_to_skip)]:
133            if (key not in server.match_options):
134                server.match_options[key] = []
135
136            server.match_options[key] += values
137
138        config_data: typing.Dict[str, typing.Any] = {
139            'server': cls.get_server_url(),
140            'backend_type': cls.backend_type,
141        }
142        config_data.update(cls.backend_args)
143
144        config = lms.model.config.Config.from_dict(config_data)
145        cls.backend = lms.backend.instance.get_backend(config, **cls.backend_args)
146
147        if (cls.server_runner is not None):
148            cls.server_runner.backend = cls.backend
149
150    @classmethod
151    def get_base_args(cls) -> typing.Dict[str, typing.Any]:
152        """ Get a copy of the base arguments for a request (function). """
153
154        return {}
155
156    def setUp(self) -> None:
157        edq.core.log.init('ERROR')
158
159        self.clear_user()
160
161    def get_backend(self) -> lms.model.backend.APIBackend:
162        """ Get the backend or fail if there is no backend. """
163
164        if (self.backend is None):
165            self.fail("No backend is set.")
166
167        return self.backend
168
169    def set_user(self, email: str) -> None:
170        """
171        Set the current user for this test.
172        This can be especially useful for child classes that need to set information based on the user
173        (like authentication headers).
174        """
175
176        self._user_email = email
177
178    def clear_user(self) -> None:
179        """
180        Clear the current user for this test.
181        This is automatically called before each test method.
182        """
183
184        self._user_email = None
185
186    def base_request_test(self,
187            request_function: typing.Callable,
188            test_cases: typing.List[typing.Tuple[typing.Dict[str, typing.Any], typing.Any, typing.Union[str, None]]],
189            stop_on_notimplemented: bool = True,
190            actual_clean_func: typing.Union[typing.Callable, None] = None,
191            expected_clean_func: typing.Union[typing.Callable, None] = None,
192            assertion_func: typing.Union[typing.Callable, None] = None,
193            disable_server_restart: bool = False,
194            ) -> None:
195        """
196        A common test for the base request functionality.
197        Test cases are passed in as: `[(kwargs (and overrides), expected, error substring), ...]`.
198        """
199
200        if ((self.backend_type is not None) and (self.allowed_backend is not None) and (self.allowed_backend != self.backend_type)):
201            self.skipTest(f"Backend '{self.backend_type.value}' has been filtered, only allowing '{self.allowed_backend.value}'.")
202
203        skip_reason = None
204
205        for (i, test_case) in enumerate(test_cases):
206            (extra_kwargs, expected, error_substring) = test_case
207
208            with self.subTest(msg = f"Case {i}:"):
209                kwargs = self.get_base_args()
210                kwargs.update(extra_kwargs)
211
212                if (disable_server_restart and (self.server_runner is not None)):
213                    self.server_runner.skip_restart = True
214
215                try:
216                    actual = request_function(**kwargs)
217                except NotImplementedError as ex:
218                    # We must handle this directly since we are in a subtest.
219                    if (stop_on_notimplemented):
220                        skip_reason = str(ex)
221                        break
222
223                    self.skipTest(f"Backend component not implemented: {str(ex)}.")
224                except Exception as ex:
225                    error_string = self.format_error_string(ex)
226                    if (error_substring is None):
227                        self.fail(f"Unexpected error: '{error_string}'.")
228
229                    self.assertIn(error_substring, error_string, 'Error is not as expected.')
230                    continue
231                finally:
232                    if (disable_server_restart and (self.server_runner is not None)):
233                        self.server_runner.skip_restart = False
234                        self.server_runner.restart()
235
236                if (error_substring is not None):
237                    self.fail(f"Did not get expected error: '{error_substring}'.")
238
239                if (actual_clean_func is not None):
240                    actual = actual_clean_func(actual)
241
242                if (expected_clean_func is not None):
243                    expected = expected_clean_func(expected)
244
245                # If we expect a tuple, compare the tuple contents instead of the tuple itself.
246                if (isinstance(expected, tuple)):
247                    if (not isinstance(actual, tuple)):
248                        raise ValueError(f"Expected results to be a tuple, found '{type(actual)}'.")
249
250                    if (len(expected) != len(actual)):
251                        raise ValueError(f"Result size mismatch. Expected: {len(expected)}, Actual: {len(actual)}.")
252                else:
253                    # Wrap the results in a tuple.
254                    expected = (expected, )
255                    actual = (actual, )
256
257                for i in range(len(expected)):  # pylint: disable=consider-using-enumerate
258                    expected_value = expected[i]
259                    actual_value = actual[i]
260
261                    if (assertion_func is not None):
262                        assertion_func(expected_value, actual_value)
263                    elif (isinstance(expected_value, lms.model.base.BaseType)):
264                        self.assertJSONEqual(expected_value, actual_value)
265                    elif (isinstance(expected_value, (dict, edq.util.serial.DictConverter))):
266                        self.assertJSONDictEqual(expected_value, actual_value)
267                    elif (isinstance(expected_value, list)):
268                        self.assertJSONListEqual(expected_value, actual_value)
269                    elif (isinstance(expected_value, edq.util.serial.PODConverter)):
270                        self.assertJSONEqual(expected_value, actual_value)
271                    else:
272                        self.assertEqual(expected_value, actual_value)
273
274        if (skip_reason is not None):
275            self.skipTest(f"Backend component not implemented: {skip_reason}.")
276
277    def modify_cli_test_info(self, test_info: edq.testing.cli.CLITestInfo) -> None:
278        """ Adjust the CLI test info to include core info (like server information). """
279
280        if ((self.backend_type is not None) and (self.backend_type.value in test_info.extra_options.get('skip_backends', []))):
281            test_info.skip_reasons.append(f"CLI test backend '{self.backend_type.value}' has been skipped by test info.")
282            return
283
284        test_info.arguments += [
285            '--config-global', CLI_GLOBAL_CONFG_PATH,
286            '--server', self.get_server_url(),
287            '--config', 'testing=true',
288        ]
289
290        if (self.backend_type is not None):
291            test_info.arguments += ['--server-type', self.backend_type.value]
292
293        # Mark this CLI test for skipping based on the backend filter.
294        if ((self.backend_type is not None) and (self.allowed_backend is not None) and (self.allowed_backend != self.backend_type)):
295            test_info.skip_reasons.append(
296                    f"CLI test backend '{self.backend_type.value}' has been filtered, only allowing '{self.allowed_backend.value}'.")
297
298    @classmethod
299    def get_test_basename(cls, path: str) -> str:
300        """ Get the test's name based off of its filename and location. """
301
302        return edq.testing.cli.compute_ancestor_basename(path, CLI_TESTS_DIR)
303
304    @classmethod
305    def _finalize_exchange(cls, exchange: edq.net.exchange.HTTPExchange) -> edq.net.exchange.HTTPExchange:
306        """
307        Finalize an exchange before loading it into the test server.
308        """
309
310        # Check for redirect locations with a slug.
311        if ('location' in exchange.response_headers):
312            location = exchange.response_headers['location'].replace(lms.model.constants.SERVER_SLUG, cls.get_server_url())
313            exchange.response_headers['location'] = location
314
315        return exchange
316
317class LMSHTTPExchangeServer(edq.net.exchangeserver.HTTPExchangeServer):
318    """ A custom exchange server for our tests. """
319
320    def missing_request(self, query: edq.net.exchange.HTTPExchange) -> typing.Union[edq.net.exchange.HTTPExchange, None]:
321        # Specal Canvas patch to handle a multi-stage file upload (which uses redirects).
322        if (query.url_path == 'files_api'):
323            query.parameters = {'filename': query.parameters['filename']}
324            query.files = []
325
326            exchange, _ = self.lookup_exchange(query)
327            return exchange
328
329        return None
330
331@typing.runtime_checkable
332class BackendTestFunction(typing.Protocol):
333    """
334    A test function for backend tests.
335    A copy of this function will be attached to a test class created for each backend.
336    Therefore, `self` will be an instance of BackendTest.
337    """
338
339    def __call__(self, test: BackendTest) -> None:
340        """
341        A unit test for a BackendTest.
342        """
343
344def _wrap_test_function(test_function: BackendTestFunction) -> typing.Callable:
345    """ Wrap the backend test function in some common code for backend tests. """
346
347    def __method(self: BackendTest) -> None:
348        try:
349            test_function(self)
350        except NotImplementedError as ex:
351            # Skip tests for backend component that do not have implementations.
352            self.skipTest(f"Backend component not implemented: {str(ex)}.")
353
354    return __method
355
356def add_test_path(target_class: type, path: str) -> None:
357    """ Add tests from the given test files. """
358
359    test_module = edq.util.pyimport.import_path(path)
360
361    for attr_name in sorted(dir(test_module)):
362        if (not attr_name.startswith(TEST_FUNC_NAME_PREFIX)):
363            continue
364
365        test_function = getattr(test_module, attr_name)
366        setattr(target_class, attr_name, _wrap_test_function(test_function))
367
368def discover_test_cases(target_class: type) -> None:
369    """ Look in the text cases directory for any test cases and add them as test methods to the test class. """
370
371    paths = list(sorted(glob.glob(os.path.join(BACKEND_TESTS_DIR, "**", TEST_FILENAME_GLOB_PATTERN), recursive = True)))
372    for path in sorted(paths):
373        add_test_path(target_class, path)
374
375def attach_test_cases(target_class: type) -> None:
376    """ Attach all the standard test cases to the given class. """
377
378    # Attach backend tests.
379    discover_test_cases(target_class)
380
381    # Attach CLI tests.
382    edq.testing.cli.discover_test_cases(target_class, CLI_TESTS_DIR, CLI_DATA_DIR, test_method_wrapper = _wrap_cli_test_method)
383
384def _wrap_cli_test_method(test_method: typing.Callable, test_info_path: str) -> typing.Callable:
385    """ Wrap the CLI tests to ignore NotImplemented errors. """
386
387    def __method(self: edq.testing.unittest.BaseTest) -> None:
388        try:
389            test_method(self, reraise_exception_types = (NotImplementedError,))
390        except NotImplementedError as ex:
391            # Skip tests for backend component that do not have implementations.
392            self.skipTest(f"Backend component not implemented: {str(ex)}.")
393
394    return __method