test_routes_server_threading

Tests for the threaded routes server request path.

Requests are served on threads that are never STA while pyRevit directs stderr to its script output console, a WPF window that can only be created on Revit's STA UI thread. An exception escaping a serving thread is printed there and terminates the Revit process.

ThreadedHttpServer.process_request_thread must therefore contain every failure on the request path, including the per-request cleanup that runs on an already closed socket after the server has been stopped.

Access logging runs on those same threads, so HttpRequestHandler must report every request through mlogger and never reach stderr either.

Classes

ProcessRequestThreadTests

Bases: _LoggerPatchedTestCase

Tests for ThreadedHttpServer.process_request_thread.

Methods:

test_serves_request_and_runs_cleanup()

A successful request is served and its socket cleaned up.

Source code in pyrevitlib/pyrevit/unittests/test_routes_server_threading.py

def test_serves_request_and_runs_cleanup(self):
    """A successful request is served and its socket cleaned up."""
    http_server = _FakeThreadedServer()

    http_server.process_request_thread("request", ("127.0.0.1", 5000))

    self.assertEqual([("request", ("127.0.0.1", 5000))], http_server.finish_calls)
    self.assertEqual(["request"], http_server.cleanup_calls)
    self.assertEqual([], http_server.reported_errors)

test_cleanup_failure_on_closed_socket_does_not_escape()

Cleanup raising on a closed socket must not escape the thread.

Source code in pyrevitlib/pyrevit/unittests/test_routes_server_threading.py

def test_cleanup_failure_on_closed_socket_does_not_escape(self):
    """Cleanup raising on a closed socket must not escape the thread."""
    http_server = _FakeThreadedServer(
        cleanup_error=socket.error("[Errno 9] Bad file descriptor")
    )

    # must not raise
    http_server.process_request_thread("request", ("127.0.0.1", 5000))

    self.assertEqual(["request"], http_server.cleanup_calls)

test_cleanup_failure_of_any_type_does_not_escape()

Cleanup failures are contained whatever exception type is raised.

Source code in pyrevitlib/pyrevit/unittests/test_routes_server_threading.py

def test_cleanup_failure_of_any_type_does_not_escape(self):
    """Cleanup failures are contained whatever exception type is raised."""
    for error in (
        socket.error("closed socket"),
        ValueError("socket is closed"),
        RuntimeError("cannot access a disposed object"),
    ):
        http_server = _FakeThreadedServer(cleanup_error=error)

        # must not raise
        http_server.process_request_thread("request", ("127.0.0.1", 5000))

        self.assertEqual(["request"], http_server.cleanup_calls)

test_cleanup_failure_is_not_reported_as_a_request_error()

A served request is not reported as failed when cleanup raises.

Source code in pyrevitlib/pyrevit/unittests/test_routes_server_threading.py

def test_cleanup_failure_is_not_reported_as_a_request_error(self):
    """A served request is not reported as failed when cleanup raises."""
    http_server = _FakeThreadedServer(cleanup_error=socket.error("closed socket"))

    http_server.process_request_thread("request", ("127.0.0.1", 5000))

    self.assertEqual([], http_server.reported_errors)

test_inherited_cleanup_runs_against_a_closed_socket()

The inherited cleanup path is safe on a genuinely closed socket.

Source code in pyrevitlib/pyrevit/unittests/test_routes_server_threading.py

def test_inherited_cleanup_runs_against_a_closed_socket(self):
    """The inherited cleanup path is safe on a genuinely closed socket."""
    http_server = _InheritedCleanupServer()
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.close()

    # must not raise
    http_server.process_request_thread(sock, ("127.0.0.1", 5000))

    self.assertEqual(1, len(http_server.finish_calls))

test_request_failure_is_reported()

A request that raises is routed to handle_error.

Source code in pyrevitlib/pyrevit/unittests/test_routes_server_threading.py

def test_request_failure_is_reported(self):
    """A request that raises is routed to handle_error."""
    http_server = _FakeThreadedServer(finish_error=ValueError("boom"))

    http_server.process_request_thread("request", ("127.0.0.1", 5000))

    self.assertEqual(
        [("request", ("127.0.0.1", 5000))], http_server.reported_errors
    )

