Fix backend detection in testing runners

dash/testing/application_runners.py

@@ -28,6 +28,14 @@
 logger = logging.getLogger(__name__)
 
 
+def _run_app(app, options):
+    server_type = getattr(getattr(app, "backend", None), "server_type", "flask")
+    if server_type in ("fastapi", "quart"):
+        app.run(**options)
+    else:
+        app.run(threaded=True, **options)
+
+
 def import_app(app_file, application_name="app"):
     """Import a dash application from a module. The import path is in dot
     notation to the module. The variable named app will be returned.
@@ -173,16 +181,7 @@ def run():
                 self.port = options["port"]
 
             try:
-                module = app.server.__class__.__module__
-                # FastAPI support
-                if module.startswith("fastapi"):
-                    app.run(**options)
-                # Quart support (ASGI - runs its own async event loop)
-                elif module.startswith("quart"):
-                    app.run(**options)
-                # Flask fallback (WSGI - needs threaded mode)
-                else:
-                    app.run(threaded=True, **options)
+                _run_app(app, options)
             except SystemExit:
                 logger.info("Server stopped")
             except Exception as error:
@@ -264,16 +263,7 @@ def target():
             options = kwargs.copy()
 
             try:
-                module = app.server.__class__.__module__
-                # FastAPI support
-                if module.startswith("fastapi"):
-                    app.run(**options)
-                # Quart support (ASGI - runs its own async event loop)
-                elif module.startswith("quart"):
-                    app.run(**options)
-                # Flask fallback (WSGI - needs threaded mode)
-                else:
-                    app.run(threaded=True, **options)
+                _run_app(app, options)
             except SystemExit:
                 logger.info("Server stopped")
                 raise

tests/unit/test_app_runners.py

@@ -1,10 +1,15 @@
 import os
 import sys
+import time
+from types import SimpleNamespace
+from unittest.mock import Mock
+
 import requests
 import pytest
 
 import dash
 from dash import html
+from dash.testing.application_runners import ThreadedRunner, _run_app
 
 
 def test_threaded_server_smoke(dash_thread_server):
@@ -22,6 +27,37 @@ def test_threaded_server_smoke(dash_thread_server):
     assert 'id="react-entry-point"' in r.text, "the entrypoint is present"
 
 
+def test_threaded_server_wrapped_fastapi(monkeypatch):
+    wrapped_server = type(
+        "WrappedFastAPI", (), {"__module__": "instrumentation.wrapper"}
+    )()
+    uvicorn_server = SimpleNamespace(should_exit=False)
+    run_options = {}
+
+    def run(**options):
+        run_options.update(options)
+        while not uvicorn_server.should_exit:
+            time.sleep(0.01)
+
+    app = SimpleNamespace(
+        server=wrapped_server,
+        backend=SimpleNamespace(server_type="fastapi"),
+        scripts=SimpleNamespace(config=SimpleNamespace(serve_locally=False)),
+        css=SimpleNamespace(config=SimpleNamespace(serve_locally=False)),
+        run=run,
+        _uvicorn_server=uvicorn_server,
+    )
+    runner = ThreadedRunner()
+    monkeypatch.setattr(runner, "accessible", lambda _url: True)
+
+    try:
+        runner.start(app)
+        assert "threaded" not in run_options
+    finally:
+        if runner.started:
+            runner.stop()
+
+
 @pytest.mark.skipif(
     sys.version_info < (3,), reason="requires python3 for process testing"
 )
@@ -37,3 +73,26 @@ def test_process_server_smoke(dash_process_server):
         assert 'id="react-entry-point"' in r.text, "the entrypoint is present"
     finally:
         os.chdir(cwd)
+
+
+@pytest.mark.parametrize(
+    ("server_type", "expected_options"),
+    [
+        ("fastapi", {"port": 8050}),
+        ("quart", {"port": 8050}),
+        ("flask", {"port": 8050, "threaded": True}),
+    ],
+)
+def test_run_app_uses_backend_type(server_type, expected_options):
+    wrapped_server = type(
+        "WrappedServer", (), {"__module__": "instrumentation.wrapper"}
+    )()
+    app = SimpleNamespace(
+        server=wrapped_server,
+        backend=SimpleNamespace(server_type=server_type),
+        run=Mock(),
+    )
+
+    _run_app(app, {"port": 8050})
+
+    app.run.assert_called_once_with(**expected_options)