autograder.testing.serverrunner

  1import logging
  2import typing
  3
  4import edq.net.exchange
  5import edq.net.request
  6import edq.net.settings
  7import edq.testing.serverrunner
  8import edq.util.parse
  9import edq.util.reflection
 10
 11import autograder.api.common
 12import autograder.api.constants
 13import autograder.api.metadata.heartbeat
 14import autograder.cli.parser
 15import autograder.error
 16import autograder.model.config
 17import autograder.util.net
 18
 19class ServerRunner(edq.testing.serverrunner.ServerRunner):
 20    """ A server runner specifically for the autograder server. """
 21
 22    def __init__(self,
 23            **kwargs: typing.Any) -> None:
 24        super().__init__(**kwargs)
 25
 26        self._old_exchanges_clean_response_func: typing.Union[str, None] = None
 27        """
 28        The value of edq.net.settings.get_exchanges_clean_response_func() when start() is called.
 29        The original value may be changed in start(), and will be reset in stop().
 30        """
 31
 32        self._old_set_exchanges_clean_response_func: bool = False
 33        """
 34        The value of autograder.cli.parser._set_exchanges_clean_response_func when start() is called.
 35        The original value may be changed in start(), and will be reset in stop().
 36        """
 37
 38        self._old_request_complete_callback: typing.Union[edq.net.exchange.HTTPExchangeComplete, None] = None
 39        """
 40        The value of edq.net.request._request_complete_callback when start() is called.
 41        The original value may be changed in start(), and will be reset in stop().
 42        """
 43
 44        self._old_serverrunner_logging_level: typing.Union[int, None] = None
 45        """
 46        The logging level for the edq server runner befoe tests are run.
 47        The original value may be changed in start(), and will be reset in stop().
 48        """
 49
 50    def start(self) -> None:
 51        # Put the API in testing mode.
 52        autograder.api.common.set_testing_source_info()
 53
 54        super().start()
 55
 56        # Set configs.
 57
 58        self._old_exchanges_clean_response_func = edq.net.settings.get_exchanges_clean_response_func()
 59        edq.net.settings.set_exchanges_clean_response_func(edq.util.reflection.get_qualified_name(autograder.util.net.clean_api_response))
 60
 61        self._old_set_exchanges_clean_response_func = autograder.cli.parser._set_exchanges_clean_response_func
 62        autograder.cli.parser._set_exchanges_clean_response_func = False
 63
 64        def _request_complete_callback(exchange: edq.net.exchange.HTTPExchange) -> None:
 65            # Restart if the request is a write.
 66            if (edq.util.parse.boolean(exchange.headers.get(autograder.api.constants.HEADER_KEY_WRITE, False))):
 67                self.restart()
 68
 69        self._old_request_complete_callback = edq.net.settings.get_request_complete_callback()
 70        edq.net.settings.set_request_complete_callback(typing.cast(edq.net.exchange.HTTPExchangeComplete, _request_complete_callback))
 71
 72        # Disable logging from the runner, since it may disrupt CLI tests.
 73        logger = logging.getLogger('edq.testing.serverrunner')
 74        self._old_serverrunner_logging_level = logger.level
 75        logger.setLevel(logging.WARNING)
 76
 77    def stop(self) -> bool:
 78        if (self._old_serverrunner_logging_level is not None):
 79            logger = logging.getLogger('edq.testing.serverrunner')
 80            logger.setLevel(self._old_serverrunner_logging_level)
 81            self._old_serverrunner_logging_level = None
 82
 83        if (not super().stop()):
 84            return False
 85
 86        # Restore old configs.
 87
 88        edq.net.settings.set_exchanges_clean_response_func(self._old_exchanges_clean_response_func)
 89        self._old_exchanges_clean_response_func = None
 90
 91        autograder.cli.parser._set_exchanges_clean_response_func = self._old_set_exchanges_clean_response_func
 92        self._old_set_exchanges_clean_response_func = False
 93
 94        edq.net.settings.set_request_complete_callback(self._old_request_complete_callback)
 95        self._old_request_complete_callback = None
 96
 97        return True
 98
 99    def identify_server(self) -> bool:
100        # Check the server for a heartbeat.
101        try:
102            autograder.api.metadata.heartbeat.send(autograder.model.config.Config(server = self.server), exit_on_error = False)
103        except autograder.error.ConnectionError:
104            return False
105
106        return True