tests for middleware

Author: Archmonger

src/reactpy/asgi/middleware.py

@@ -47,6 +47,18 @@ def __init__(
                 must be valid to Python's import system.
             settings: Global ReactPy configuration settings that affect behavior and performance.
         """
+        # Validate the configuration
+        if "path_prefix" in settings:
+            reason = check_path(settings["path_prefix"])
+            if reason:
+                raise ValueError(
+                    f'Invalid `path_prefix` of "{settings["path_prefix"]}". {reason}'
+                )
+        if "web_modules_dir" in settings and not settings["web_modules_dir"].exists():
+            raise ValueError(
+                f'Web modules directory "{settings["web_modules_dir"]}" does not exist.'
+            )
+
         # Process global settings
         process_settings(settings)
 
@@ -69,15 +81,6 @@ def __init__(
         self.web_modules_dir = config.REACTPY_WEB_MODULES_DIR.current
         self.static_dir = Path(__file__).parent.parent / "static"
 
-        # Validate the configuration
-        reason = check_path(self.path_prefix)
-        if reason:
-            raise ValueError(f"Invalid `path_prefix`. {reason}")
-        if not self.web_modules_dir.exists():
-            raise ValueError(
-                f"Web modules directory {self.web_modules_dir} does not exist."
-            )
-
         # Initialize the sub-applications
         self.component_dispatch_app = ComponentDispatchApp(parent=self)
         self.static_file_app = StaticFileApp(parent=self)
@@ -162,7 +165,7 @@ async def run_dispatcher(
             # Determine component to serve by analyzing the URL and/or class parameters.
             if self.parent.multiple_root_components:
                 url_match = re.match(self.parent.dispatcher_pattern, scope["path"])
-                if not url_match:
+                if not url_match:  # pragma: no cover
                     raise RuntimeError("Could not find component in URL path.")
                 dotted_path = url_match["dotted_path"]
                 if dotted_path not in self.parent.root_components:
@@ -172,7 +175,7 @@ async def run_dispatcher(
                 component = self.parent.root_components[dotted_path]
             elif self.parent.root_component:
                 component = self.parent.root_component
-            else:
+            else:  # pragma: no cover
                 raise RuntimeError("No root component provided.")
 
             # Create a connection object by analyzing the websocket's query string.

src/reactpy/asgi/utils.py

@@ -16,6 +16,9 @@
 
 def import_dotted_path(dotted_path: str) -> Any:
     """Imports a dotted path and returns the callable."""
+    if "." not in dotted_path:
+        raise ValueError(f"{dotted_path!r} is not a valid dotted path.")
+
     module_name, component_name = dotted_path.rsplit(".", 1)
 
     try:

src/reactpy/testing/backend.py

@@ -11,7 +11,7 @@
 import uvicorn
 from asgiref import typing as asgi_types
 
-from reactpy.asgi.standalone import ReactPy
+from reactpy.asgi.standalone import ReactPy, ReactPyMiddleware
 from reactpy.config import REACTPY_TESTS_DEFAULT_TIMEOUT
 from reactpy.core.component import component
 from reactpy.core.hooks import use_callback, use_effect, use_state
@@ -21,7 +21,7 @@
     list_logged_exceptions,
 )
 from reactpy.testing.utils import find_available_port
-from reactpy.types import ComponentConstructor
+from reactpy.types import ComponentConstructor, ReactPyConfig
 from reactpy.utils import Ref
 
 
@@ -37,7 +37,7 @@ class BackendFixture:
                 server.mount(MyComponent)
     """
 
-    _records: list[logging.LogRecord]
+    log_records: list[logging.LogRecord]
     _server_future: asyncio.Task[Any]
     _exit_stack = AsyncExitStack()
 
@@ -47,25 +47,33 @@ def __init__(
         host: str = "127.0.0.1",
         port: int | None = None,
         timeout: float | None = None,
+        reactpy_config: ReactPyConfig | None = None,
     ) -> None:
         self.host = host
         self.port = port or find_available_port(host)
-        self.mount, self._root_component = _hotswap()
+        self.mount = mount_to_hotswap
         self.timeout = (
             REACTPY_TESTS_DEFAULT_TIMEOUT.current if timeout is None else timeout
         )
-        self._app = app or ReactPy(self._root_component)
+        if isinstance(app, (ReactPyMiddleware, ReactPy)):
+            self._app = app
+        elif app:
+            self._app = ReactPyMiddleware(
+                app,
+                root_components=["reactpy.testing.backend.root_hotswap_component"],
+                **(reactpy_config or {}),
+            )
+        else:
+            self._app = ReactPy(
+                root_hotswap_component,
+                **(reactpy_config or {}),
+            )
         self.webserver = uvicorn.Server(
             uvicorn.Config(
                 app=self._app, host=self.host, port=self.port, loop="asyncio"
             )
         )
 
-    @property
-    def log_records(self) -> list[logging.LogRecord]:
-        """A list of captured log records"""
-        return self._records
-
     def url(self, path: str = "", query: Any | None = None) -> str:
         """Return a URL string pointing to the host and point of the server
 
@@ -108,7 +116,7 @@ def list_logged_exceptions(
 
     async def __aenter__(self) -> BackendFixture:
         self._exit_stack = AsyncExitStack()
-        self._records = self._exit_stack.enter_context(capture_reactpy_logs())
+        self.log_records = self._exit_stack.enter_context(capture_reactpy_logs())
 
         # Wait for the server to start
         Thread(target=self.webserver.run, daemon=True).start()
@@ -215,3 +223,6 @@ def swap(constructor: Callable[[], Any] | None) -> None:
             constructor_ref.current = constructor or (lambda: None)
 
     return swap, HotSwap
+
+
+mount_to_hotswap, root_hotswap_component = _hotswap()

tests/templates/index.html

@@ -0,0 +1,11 @@
+<!doctype html>
+<html lang="en">
+
+<head></head>
+
+<body>
+  <div id="app"></div>
+  {% component "reactpy.testing.backend.root_hotswap_component" %}
+</body>
+
+</html>

tests/test_asgi/test_middleware.py

@@ -0,0 +1,95 @@
+from pathlib import Path
+
+import pytest
+from jinja2 import Environment as JinjaEnvironment
+from jinja2 import FileSystemLoader as JinjaFileSystemLoader
+from starlette.applications import Starlette
+from starlette.routing import Route
+from starlette.templating import Jinja2Templates
+
+import reactpy
+from reactpy.asgi.middleware import ReactPyMiddleware
+from reactpy.testing import BackendFixture, DisplayFixture
+
+
+@pytest.fixture()
+async def display(page):
+    templates = Jinja2Templates(
+        env=JinjaEnvironment(
+            loader=JinjaFileSystemLoader("tests/templates"),
+            extensions=["reactpy.jinja.ReactPyTemplateTag"],
+        )
+    )
+
+    async def homepage(request):
+        return templates.TemplateResponse(request, "index.html")
+
+    app = Starlette(routes=[Route("/", homepage)])
+
+    async with BackendFixture(app) as server:
+        async with DisplayFixture(backend=server, driver=page) as new_display:
+            yield new_display
+
+
+def test_invalid_path_prefix():
+    with pytest.raises(ValueError, match="Invalid `path_prefix`*"):
+
+        async def app(scope, receive, send):
+            pass
+
+        reactpy.ReactPyMiddleware(app, root_components=["abc"], path_prefix="invalid")
+
+
+def test_invalid_web_modules_dir():
+    with pytest.raises(
+        ValueError, match='Web modules directory "invalid" does not exist.'
+    ):
+
+        async def app(scope, receive, send):
+            pass
+
+        reactpy.ReactPyMiddleware(
+            app, root_components=["abc"], web_modules_dir=Path("invalid")
+        )
+
+
+async def test_unregistered_root_component():
+    templates = Jinja2Templates(
+        env=JinjaEnvironment(
+            loader=JinjaFileSystemLoader("tests/templates"),
+            extensions=["reactpy.jinja.ReactPyTemplateTag"],
+        )
+    )
+
+    async def homepage(request):
+        return templates.TemplateResponse(request, "index.html")
+
+    @reactpy.component
+    def Stub():
+        return reactpy.html.p("Hello")
+
+    app = Starlette(routes=[Route("/", homepage)])
+    app = ReactPyMiddleware(app, root_components=["tests.sample.SampleApp"])
+
+    async with BackendFixture(app) as server:
+        async with DisplayFixture(backend=server) as new_display:
+            await new_display.show(Stub)
+            assert (
+                "Attempting to use an unregistered root component"
+                in server.log_records[-1].message
+            )
+
+
+async def test_display_simple_hello_world(display: DisplayFixture):
+    @reactpy.component
+    def Hello():
+        return reactpy.html.p({"id": "hello"}, ["Hello World"])
+
+    await display.show(Hello)
+
+    await display.page.wait_for_selector("#hello")
+
+    # test that we can reconnect successfully
+    await display.page.reload()
+
+    await display.page.wait_for_selector("#hello")

tests/test_asgi/test_standalone.py

@@ -12,7 +12,7 @@
 
 
 @pytest.fixture()
-async def display(page, request):
+async def display(page):
     async with BackendFixture() as server:
         async with DisplayFixture(backend=server, driver=page) as display:
             yield display