test_request_failure_still_runs_cleanup()

Cleanup runs even when serving the request raised.

Source code in pyrevitlib/pyrevit/unittests/test_routes_server_threading.py

def test_request_failure_still_runs_cleanup(self):
    """Cleanup runs even when serving the request raised."""
    http_server = _FakeThreadedServer(finish_error=ValueError("boom"))

    http_server.process_request_thread("request", ("127.0.0.1", 5000))

    self.assertEqual(["request"], http_server.cleanup_calls)

test_request_and_cleanup_failures_are_both_contained()

Nothing escapes when both the request and its cleanup raise.

Source code in pyrevitlib/pyrevit/unittests/test_routes_server_threading.py

def test_request_and_cleanup_failures_are_both_contained(self):
    """Nothing escapes when both the request and its cleanup raise."""
    http_server = _FakeThreadedServer(
        finish_error=ValueError("boom"),
        cleanup_error=socket.error("closed socket"),
    )

    # must not raise
    http_server.process_request_thread("request", ("127.0.0.1", 5000))

    self.assertEqual(1, len(http_server.reported_errors))
    self.assertEqual(["request"], http_server.cleanup_calls)

test_error_reporting_failure_does_not_escape()

A failure inside handle_error must not escape either.

Source code in pyrevitlib/pyrevit/unittests/test_routes_server_threading.py

def test_error_reporting_failure_does_not_escape(self):
    """A failure inside handle_error must not escape either."""
    http_server = _FakeThreadedServer(
        finish_error=ValueError("boom"),
        report_error=RuntimeError("logging service is gone"),
    )

    # must not raise
    http_server.process_request_thread("request", ("127.0.0.1", 5000))

    self.assertEqual(["request"], http_server.cleanup_calls)

test_nothing_is_written_to_stderr()

Serving and cleanup failures must never reach stderr.

Source code in pyrevitlib/pyrevit/unittests/test_routes_server_threading.py

def test_nothing_is_written_to_stderr(self):
    """Serving and cleanup failures must never reach stderr."""
    http_server = _FakeThreadedServer(
        finish_error=ValueError("boom"),
        cleanup_error=socket.error("closed socket"),
    )
    recorder = _RecordingStream()
    original_stderr = sys.stderr
    sys.stderr = recorder
    try:
        http_server.process_request_thread("request", ("127.0.0.1", 5000))
    finally:
        sys.stderr = original_stderr

    self.assertEqual([], recorder.writes)

test_no_exception_escapes_the_serving_thread()

Run on a thread as the server does; nothing may escape it.

Source code in pyrevitlib/pyrevit/unittests/test_routes_server_threading.py

def test_no_exception_escapes_the_serving_thread(self):
    """Run on a thread as the server does; nothing may escape it."""
    http_server = _FakeThreadedServer(
        finish_error=ValueError("boom"),
        cleanup_error=socket.error("closed socket"),
        report_error=RuntimeError("logging service is gone"),
    )
    escaped = []

    def serve():
        try:
            http_server.process_request_thread("request", ("127.0.0.1", 5000))
        except BaseException as exc:  # pylint: disable=broad-except
            escaped.append(exc)

    thread = threading.Thread(target=serve)
    thread.daemon = True
    thread.start()
    thread.join(5)

    self.assertFalse(thread.is_alive())
    self.assertEqual([], escaped)

setUp()

Source code in pyrevitlib/pyrevit/unittests/test_routes_server_threading.py

def setUp(self):
    self.logger = _RecordingLogger()
    self._original_mlogger = server.mlogger
    server.mlogger = self.logger

tearDown()

Source code in pyrevitlib/pyrevit/unittests/test_routes_server_threading.py

def tearDown(self):
    server.mlogger = self._original_mlogger

HandleErrorTests

Bases: _LoggerPatchedTestCase

Tests for ThreadedHttpServer.handle_error.

Methods:

test_reports_through_the_logger()

Request errors are recorded through mlogger, not printed.

