lms.testing.serverrunner

  1import logging
  2import typing
  3
  4import edq.net.exchange
  5import edq.net.request
  6import edq.testing.serverrunner
  7import edq.util.parse
  8import edq.util.reflection
  9
 10import lms.backend.instance
 11import lms.cli.parser
 12import lms.model.constants
 13import lms.util.net
 14
 15BACKEND_REQUEST_CLEANING_FUNCS: typing.Dict[typing.Union[str, None], typing.Callable] = {
 16    lms.model.constants.BACKEND_TYPE_BLACKBOARD: lms.util.net.clean_blackboard_response,
 17    lms.model.constants.BACKEND_TYPE_CANVAS: lms.util.net.clean_canvas_response,
 18    lms.model.constants.BACKEND_TYPE_MOODLE: lms.util.net.clean_moodle_response,
 19}
 20
 21BACKEND_EXCHANGE_FINALIZING_FUNCS: typing.Dict[typing.Union[str, None], typing.Callable] = {
 22    lms.model.constants.BACKEND_TYPE_MOODLE: lms.util.net.finalize_moodle_exchange,
 23}
 24
 25class LMSServerRunner(edq.testing.serverrunner.ServerRunner):
 26    """ A server runner specifically for LMS servers. """
 27
 28    def __init__(self,
 29            backend_type: typing.Union[str, None] = None,
 30            **kwargs: typing.Any) -> None:
 31        super().__init__(**kwargs)
 32
 33        self.backend_type: typing.Union[str, None] = backend_type
 34        """
 35        The type of server being run.
 36        This value will be resolved after the server is started
 37        (since part of resolution may involve pinging the server.
 38        """
 39
 40        self._old_exchanges_clean_func: typing.Union[str, None] = None
 41        """
 42        The value of edq.net.exchange._exchanges_clean_func when start() is called.
 43        The original value may be changed in start(), and will be reset in stop().
 44        """
 45
 46        self._old_exchanges_finalize_func: typing.Union[str, None] = None
 47        """
 48        The value of edq.net.exchange._exchanges_finalize_func when start() is called.
 49        The original value may be changed in start(), and will be reset in stop().
 50        """
 51
 52        self._old_set_exchanges_clean_func: bool = False
 53        """
 54        The value of lms.cli.parser._set_exchanges_clean_func when start() is called.
 55        The original value may be changed in start(), and will be reset in stop().
 56        """
 57
 58        self._old_make_request_exchange_complete_func: typing.Union[edq.net.exchange.HTTPExchangeComplete, None] = None
 59        """
 60        The value of edq.net.request._make_request_exchange_complete_func when start() is called.
 61        The original value may be changed in start(), and will be reset in stop().
 62        """
 63
 64        self._old_serverrunner_logging_level: typing.Union[int, None] = None
 65        """
 66        The logging level for the edq server runner befoe tests are run.
 67        The original value may be changed in start(), and will be reset in stop().
 68        """
 69
 70    def start(self) -> None:
 71        # Set configs.
 72
 73        exchange_clean_func = BACKEND_REQUEST_CLEANING_FUNCS.get(self.backend_type, lms.util.net.clean_lms_response)
 74        exchange_clean_func_name = edq.util.reflection.get_qualified_name(exchange_clean_func)
 75        self._old_exchanges_clean_func = edq.net.exchange._exchanges_clean_func
 76        edq.net.exchange._exchanges_clean_func = exchange_clean_func_name
 77
 78        self._old_exchanges_finalize_func = edq.net.exchange._exchanges_finalize_func
 79        exchange_finalize_func = BACKEND_EXCHANGE_FINALIZING_FUNCS.get(self.backend_type, None)
 80        if (exchange_finalize_func is not None):
 81            exchange_finalize_func_name = edq.util.reflection.get_qualified_name(exchange_finalize_func)
 82            edq.net.exchange._exchanges_finalize_func = exchange_finalize_func_name
 83        else:
 84            edq.net.exchange._exchanges_finalize_func = None
 85
 86        self._old_set_exchanges_clean_func = lms.cli.parser._set_exchanges_clean_func
 87        lms.cli.parser._set_exchanges_clean_func = False
 88
 89        def _make_request_callback(exchange: edq.net.exchange.HTTPExchange) -> None:
 90            # Restart if the request is a write.
 91            if (edq.util.parse.boolean(exchange.headers.get(lms.model.constants.HEADER_KEY_WRITE, False))):
 92                self.restart()
 93
 94        self._old_make_request_exchange_complete_func = edq.net.request._make_request_exchange_complete_func
 95        edq.net.request._make_request_exchange_complete_func = typing.cast(edq.net.exchange.HTTPExchangeComplete, _make_request_callback)
 96
 97        # Disable logging from the runner, since it may disrupt CLI tests.
 98        logger = logging.getLogger('edq.testing.serverrunner')
 99        self._old_serverrunner_logging_level = logger.level
100        logger.setLevel(logging.WARNING)
101
102        # Start the server.
103        super().start()
104
105    def stop(self) -> bool:
106        if (self._old_serverrunner_logging_level is not None):
107            logger = logging.getLogger('edq.testing.serverrunner')
108            logger.setLevel(self._old_serverrunner_logging_level)
109            self._old_serverrunner_logging_level = None
110
111        if (not super().stop()):
112            return False
113
114        # Restore old configs.
115
116        edq.net.exchange._exchanges_clean_func = self._old_exchanges_clean_func
117        self._old_exchanges_clean_func = None
118
119        lms.cli.parser._set_exchanges_clean_func = self._old_set_exchanges_clean_func
120        self._old_set_exchanges_clean_func = False
121
122        edq.net.request._make_request_exchange_complete_func = self._old_make_request_exchange_complete_func
123        self._old_make_request_exchange_complete_func = None
124
125        return True
126
127    def identify_server(self) ->  bool:
128        backend_type = lms.backend.instance.guess_backend_type_from_request(self.server, timeout_secs = self.identify_wait_secs)
129        return (backend_type is not None)