test_routes_server_lifecycle

Tests for the routes server lifecycle across a session reload.

A reload tears the session down and builds a new one on a worker thread, so every step exercised here runs off Revit's STA UI thread, where an escaping exception terminates the host. Two properties matter:

• stopping must not raise, and must release the port whatever else fails - a port left bound is a port the reloaded session cannot rebind;
• the registration must be cleared even when the stop failed, otherwise the next activation hands the dead server back instead of binding, and the reloaded session silently serves no routes at all.

Both were lost in the reload that killed Revit (#3473). Nothing on this path may write to stderr either, for the reason the request path may not: an exception printed off a worker thread is what terminated the process.

Classes

RoutesServerStopTests

Bases: TestCase

Tests for RoutesServer.stop, which runs on the session reload thread.

Methods:

setUp()

Isolate the server module logger.

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

def setUp(self):
    """Isolate the server module logger."""
    self.logger = _RecordingLogger()
    self._original_mlogger = routes_server.mlogger
    routes_server.mlogger = self.logger

tearDown()

Restore the server module logger.

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

def tearDown(self):
    """Restore the server module logger."""
    routes_server.mlogger = self._original_mlogger

test_stop_joins_the_accept_loop_and_closes_the_socket()

A stop shuts the accept loop down, releases the socket and forgets the thread.

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

def test_stop_joins_the_accept_loop_and_closes_the_socket(self):
    """A stop shuts the accept loop down, releases the socket and forgets the thread."""
    instance = self._make_routes_server()

    instance.stop()

    self.assertEqual(1, instance.server.shutdown_calls)
    self.assertEqual(1, instance.server.close_calls)
    self.assertIsNone(instance.server_thread)

test_stop_releases_the_port_even_when_shutdown_fails()

A failed shutdown must still release the listening socket.

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

def test_stop_releases_the_port_even_when_shutdown_fails(self):
    """A failed shutdown must still release the listening socket."""
    instance = self._make_routes_server(
        http_server=_StubHttpServer(shutdown_error=OSError("bad file descriptor"))
    )

    instance.stop()

    self.assertEqual(1, instance.server.close_calls)

test_stop_contains_failures_of_either_kind()

Shutdown and close failures are contained whatever exception type is raised.

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

def test_stop_contains_failures_of_either_kind(self):
    """Shutdown and close failures are contained whatever exception type is raised."""
    for error in (
        OSError("bad file descriptor"),
        ValueError("socket is closed"),
        RuntimeError("cannot access a disposed object"),
    ):
        instance = self._make_routes_server(
            http_server=_StubHttpServer(shutdown_error=error, close_error=error)
        )

        instance.stop()

test_stop_does_not_wait_on_a_dead_accept_loop()

An exited accept loop can never acknowledge, so the stop must skip the wait.

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

def test_stop_does_not_wait_on_a_dead_accept_loop(self):
    """An exited accept loop can never acknowledge, so the stop must skip the wait."""
    instance = self._make_routes_server(server_thread="dead")

    instance.stop()

    self.assertEqual(0, instance.server.shutdown_calls)
    self.assertEqual(1, instance.server.close_calls)

test_stop_gives_up_on_a_loop_that_never_acknowledges()

A shutdown that never returns must not hold the reload.

BaseServer.shutdown blocks until the accept loop acknowledges, with no timeout of its own, so a wedged loop would keep the port bound and the registration alive - the reload would never finish. The wait is bounded instead, and the port is released on the way out either way.

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

def test_stop_gives_up_on_a_loop_that_never_acknowledges(self):
    """A shutdown that never returns must not hold the reload.

    BaseServer.shutdown blocks until the accept loop acknowledges, with no
    timeout of its own, so a wedged loop would keep the port bound and the
    registration alive - the reload would never finish. The wait is bounded
    instead, and the port is released on the way out either way.
    """
    http_server = _UnacknowledgedHttpServer()
    instance = self._make_routes_server(http_server=http_server)

    try:
        started = time.time()
        instance.stop(timeout=0.05)
        elapsed = time.time() - started

        self.assertLess(
            elapsed,
            _UnacknowledgedHttpServer.release_after / 2,
            "the stop waited for the shutdown instead of bounding it",
        )
        self.assertEqual(1, http_server.close_calls)
        self.assertIsNone(instance.server_thread)
        self.assertTrue(
            self.logger.has_errors(),
            "a loop that outlives the wait is reported, not silently dropped",
        )
    finally:
        http_server.release.set()

test_stop_spends_one_budget_across_both_waits()

A wedged loop must not cost the reload one timeout per wait.

The shutdown request and the join that follows each get a timeout, so taken separately they let a reload wait twice what it was promised. The second wait has to be given only what the first one left over.

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

def test_stop_spends_one_budget_across_both_waits(self):
    """A wedged loop must not cost the reload one timeout per wait.

    The shutdown request and the join that follows each get a timeout, so
    taken separately they let a reload wait twice what it was promised. The
    second wait has to be given only what the first one left over.
    """
    http_server = _UnacknowledgedHttpServer()
    instance = self._make_routes_server(
        http_server=http_server, server_thread="unstoppable"
    )
    accept_loop = instance.server_thread

    try:
        instance.stop(timeout=0.05)

        self.assertEqual(1, len(accept_loop.join_timeouts))
        self.assertLess(
            accept_loop.join_timeouts[0],
            0.05,
            "the join repeated a budget the shutdown request had already spent",
        )
    finally:
        http_server.release.set()

test_stop_does_not_wait_on_a_server_that_never_started()

A server whose accept loop never started must still be releasable.

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

def test_stop_does_not_wait_on_a_server_that_never_started(self):
    """A server whose accept loop never started must still be releasable."""
    instance = self._make_routes_server(server_thread="never-started")

    instance.stop()

    self.assertEqual(0, instance.server.shutdown_calls)
    self.assertEqual(1, instance.server.close_calls)

test_stop_is_safe_to_call_twice()

Deactivation may reach an already stopped server.

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

def test_stop_is_safe_to_call_twice(self):
    """Deactivation may reach an already stopped server."""
    instance = self._make_routes_server()

    instance.stop()
    shutdown_calls = instance.server.shutdown_calls
    instance.stop()

    self.assertEqual(
        shutdown_calls,
        instance.server.shutdown_calls,
        "a second stop has no accept loop left to shut down",
    )

test_stop_does_not_write_to_stderr()

A failing stop must never reach the script output console.

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

def test_stop_does_not_write_to_stderr(self):
    """A failing stop must never reach the script output console."""
    instance = self._make_routes_server(
        http_server=_StubHttpServer(
            shutdown_error=OSError("bad file descriptor"),
            close_error=OSError("bad file descriptor"),
        )
    )
    recorder = _RecordingStream()
    original_stderr = sys.stderr
    sys.stderr = recorder
    try:
        instance.stop()
    finally:
        sys.stderr = original_stderr

    self.assertEqual([], recorder.writes)

RoutesServerStartTests

Bases: TestCase

Tests for the accept loop that RoutesServer.start owns.

Methods:

setUp()

Isolate the server module logger.

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

def setUp(self):
    """Isolate the server module logger."""
    self.logger = _RecordingLogger()
    self._original_mlogger = routes_server.mlogger
    routes_server.mlogger = self.logger
    self._servers = []

tearDown()

Restore the server module logger.

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

def tearDown(self):
    """Restore the server module logger."""
    for instance in self._servers:
        instance.stop(timeout=0.5)
    routes_server.mlogger = self._original_mlogger

test_a_live_server_runs_exactly_one_accept_loop()

A second start must not put a second accept loop on the same socket.

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

def test_a_live_server_runs_exactly_one_accept_loop(self):
    """A second start must not put a second accept loop on the same socket."""
    instance = self._start_real_server()

    first_thread = instance.server_thread
    instance.start()

    self.assertIs(first_thread, instance.server_thread)
    self.assertTrue(instance.is_running)

test_a_stopped_server_reports_itself_as_not_running()

A stopped server no longer reports a live accept loop.

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

def test_a_stopped_server_reports_itself_as_not_running(self):
    """A stopped server no longer reports a live accept loop."""
    instance = self._start_real_server()

    instance.stop()

    self.assertFalse(instance.is_running)

test_stop_releases_the_port_for_the_next_session()

The next session can bind the port the stopped server had.

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

def test_stop_releases_the_port_for_the_next_session(self):
    """The next session can bind the port the stopped server had."""
    instance = self._start_real_server()
    port = instance.server.server_address[1]

    instance.stop()

    replacement = routes_server.RoutesServer(host="127.0.0.1", port=port)
    self._servers.append(replacement)
    self.assertEqual(port, replacement.server.server_address[1])

test_request_threads_are_not_joined_on_close()

A request in flight during a reload must not hold the host.

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

def test_request_threads_are_not_joined_on_close(self):
    """A request in flight during a reload must not hold the host."""
    self.assertTrue(routes_server.ThreadedHttpServer.daemon_threads)
    self.assertFalse(routes_server.ThreadedHttpServer.block_on_close)

DeactivateServerTests

Bases: _LifecycleTestCase

Tests for routes.deactivate_server, the reload's teardown step.

Methods:

test_deactivation_stops_and_deregisters_the_server()

Deactivation stops the server and drops its registration.

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

def test_deactivation_stops_and_deregisters_the_server(self):
    """Deactivation stops the server and drops its registration."""
    stub_server = _StubRoutesServer()
    self._activate(stub_server)

    routes.deactivate_server()

    self.assertEqual(1, stub_server.stop_calls)
    self.assertIsNone(self.env_vars.get(envvars.ROUTES_SERVER))
    self.assertEqual(1, len(self.unregister_calls))

test_deactivation_clears_registration_even_when_stop_fails()

A dead server must not stay registered: the next activation would skip binding.

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

def test_deactivation_clears_registration_even_when_stop_fails(self):
    """A dead server must not stay registered: the next activation would skip binding."""
    self._activate(_StubRoutesServer(stop_error=RuntimeError("socket is closed")))

    routes.deactivate_server()

    self.assertIsNone(self.env_vars.get(envvars.ROUTES_SERVER))
    self.assertEqual(1, len(self.unregister_calls))

test_deactivation_survives_a_failing_unregister()

A failed deregistration still clears the session env var.

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

def test_deactivation_survives_a_failing_unregister(self):
    """A failed deregistration still clears the session env var."""
    self._activate(_StubRoutesServer())
    serverinfo.unregister = self._raise_unregister

    routes.deactivate_server()

    self.assertIsNone(self.env_vars.get(envvars.ROUTES_SERVER))

test_deactivation_deregisters_when_clearing_the_env_var_fails()

A failing env var write must not skip the deregistration.

The deregistration is what keeps the next session from being handed this server, so it cannot be the step that an earlier failure takes out.

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

def test_deactivation_deregisters_when_clearing_the_env_var_fails(self):
    """A failing env var write must not skip the deregistration.

    The deregistration is what keeps the next session from being handed this
    server, so it cannot be the step that an earlier failure takes out.
    """
    self._activate(_StubRoutesServer())
    self.env_vars.set_error = RuntimeError("the session is already gone")

    routes.deactivate_server()

    self.assertEqual(1, len(self.unregister_calls))
    self.assertTrue(self.logger.has_errors())

test_deactivation_with_no_active_server_does_nothing()

Deactivating with nothing active touches nothing.

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

def test_deactivation_with_no_active_server_does_nothing(self):
    """Deactivating with nothing active touches nothing."""
    routes.deactivate_server()

    self.assertEqual([], self.unregister_calls)

test_deactivation_does_not_write_to_stderr()

A failing deactivation must never reach the script output console.

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

def test_deactivation_does_not_write_to_stderr(self):
    """A failing deactivation must never reach the script output console."""
    self._activate(_StubRoutesServer(stop_error=RuntimeError("socket is closed")))
    recorder = _RecordingStream()
    original_stderr = sys.stderr
    sys.stderr = recorder
    try:
        routes.deactivate_server()
    finally:
        sys.stderr = original_stderr

    self.assertEqual([], recorder.writes)

test_a_failed_stop_is_reported()

A stop that failed during a reload has to be visible.

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

def test_a_failed_stop_is_reported(self):
    """A stop that failed during a reload has to be visible."""
    self._activate(_StubRoutesServer(stop_error=RuntimeError("socket is closed")))

    routes.deactivate_server()

    self.assertTrue(
        self.logger.has_errors(),
        "a stop that failed during a reload has to be visible",
    )

setUp()

Swap in the recording logger, env var store and registry stubs.

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

def setUp(self):
    """Swap in the recording logger, env var store and registry stubs."""
    self.logger = _RecordingLogger()
    self._original_mlogger = routes.mlogger
    routes.mlogger = self.logger
    self.env_vars = _EnvVarStore()
    self.unregister_calls = []
    self.register_calls = []
    self._original_get = envvars.get_pyrevit_env_var
    self._original_set = envvars.set_pyrevit_env_var
    self._original_register = serverinfo.register
    self._original_unregister = serverinfo.unregister
    envvars.get_pyrevit_env_var = self.env_vars.get
    envvars.set_pyrevit_env_var = self.env_vars.set
    serverinfo.register = self._register
    serverinfo.unregister = self._count_unregister

tearDown()

Restore the session env var access and the serverinfo registry.

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

def tearDown(self):
    """Restore the session env var access and the serverinfo registry."""
    routes.mlogger = self._original_mlogger
    envvars.get_pyrevit_env_var = self._original_get
    envvars.set_pyrevit_env_var = self._original_set
    serverinfo.register = self._original_register
    serverinfo.unregister = self._original_unregister

ActivateServerTests

Bases: _LifecycleTestCase

Tests for routes.activate_server, the reload's last step.

Methods:

test_activation_hands_back_a_server_that_is_already_active()

Activating twice in one session reuses the running server.

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

def test_activation_hands_back_a_server_that_is_already_active(self):
    """Activating twice in one session reuses the running server."""
    stub_server = _StubRoutesServer()
    self._activate(stub_server)

    self.assertIs(stub_server, routes.activate_server())
    self.assertEqual(0, stub_server.stop_calls)

test_activation_replaces_a_server_whose_accept_loop_exited()

A registered server that no longer accepts is torn down, not handed back.

Handing it back is the dead registration #3473 was reported for, reached from the other end: the server is in the env var but nothing is listening.

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

def test_activation_replaces_a_server_whose_accept_loop_exited(self):
    """A registered server that no longer accepts is torn down, not handed back.

    Handing it back is the dead registration #3473 was reported for, reached
    from the other end: the server is in the env var but nothing is listening.
    """
    dead_server = _StubRoutesServer(is_running=False)
    replacement = self._patch_routes_server_class(_StubRoutesServer())
    self._activate(dead_server)

    self.assertIs(replacement, routes.activate_server())
    self.assertEqual(1, dead_server.stop_calls)
    self.assertEqual(1, len(self.register_calls))
    self.assertIs(replacement, self.env_vars.get(envvars.ROUTES_SERVER))

test_init_stops_the_active_server_before_activating_again()

Every session load still stops the previous server.

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

def test_init_stops_the_active_server_before_activating_again(self):
    """Every session load still stops the previous server."""
    stub_server = _StubRoutesServer()
    self._activate(stub_server)

    routes.init()

    self.assertEqual(1, stub_server.stop_calls)
    self.assertIsNone(self.env_vars.get(envvars.ROUTES_SERVER))

test_a_failed_activation_is_reported()

A server that cannot bind the port is reported.

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

def test_a_failed_activation_is_reported(self):
    """A server that cannot bind the port is reported."""
    serverinfo.register = self._raise_register

    self.assertIsNone(routes.activate_server())

    self.assertTrue(self.logger.has_errors())
    self.assertEqual(1, len(self.unregister_calls))

test_a_failed_activation_survives_a_failing_unregister()

A failed activation reports both failures instead of raising the second.

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

def test_a_failed_activation_survives_a_failing_unregister(self):
    """A failed activation reports both failures instead of raising the second."""
    serverinfo.register = self._raise_register
    serverinfo.unregister = self._raise_unregister

    self.assertIsNone(routes.activate_server())

    self.assertTrue(self.logger.has_errors())

setUp()

Swap in the recording logger, env var store and registry stubs.

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

def setUp(self):
    """Swap in the recording logger, env var store and registry stubs."""
    self.logger = _RecordingLogger()
    self._original_mlogger = routes.mlogger
    routes.mlogger = self.logger
    self.env_vars = _EnvVarStore()
    self.unregister_calls = []
    self.register_calls = []
    self._original_get = envvars.get_pyrevit_env_var
    self._original_set = envvars.set_pyrevit_env_var
    self._original_register = serverinfo.register
    self._original_unregister = serverinfo.unregister
    envvars.get_pyrevit_env_var = self.env_vars.get
    envvars.set_pyrevit_env_var = self.env_vars.set
    serverinfo.register = self._register
    serverinfo.unregister = self._count_unregister

tearDown()

Restore the session env var access and the serverinfo registry.

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

def tearDown(self):
    """Restore the session env var access and the serverinfo registry."""
    routes.mlogger = self._original_mlogger
    envvars.get_pyrevit_env_var = self._original_get
    envvars.set_pyrevit_env_var = self._original_set
    serverinfo.register = self._original_register
    serverinfo.unregister = self._original_unregister