Source code in pyrevitlib/pyrevit/unittests/test_routes_server_threading.py

def test_reports_through_the_logger(self):
    """Request errors are recorded through mlogger, not printed."""
    http_server = _FakeThreadedServer()

    server.ThreadedHttpServer.handle_error(
        http_server, "request", ("127.0.0.1", 5000)
    )

    self.assertEqual(1, len(self.logger.messages))
    self.assertIn("127.0.0.1", self.logger.messages[0])

test_does_not_write_to_stderr()

handle_error must not reach the script output console.

Source code in pyrevitlib/pyrevit/unittests/test_routes_server_threading.py

def test_does_not_write_to_stderr(self):
    """handle_error must not reach the script output console."""
    http_server = _FakeThreadedServer()
    recorder = _RecordingStream()
    original_stderr = sys.stderr
    sys.stderr = recorder
    try:
        try:
            raise ValueError("boom")
        except ValueError:
            server.ThreadedHttpServer.handle_error(
                http_server, "request", ("127.0.0.1", 5000)
            )
    finally:
        sys.stderr = original_stderr

    self.assertEqual([], recorder.writes)

setUp()

Source code in pyrevitlib/pyrevit/unittests/test_routes_server_threading.py

def setUp(self):
    self.logger = _RecordingLogger()
    self._original_mlogger = server.mlogger
    server.mlogger = self.logger

tearDown()

Source code in pyrevitlib/pyrevit/unittests/test_routes_server_threading.py

def tearDown(self):
    server.mlogger = self._original_mlogger

LogMessageTests

Bases: _LoggerPatchedTestCase

Tests for HttpRequestHandler.log_message.

Methods:

test_records_the_message_through_the_logger()

Request logging is recorded through mlogger, not printed.

Source code in pyrevitlib/pyrevit/unittests/test_routes_server_threading.py

def test_records_the_message_through_the_logger(self):
    """Request logging is recorded through mlogger, not printed."""
    request_handler = _FakeRequestHandler()

    request_handler.log_message('"%s" %s', "GET /status HTTP/1.1", 200)

    self.assertEqual(1, len(self.logger.messages))
    self.assertIn("Routes request from 127.0.0.1 |", self.logger.messages[0])
    self.assertIn('"GET /status HTTP/1.1" 200', self.logger.messages[0])

test_does_not_write_to_stderr()

log_message must not reach the script output console.

Source code in pyrevitlib/pyrevit/unittests/test_routes_server_threading.py

def test_does_not_write_to_stderr(self):
    """log_message must not reach the script output console."""
    request_handler = _FakeRequestHandler()
    recorder = _RecordingStream()
    original_stderr = sys.stderr
    sys.stderr = recorder
    try:
        request_handler.log_message('"%s" %s', "GET /status HTTP/1.1", 200)
    finally:
        sys.stderr = original_stderr

    self.assertEqual([], recorder.writes)

test_served_and_rejected_requests_both_stay_off_stderr()

Every inherited logging entry point goes through the override.

Source code in pyrevitlib/pyrevit/unittests/test_routes_server_threading.py

def test_served_and_rejected_requests_both_stay_off_stderr(self):
    """Every inherited logging entry point goes through the override."""
    request_handler = _FakeRequestHandler()
    recorder = _RecordingStream()
    original_stderr = sys.stderr
    sys.stderr = recorder
    try:
        request_handler.log_request(200, 15)
        request_handler.log_error("code %d, message %s", 400, "Bad request")
    finally:
        sys.stderr = original_stderr

    self.assertEqual([], recorder.writes)
    self.assertEqual(2, len(self.logger.messages))
    self.assertIn("GET /pyrevit-core/status HTTP/1.1", self.logger.messages[0])

setUp()

Source code in pyrevitlib/pyrevit/unittests/test_routes_server_threading.py

def setUp(self):
    self.logger = _RecordingLogger()
    self._original_mlogger = server.mlogger
    server.mlogger = self.logger

tearDown()

Source code in pyrevitlib/pyrevit/unittests/test_routes_server_threading.py

def tearDown(self):
    server.mlogger = self._original_mlogger