test auto reconnect

Author: rmorshea

src/idom/server/sanic.py

@@ -36,10 +36,12 @@ class SanicRenderServer(AbstractRenderServer[Sanic, Config]):
 
     _loop: asyncio.AbstractEventLoop
     _dispatcher_type: Type[AbstractDispatcher]
+    _did_stop: Event
 
     def stop(self) -> None:
         """Stop the running application"""
         self._loop.call_soon_threadsafe(self.application.stop)
+        self._did_stop.wait(5)
 
     def _create_config(self, config: Optional[Config]) -> Config:
         new_config: Config = {
@@ -59,6 +61,12 @@ def _setup_application(self, config: Config, app: Sanic) -> None:
 
         self._setup_blueprint_routes(config, bp)
 
+        self._did_stop = did_stop = Event()
+
+        @app.listener("before_server_stop")  # type: ignore
+        async def server_did_stop(app: Sanic, loop: asyncio.AbstractEventLoop) -> None:
+            did_stop.set()
+
         cors_config = config["cors"]
         if cors_config:  # pragma: no cover
             cors_params = cors_config if isinstance(cors_config, dict) else {}

src/idom/testing.py

@@ -162,6 +162,7 @@ def __exit__(
         self.server.stop()
         logging.getLogger().removeHandler(self._log_handler)
         self.raise_if_logged_exception()
+        del self.mount, self.server
         return None
 
 

tests/conftest.py

@@ -3,7 +3,7 @@
 import inspect
 import logging
 import os
-from typing import Iterator, List
+from typing import Any, Iterator, List
 
 import pyalect.builtins.pytest  # noqa
 import pytest
@@ -89,10 +89,10 @@ def create_driver(driver_is_headless):
     """A Selenium web driver"""
     drivers = []
 
-    def create():
+    def create(**kwargs: Any):
         options = ChromeOptions()
         options.headless = driver_is_headless
-        driver = create_simple_selenium_web_driver(driver_options=options)
+        driver = create_simple_selenium_web_driver(driver_options=options, **kwargs)
         drivers.append(driver)
         return driver
 

tests/test_client/test_app.py

@@ -1,2 +1,38 @@
-def test_automatic_reconnect():
-    assert False
+import idom
+from idom.testing import ServerMountPoint
+
+
+def test_automatic_reconnect(create_driver):
+    # we need to wait longer here because the automatic reconnect is not instance
+    driver = create_driver(implicit_wait_timeout=10)
+
+    @idom.component
+    def OldComponent():
+        return idom.html.p({"id": "old-component"}, "old")
+
+    mount_point = ServerMountPoint()
+
+    with mount_point:
+        mount_point.mount(OldComponent)
+        driver.get(mount_point.url())
+
+    # the server is disconnected but the last view state is still shown
+    driver.find_element_by_id("old-component")
+
+    set_state = idom.Ref(None)
+
+    @idom.component
+    def NewComponent():
+        state, set_state.current = idom.hooks.use_state(0)
+        return idom.html.p({"id": f"new-component-{state}"}, f"new-{state}")
+
+    with mount_point:
+        mount_point.mount(NewComponent)
+
+        # Note the lack of a page refresh before looking up this new component. The
+        # client should attempt to reconnect and display the new view automatically.
+        driver.find_element_by_id("new-component-0")
+
+        # check that we can resume normal operation
+        set_state.current(1)
+        driver.find_element_by_id("new-component-1")