lms.testing.serverrunner

  1import logging
  2import typing
  3
  4import edq.core.errors
  5import edq.net.exchange
  6import edq.net.request
  7import edq.net.settings
  8import edq.testing.serverrunner
  9import edq.util.parse
 10import edq.util.reflection
 11
 12import lms.backend.instance
 13import lms.cli.parser
 14import lms.model.backend
 15import lms.model.constants
 16import lms.util.net
 17
 18BACKEND_REQUEST_CLEANING_FUNCS: typing.Dict[typing.Union[lms.model.constants.BackendType, None], typing.Callable] = {
 19    lms.model.constants.BackendType.BLACKBOARD: lms.util.net.clean_blackboard_response,
 20    lms.model.constants.BackendType.CANVAS: lms.util.net.clean_canvas_response,
 21    lms.model.constants.BackendType.MOODLE: lms.util.net.clean_moodle_response,
 22}
 23
 24BACKEND_EXCHANGE_FINALIZING_FUNCS: typing.Dict[typing.Union[lms.model.constants.BackendType, None], typing.Callable] = {
 25    lms.model.constants.BackendType.CANVAS: lms.util.net.finalize_canvas_exchange,
 26    lms.model.constants.BackendType.MOODLE: lms.util.net.finalize_moodle_exchange,
 27}
 28
 29class LMSServerRunner(edq.testing.serverrunner.ServerRunner):
 30    """ A server runner specifically for LMS servers. """
 31
 32    def __init__(self,
 33            backend_type: typing.Union[lms.model.constants.BackendType, str, None] = None,
 34            **kwargs: typing.Any) -> None:
 35        super().__init__(**kwargs)
 36
 37        if (isinstance(backend_type, str)):
 38            backend_type = lms.model.constants.BackendType(backend_type)
 39
 40        self.backend_type: typing.Union[lms.model.constants.BackendType, None] = backend_type
 41        """
 42        The type of server being run.
 43        This value will be resolved after the server is started
 44        (since part of resolution may involve pinging the server.
 45        """
 46
 47        self.backend: typing.Union[lms.model.backend.APIBackend, None] = None
 48        """
 49        The backend currently being used while this server runner is active.
 50        This may not always be set.
 51        """
 52
 53        self._old_exchanges_clean_response_func: typing.Union[str, None] = None
 54        """
 55        The value of edq.net.settings.get_exchanges_clean_response_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_exchanges_finalize_func: typing.Union[str, None] = None
 60        """
 61        The value of edq.net.settings.get_exchanges_finalize_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_set_exchanges_clean_response_func: bool = False
 66        """
 67        The value of lms.cli.parser._set_exchanges_clean_response_func when start() is called.
 68        The original value may be changed in start(), and will be reset in stop().
 69        """
 70
 71        self._old_request_complete_callback: typing.Union[edq.net.exchange.HTTPExchangeComplete, None] = None
 72        """
 73        The value of edq.net.settings.get_request_complete_callback() when start() is called.
 74        The original value may be changed in start(), and will be reset in stop().
 75        """
 76
 77        self._old_serverrunner_logging_level: typing.Union[int, None] = None
 78        """
 79        The logging level for the edq server runner befoe tests are run.
 80        The original value may be changed in start(), and will be reset in stop().
 81        """
 82
 83    def start(self) -> None:
 84        # Set configs.
 85
 86        exchange_clean_func = BACKEND_REQUEST_CLEANING_FUNCS.get(self.backend_type, lms.util.net.clean_lms_response)
 87        exchange_clean_func_name = edq.util.reflection.get_qualified_name(exchange_clean_func)
 88        self._old_exchanges_clean_response_func = edq.net.settings.get_exchanges_clean_response_func()
 89        edq.net.settings.set_exchanges_clean_response_func(exchange_clean_func_name)
 90
 91        self._old_exchanges_finalize_func = edq.net.settings.get_exchanges_finalize_func()
 92        exchange_finalize_func = BACKEND_EXCHANGE_FINALIZING_FUNCS.get(self.backend_type, None)
 93        if (exchange_finalize_func is not None):
 94            exchange_finalize_func_name = edq.util.reflection.get_qualified_name(exchange_finalize_func)
 95            edq.net.settings.set_exchanges_finalize_func(exchange_finalize_func_name)
 96        else:
 97            edq.net.settings.set_exchanges_finalize_func()
 98
 99        self._old_set_exchanges_clean_response_func = lms.cli.parser._set_exchanges_clean_response_func
100        lms.cli.parser._set_exchanges_clean_response_func = False
101
102        def _request_complete_callback(exchange: edq.net.exchange.HTTPExchange) -> None:
103            # Restart if the request is a write.
104            if (edq.util.parse.boolean(exchange.headers.get(lms.model.constants.HEADER_KEY_WRITE, False))):
105                self.restart()
106
107        self._old_request_complete_callback = edq.net.settings.get_request_complete_callback()
108        edq.net.settings.set_request_complete_callback(typing.cast(edq.net.exchange.HTTPExchangeComplete, _request_complete_callback))
109
110        # Disable logging from the runner, since it may disrupt CLI tests.
111        logger = logging.getLogger('edq.testing.serverrunner')
112        self._old_serverrunner_logging_level = logger.level
113        logger.setLevel(logging.WARNING)
114
115        # Start the server.
116        super().start()
117
118    def stop(self) -> bool:
119        if (self._old_serverrunner_logging_level is not None):
120            logger = logging.getLogger('edq.testing.serverrunner')
121            logger.setLevel(self._old_serverrunner_logging_level)
122            self._old_serverrunner_logging_level = None
123
124        if (not super().stop()):
125            return False
126
127        # Restore old configs.
128
129        edq.net.settings.set_exchanges_clean_response_func(self._old_exchanges_clean_response_func)
130        self._old_exchanges_clean_response_func = None
131
132        lms.cli.parser._set_exchanges_clean_response_func = self._old_set_exchanges_clean_response_func
133        self._old_set_exchanges_clean_response_func = False
134
135        edq.net.settings.set_request_complete_callback(self._old_request_complete_callback)
136        self._old_request_complete_callback = None
137
138        edq.net.settings.set_exchanges_finalize_func(self._old_exchanges_finalize_func)
139        self._old_exchanges_finalize_func = None
140
141        return True
142
143    def restart(self) -> None:
144        super().restart()
145
146        # Inform the backend that a restart happened.
147        if (self.backend is not None):
148            self.backend.reset_connection()
149
150    def identify_server(self) ->  bool:
151        try:
152            backend_type = lms.backend.instance.guess_backend_type_from_request(self.server, timeout_secs = self.identify_wait_secs)
153        except edq.core.errors.RetryError:
154            return False
155
156        return (backend_type is not None)