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        return edq.testing.cli.compute_ancestor_basename(path, CLI_TESTS_DIR)
237
238@typing.runtime_checkable
239class BackendTestFunction(typing.Protocol):
240    """
241    A test function for backend tests.
242    A copy of this function will be attached to a test class created for each backend.
243    Therefore, `self` will be an instance of BackendTest.
244    """
245
246    def __call__(self, test: BackendTest) -> None:
247        """
248        A unit test for a BackendTest.
249        """
250
251def _wrap_test_function(test_function: BackendTestFunction) -> typing.Callable:
252    """ Wrap the backend test function in some common code for backend tests. """
253
254    def __method(self: BackendTest) -> None:
255        try:
256            test_function(self)
257        except NotImplementedError as ex:
258            # Skip tests for backend component that do not have implementations.
259            self.skipTest(f"Backend component not implemented: {str(ex)}.")
260
261    return __method
262
263def add_test_path(target_class: type, path: str) -> None:
264    """ Add tests from the given test files. """
265
266    test_module = edq.util.pyimport.import_path(path)
267
268    for attr_name in sorted(dir(test_module)):
269        if (not attr_name.startswith(TEST_FUNC_NAME_PREFIX)):
270            continue
271
272        test_function = getattr(test_module, attr_name)
273        setattr(target_class, attr_name, _wrap_test_function(test_function))
274
275def discover_test_cases(target_class: type) -> None:
276    """ Look in the text cases directory for any test cases and add them as test methods to the test class. """
277
278    paths = list(sorted(glob.glob(os.path.join(BACKEND_TESTS_DIR, "**", TEST_FILENAME_GLOB_PATTERN), recursive = True)))
279    for path in sorted(paths):
280        add_test_path(target_class, path)
281
282def attach_test_cases(target_class: type) -> None:
283    """ Attach all the standard test cases to the given class. """
284
285    # Attach backend tests.
286    discover_test_cases(target_class)
287
288    # Attach CLI tests.
289    edq.testing.cli.discover_test_cases(target_class, CLI_TESTS_DIR, CLI_DATA_DIR, test_method_wrapper = _wrap_cli_test_method)
290
291def _wrap_cli_test_method(test_method: typing.Callable, test_info_path: str) -> typing.Callable:
292    """ Wrap the CLI tests to ignore NotImplemented errors. """
293
294    def __method(self: edq.testing.unittest.BaseTest) -> None:
295        try:
296            test_method(self, reraise_exception_types = (NotImplementedError,))
297        except NotImplementedError as ex:
298            # Skip tests for backend component that do not have implementations.
299            self.skipTest(f"Backend component not implemented: {str(ex)}.")
300
301    return __method