lms.testing.serverrunner

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