diff --git a/tests/test_app/settings_multi_db.py b/tests/test_app/settings_multi_db.py index b0eeacc1..89784309 100644 --- a/tests/test_app/settings_multi_db.py +++ b/tests/test_app/settings_multi_db.py @@ -110,7 +110,6 @@ LANGUAGE_CODE = "en-us" TIME_ZONE = "UTC" USE_I18N = True -USE_L10N = True USE_TZ = True # Default primary key field type diff --git a/tests/test_app/settings_single_db.py b/tests/test_app/settings_single_db.py index e4dbc9aa..ed63dbe0 100644 --- a/tests/test_app/settings_single_db.py +++ b/tests/test_app/settings_single_db.py @@ -98,7 +98,6 @@ LANGUAGE_CODE = "en-us" TIME_ZONE = "UTC" USE_I18N = True -USE_L10N = True USE_TZ = True # Default primary key field type diff --git a/tests/test_app/tests/test_components.py b/tests/test_app/tests/test_components.py index 90d8b93c..b456e67b 100644 --- a/tests/test_app/tests/test_components.py +++ b/tests/test_app/tests/test_components.py @@ -1,112 +1,25 @@ -import asyncio import os import socket -import sys -from functools import partial from time import sleep -from channels.testing import ChannelsLiveServerTestCase -from channels.testing.live import make_application -from django.core.exceptions import ImproperlyConfigured -from django.core.management import call_command -from django.db import connections -from django.test.utils import modify_settings -from playwright.sync_api import TimeoutError, sync_playwright +from playwright.sync_api import TimeoutError from reactpy_django.models import ComponentSession from reactpy_django.utils import strtobool -GITHUB_ACTIONS = os.getenv("GITHUB_ACTIONS", "False") -CLICK_DELAY = 250 if strtobool(GITHUB_ACTIONS) else 25 # Delay in miliseconds. +from .utils import GITHUB_ACTIONS, PlaywrightTestCase +CLICK_DELAY = 250 if strtobool(GITHUB_ACTIONS) else 25 # Delay in miliseconds. -class ComponentTests(ChannelsLiveServerTestCase): - from reactpy_django import config +class GenericComponentTests(PlaywrightTestCase): databases = {"default"} @classmethod def setUpClass(cls): - # Repurposed from ChannelsLiveServerTestCase._pre_setup - for connection in connections.all(): - if cls._is_in_memory_db(cls, connection): - raise ImproperlyConfigured( - "ChannelLiveServerTestCase can not be used with in memory databases" - ) - cls._live_server_modified_settings = modify_settings( - ALLOWED_HOSTS={"append": cls.host} - ) - cls._live_server_modified_settings.enable() - get_application = partial( - make_application, - static_wrapper=cls.static_wrapper if cls.serve_static else None, - ) - cls._server_process = cls.ProtocolServerProcess(cls.host, get_application) - cls._server_process.start() - cls._server_process.ready.wait() - cls._port = cls._server_process.port.value - - # Open the second server process, used for testing custom hosts - cls._server_process2 = cls.ProtocolServerProcess(cls.host, get_application) - cls._server_process2.start() - cls._server_process2.ready.wait() - cls._port2 = cls._server_process2.port.value - - # Open the third server process, used for testing offline fallback - cls._server_process3 = cls.ProtocolServerProcess(cls.host, get_application) - cls._server_process3.start() - cls._server_process3.ready.wait() - cls._port3 = cls._server_process3.port.value - - # Open a Playwright browser window - if sys.platform == "win32": - asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy()) - cls.playwright = sync_playwright().start() - headless = strtobool(os.environ.get("PLAYWRIGHT_HEADLESS", GITHUB_ACTIONS)) - cls.browser = cls.playwright.chromium.launch(headless=bool(headless)) - cls.page = cls.browser.new_page() - cls.page.set_default_timeout(5000) - - @classmethod - def tearDownClass(cls): - from django.db import DEFAULT_DB_ALIAS - - from reactpy_django import config - - # Close the Playwright browser - cls.playwright.stop() - - # Close the other server processes - cls._server_process.terminate() - cls._server_process.join() - cls._server_process2.terminate() - cls._server_process2.join() - cls._server_process3.terminate() - cls._server_process3.join() - - # Repurposed from ChannelsLiveServerTestCase._post_teardown - cls._live_server_modified_settings.disable() - for db_name in [DEFAULT_DB_ALIAS, config.REACTPY_DATABASE]: - call_command( - "flush", - verbosity=0, - interactive=False, - database=db_name, - reset_sequences=False, - ) - - def _pre_setup(self): - """Handled manually in `setUpClass` to speed things up.""" - - def _post_teardown(self): - """Handled manually in `tearDownClass` to prevent TransactionTestCase from doing - database flushing. This is needed to prevent a `SynchronousOnlyOperation` from - occuring due to a bug within `ChannelsLiveServerTestCase`.""" - - def setUp(self): - if self.page.url == "about:blank": - self.page.goto(self.live_server_url) + super().setUpClass() + cls.page.goto(f"http://{cls.host}:{cls._port}") def test_hello_world(self): self.page.wait_for_selector("#hello-world") @@ -298,148 +211,6 @@ def test_component_session_missing(self): os.environ.pop("DJANGO_ALLOW_ASYNC_UNSAFE") self.assertFalse(query_exists) - def test_custom_host(self): - """Make sure that the component is rendered by a separate server.""" - new_page = self.browser.new_page() - new_page.goto(f"{self.live_server_url}/port/{self._port2}/") - try: - elem = new_page.locator(".custom_host-0") - elem.wait_for() - self.assertIn( - f"Server Port: {self._port2}", - elem.text_content(), - ) - finally: - new_page.close() - - def test_custom_host_wrong_port(self): - """Make sure that other ports are not rendering components.""" - new_page = self.browser.new_page() - try: - tmp_sock = socket.socket() - tmp_sock.bind((self._server_process.host, 0)) - random_port = tmp_sock.getsockname()[1] - new_page.goto(f"{self.live_server_url}/port/{random_port}/") - with self.assertRaises(TimeoutError): - new_page.locator(".custom_host").wait_for(timeout=1000) - finally: - new_page.close() - - def test_host_roundrobin(self): - """Verify if round-robin host selection is working.""" - new_page = self.browser.new_page() - new_page.goto(f"{self.live_server_url}/roundrobin/{self._port}/{self._port2}/8") - try: - elem0 = new_page.locator(".custom_host-0") - elem1 = new_page.locator(".custom_host-1") - elem2 = new_page.locator(".custom_host-2") - elem3 = new_page.locator(".custom_host-3") - - elem0.wait_for() - elem1.wait_for() - elem2.wait_for() - elem3.wait_for() - - current_ports = { - elem0.get_attribute("data-port"), - elem1.get_attribute("data-port"), - elem2.get_attribute("data-port"), - elem3.get_attribute("data-port"), - } - correct_ports = { - str(self._port), - str(self._port2), - } - - # There should only be two ports in the set - self.assertEqual(current_ports, correct_ports) - self.assertEqual(len(current_ports), 2) - finally: - new_page.close() - - def test_prerender(self): - """Verify if round-robin host selection is working.""" - new_page = self.browser.new_page() - new_page.goto(f"{self.live_server_url}/prerender/") - try: - string = new_page.locator("#prerender_string") - vdom = new_page.locator("#prerender_vdom") - component = new_page.locator("#prerender_component") - use_root_id_http = new_page.locator("#use-root-id-http") - use_root_id_ws = new_page.locator("#use-root-id-ws") - use_user_http = new_page.locator("#use-user-http[data-success=True]") - use_user_ws = new_page.locator("#use-user-ws[data-success=true]") - - # Check if the prerender occurred properly - string.wait_for() - vdom.wait_for() - component.wait_for() - use_root_id_http.wait_for() - use_user_http.wait_for() - self.assertEqual( - string.all_inner_texts(), ["prerender_string: Prerendered"] - ) - self.assertEqual(vdom.all_inner_texts(), ["prerender_vdom: Prerendered"]) - self.assertEqual( - component.all_inner_texts(), ["prerender_component: Prerendered"] - ) - root_id_value = use_root_id_http.get_attribute("data-value") - self.assertEqual(len(root_id_value), 36) - - # Check if the full render occurred - sleep(2) - self.assertEqual( - string.all_inner_texts(), ["prerender_string: Fully Rendered"] - ) - self.assertEqual(vdom.all_inner_texts(), ["prerender_vdom: Fully Rendered"]) - self.assertEqual( - component.all_inner_texts(), ["prerender_component: Fully Rendered"] - ) - use_root_id_ws.wait_for() - use_user_ws.wait_for() - self.assertEqual(use_root_id_ws.get_attribute("data-value"), root_id_value) - - finally: - new_page.close() - - def test_component_errors(self): - new_page = self.browser.new_page() - new_page.goto(f"{self.live_server_url}/errors/") - try: - # ComponentDoesNotExistError - broken_component = new_page.locator("#component_does_not_exist_error") - broken_component.wait_for() - self.assertIn( - "ComponentDoesNotExistError:", broken_component.text_content() - ) - - # ComponentParamError - broken_component = new_page.locator("#component_param_error") - broken_component.wait_for() - self.assertIn("ComponentParamError:", broken_component.text_content()) - - # InvalidHostError - broken_component = new_page.locator("#invalid_host_error") - broken_component.wait_for() - self.assertIn("InvalidHostError:", broken_component.text_content()) - - # SynchronousOnlyOperation - broken_component = new_page.locator("#broken_postprocessor_query pre") - broken_component.wait_for() - self.assertIn("SynchronousOnlyOperation:", broken_component.text_content()) - - # ViewNotRegisteredError - broken_component = new_page.locator("#view_to_iframe_not_registered pre") - broken_component.wait_for() - self.assertIn("ViewNotRegisteredError:", broken_component.text_content()) - - # DecoratorParamError - broken_component = new_page.locator("#incorrect_user_passes_test_decorator") - broken_component.wait_for() - self.assertIn("DecoratorParamError:", broken_component.text_content()) - finally: - new_page.close() - def test_use_user_data(self): text_input = self.page.wait_for_selector("#use-user-data input") login_1 = self.page.wait_for_selector("#use-user-data .login-1") @@ -536,170 +307,333 @@ def test_use_user_data_with_default(self): user_data_div.text_content(), ) + +class PrerenderTests(PlaywrightTestCase): + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.page.goto(f"http://{cls.host}:{cls._port}/prerender/") + + def test_prerender(self): + """Verify if round-robin host selection is working.""" + string = self.page.locator("#prerender_string") + vdom = self.page.locator("#prerender_vdom") + component = self.page.locator("#prerender_component") + use_root_id_http = self.page.locator("#use-root-id-http") + use_root_id_ws = self.page.locator("#use-root-id-ws") + use_user_http = self.page.locator("#use-user-http[data-success=True]") + use_user_ws = self.page.locator("#use-user-ws[data-success=true]") + + # Check if the prerender occurred properly + string.wait_for() + vdom.wait_for() + component.wait_for() + use_root_id_http.wait_for() + use_user_http.wait_for() + self.assertEqual(string.all_inner_texts(), ["prerender_string: Prerendered"]) + self.assertEqual(vdom.all_inner_texts(), ["prerender_vdom: Prerendered"]) + self.assertEqual( + component.all_inner_texts(), ["prerender_component: Prerendered"] + ) + root_id_value = use_root_id_http.get_attribute("data-value") + self.assertEqual(len(root_id_value), 36) + + # Check if the full render occurred + sleep(2) + self.assertEqual(string.all_inner_texts(), ["prerender_string: Fully Rendered"]) + self.assertEqual(vdom.all_inner_texts(), ["prerender_vdom: Fully Rendered"]) + self.assertEqual( + component.all_inner_texts(), ["prerender_component: Fully Rendered"] + ) + use_root_id_ws.wait_for() + use_user_ws.wait_for() + self.assertEqual(use_root_id_ws.get_attribute("data-value"), root_id_value) + + +class ErrorTests(PlaywrightTestCase): + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.page.goto(f"http://{cls.host}:{cls._port}/errors/") + + def test_component_does_not_exist_error(self): + broken_component = self.page.locator("#component_does_not_exist_error") + broken_component.wait_for() + self.assertIn("ComponentDoesNotExistError:", broken_component.text_content()) + + def test_component_param_error(self): + broken_component = self.page.locator("#component_param_error") + broken_component.wait_for() + self.assertIn("ComponentParamError:", broken_component.text_content()) + + def test_invalid_host_error(self): + broken_component = self.page.locator("#invalid_host_error") + broken_component.wait_for() + self.assertIn("InvalidHostError:", broken_component.text_content()) + + def test_synchronous_only_operation_error(self): + broken_component = self.page.locator("#broken_postprocessor_query pre") + broken_component.wait_for() + self.assertIn("SynchronousOnlyOperation:", broken_component.text_content()) + + def test_view_not_registered_error(self): + broken_component = self.page.locator("#view_to_iframe_not_registered pre") + broken_component.wait_for() + self.assertIn("ViewNotRegisteredError:", broken_component.text_content()) + + def test_decorator_param_error(self): + broken_component = self.page.locator("#incorrect_user_passes_test_decorator") + broken_component.wait_for() + self.assertIn("DecoratorParamError:", broken_component.text_content()) + + +class UrlRouterTests(PlaywrightTestCase): + def test_url_router(self): - new_page = self.browser.new_page() - try: - new_page.goto(f"{self.live_server_url}/router/") - path = new_page.wait_for_selector("#router-path") - self.assertIn("/router/", path.get_attribute("data-path")) - string = new_page.query_selector("#router-string") - self.assertEqual("/router/", string.text_content()) - - new_page.goto(f"{self.live_server_url}/router/subroute/") - path = new_page.wait_for_selector("#router-path") - self.assertIn("/router/subroute/", path.get_attribute("data-path")) - string = new_page.query_selector("#router-string") - self.assertEqual("subroute/", string.text_content()) - - new_page.goto(f"{self.live_server_url}/router/unspecified/123/") - path = new_page.wait_for_selector("#router-path") - self.assertIn("/router/unspecified/123/", path.get_attribute("data-path")) - string = new_page.query_selector("#router-string") - self.assertEqual("/router/unspecified//", string.text_content()) - - new_page.goto(f"{self.live_server_url}/router/integer/123/") - path = new_page.wait_for_selector("#router-path") - self.assertIn("/router/integer/123/", path.get_attribute("data-path")) - string = new_page.query_selector("#router-string") - self.assertEqual("/router/integer//", string.text_content()) - - new_page.goto(f"{self.live_server_url}/router/path/abc/123/") - path = new_page.wait_for_selector("#router-path") - self.assertIn("/router/path/abc/123/", path.get_attribute("data-path")) - string = new_page.query_selector("#router-string") - self.assertEqual("/router/path//", string.text_content()) - - new_page.goto(f"{self.live_server_url}/router/slug/abc-123/") - path = new_page.wait_for_selector("#router-path") - self.assertIn("/router/slug/abc-123/", path.get_attribute("data-path")) - string = new_page.query_selector("#router-string") - self.assertEqual("/router/slug//", string.text_content()) - - new_page.goto(f"{self.live_server_url}/router/string/abc/") - path = new_page.wait_for_selector("#router-path") - self.assertIn("/router/string/abc/", path.get_attribute("data-path")) - string = new_page.query_selector("#router-string") - self.assertEqual("/router/string//", string.text_content()) - - new_page.goto( - f"{self.live_server_url}/router/uuid/123e4567-e89b-12d3-a456-426614174000/" - ) - path = new_page.wait_for_selector("#router-path") - self.assertIn( - "/router/uuid/123e4567-e89b-12d3-a456-426614174000/", - path.get_attribute("data-path"), - ) - string = new_page.query_selector("#router-string") - self.assertEqual("/router/uuid//", string.text_content()) + self.page.goto(f"{self.live_server_url}/router/") + path = self.page.wait_for_selector("#router-path") + self.assertIn("/router/", path.get_attribute("data-path")) + string = self.page.query_selector("#router-string") + self.assertEqual("/router/", string.text_content()) + + def test_url_router_subroute(self): + self.page.goto(f"{self.live_server_url}/router/subroute/") + path = self.page.wait_for_selector("#router-path") + self.assertIn("/router/subroute/", path.get_attribute("data-path")) + string = self.page.query_selector("#router-string") + self.assertEqual("subroute/", string.text_content()) + + def test_url_unspecified(self): + self.page.goto(f"{self.live_server_url}/router/unspecified/123/") + path = self.page.wait_for_selector("#router-path") + self.assertIn("/router/unspecified/123/", path.get_attribute("data-path")) + string = self.page.query_selector("#router-string") + self.assertEqual("/router/unspecified//", string.text_content()) + + def test_url_router_integer(self): + self.page.goto(f"{self.live_server_url}/router/integer/123/") + path = self.page.wait_for_selector("#router-path") + self.assertIn("/router/integer/123/", path.get_attribute("data-path")) + string = self.page.query_selector("#router-string") + self.assertEqual("/router/integer//", string.text_content()) + + def test_url_router_path(self): + self.page.goto(f"{self.live_server_url}/router/path/abc/123/") + path = self.page.wait_for_selector("#router-path") + self.assertIn("/router/path/abc/123/", path.get_attribute("data-path")) + string = self.page.query_selector("#router-string") + self.assertEqual("/router/path//", string.text_content()) + + def test_url_router_slug(self): + self.page.goto(f"{self.live_server_url}/router/slug/abc-123/") + path = self.page.wait_for_selector("#router-path") + self.assertIn("/router/slug/abc-123/", path.get_attribute("data-path")) + string = self.page.query_selector("#router-string") + self.assertEqual("/router/slug//", string.text_content()) + + def test_url_router_string(self): + self.page.goto(f"{self.live_server_url}/router/string/abc/") + path = self.page.wait_for_selector("#router-path") + self.assertIn("/router/string/abc/", path.get_attribute("data-path")) + string = self.page.query_selector("#router-string") + self.assertEqual("/router/string//", string.text_content()) + + def test_url_router_uuid(self): + self.page.goto( + f"{self.live_server_url}/router/uuid/123e4567-e89b-12d3-a456-426614174000/" + ) + path = self.page.wait_for_selector("#router-path") + self.assertIn( + "/router/uuid/123e4567-e89b-12d3-a456-426614174000/", + path.get_attribute("data-path"), + ) + string = self.page.query_selector("#router-string") + self.assertEqual("/router/uuid//", string.text_content()) - new_page.goto( - f"{self.live_server_url}/router/any/adslkjgklasdjhfah/6789543256/" - ) - path = new_page.wait_for_selector("#router-path") - self.assertIn( - "/router/any/adslkjgklasdjhfah/6789543256/", - path.get_attribute("data-path"), - ) - string = new_page.query_selector("#router-string") - self.assertEqual("/router/any/", string.text_content()) - - new_page.goto(f"{self.live_server_url}/router/two/123/abc/") - path = new_page.wait_for_selector("#router-path") - self.assertIn("/router/two/123/abc/", path.get_attribute("data-path")) - string = new_page.query_selector("#router-string") - self.assertEqual( - "/router/two///", string.text_content() - ) + def test_url_router_any(self): + self.page.goto( + f"{self.live_server_url}/router/any/adslkjgklasdjhfah/6789543256/" + ) + path = self.page.wait_for_selector("#router-path") + self.assertIn( + "/router/any/adslkjgklasdjhfah/6789543256/", + path.get_attribute("data-path"), + ) + string = self.page.query_selector("#router-string") + self.assertEqual("/router/any/", string.text_content()) - finally: - new_page.close() + def test_url_router_int_and_string(self): + self.page.goto(f"{self.live_server_url}/router/two/123/abc/") + path = self.page.wait_for_selector("#router-path") + self.assertIn("/router/two/123/abc/", path.get_attribute("data-path")) + string = self.page.query_selector("#router-string") + self.assertEqual("/router/two///", string.text_content()) - def test_offline_components(self): - new_page = self.browser.new_page() - try: - server3_url = self.live_server_url.replace( - str(self._port), str(self._port3) - ) - new_page.goto(f"{server3_url}/offline/") - new_page.wait_for_selector("div:not([hidden]) > #online") - self.assertIsNotNone(new_page.query_selector("div[hidden] > #offline")) - self._server_process3.terminate() - self._server_process3.join() - new_page.wait_for_selector("div:not([hidden]) > #offline") - self.assertIsNotNone(new_page.query_selector("div[hidden] > #online")) - finally: - new_page.close() +class ChannelLayersTests(PlaywrightTestCase): + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.page.goto(f"http://{cls.host}:{cls._port}/channel-layers/") def test_channel_layer_components(self): - new_page = self.browser.new_page() - try: - new_page.goto(f"{self.live_server_url}/channel-layers/") - sender = new_page.wait_for_selector("#sender") - sender.type("test", delay=CLICK_DELAY) - sender.press("Enter", delay=CLICK_DELAY) - receiver = new_page.wait_for_selector("#receiver[data-message='test']") - self.assertIsNotNone(receiver) - - sender = new_page.wait_for_selector("#group-sender") - sender.type("1234", delay=CLICK_DELAY) - sender.press("Enter", delay=CLICK_DELAY) - receiver_1 = new_page.wait_for_selector( - "#group-receiver-1[data-message='1234']" - ) - receiver_2 = new_page.wait_for_selector( - "#group-receiver-2[data-message='1234']" - ) - receiver_3 = new_page.wait_for_selector( - "#group-receiver-3[data-message='1234']" - ) - self.assertIsNotNone(receiver_1) - self.assertIsNotNone(receiver_2) - self.assertIsNotNone(receiver_3) - - finally: - new_page.close() - - def test_pyscript_components(self): - new_page = self.browser.new_page() - try: - new_page.goto(f"{self.live_server_url}/pyscript/") - new_page.wait_for_selector("#hello-world-loading") - new_page.wait_for_selector("#hello-world") - new_page.wait_for_selector("#custom-root") - new_page.wait_for_selector("#multifile-parent") - new_page.wait_for_selector("#multifile-child") - - new_page.wait_for_selector("#counter") - new_page.wait_for_selector("#counter pre[data-value='0']") - new_page.wait_for_selector("#counter .plus").click(delay=CLICK_DELAY) - new_page.wait_for_selector("#counter pre[data-value='1']") - new_page.wait_for_selector("#counter .plus").click(delay=CLICK_DELAY) - new_page.wait_for_selector("#counter pre[data-value='2']") - new_page.wait_for_selector("#counter .minus").click(delay=CLICK_DELAY) - new_page.wait_for_selector("#counter pre[data-value='1']") - - new_page.wait_for_selector("#parent") - new_page.wait_for_selector("#child") - new_page.wait_for_selector("#child pre[data-value='0']") - new_page.wait_for_selector("#child .plus").click(delay=CLICK_DELAY) - new_page.wait_for_selector("#child pre[data-value='1']") - new_page.wait_for_selector("#child .plus").click(delay=CLICK_DELAY) - new_page.wait_for_selector("#child pre[data-value='2']") - new_page.wait_for_selector("#child .minus").click(delay=CLICK_DELAY) - new_page.wait_for_selector("#child pre[data-value='1']") - - new_page.wait_for_selector("#parent-toggle") - new_page.wait_for_selector("#parent-toggle button").click(delay=CLICK_DELAY) - new_page.wait_for_selector("#parent-toggle") - new_page.wait_for_selector("#parent-toggle pre[data-value='0']") - new_page.wait_for_selector("#parent-toggle .plus").click(delay=CLICK_DELAY) - new_page.wait_for_selector("#parent-toggle pre[data-value='1']") - new_page.wait_for_selector("#parent-toggle .plus").click(delay=CLICK_DELAY) - new_page.wait_for_selector("#parent-toggle pre[data-value='2']") - new_page.wait_for_selector("#parent-toggle .minus").click(delay=CLICK_DELAY) - new_page.wait_for_selector("#parent-toggle pre[data-value='1']") - - new_page.wait_for_selector("#moment[data-success=true]") - finally: - new_page.close() + sender = self.page.wait_for_selector("#sender") + sender.type("test", delay=CLICK_DELAY) + sender.press("Enter", delay=CLICK_DELAY) + receiver = self.page.wait_for_selector("#receiver[data-message='test']") + self.assertIsNotNone(receiver) + + sender = self.page.wait_for_selector("#group-sender") + sender.type("1234", delay=CLICK_DELAY) + sender.press("Enter", delay=CLICK_DELAY) + receiver_1 = self.page.wait_for_selector( + "#group-receiver-1[data-message='1234']" + ) + receiver_2 = self.page.wait_for_selector( + "#group-receiver-2[data-message='1234']" + ) + receiver_3 = self.page.wait_for_selector( + "#group-receiver-3[data-message='1234']" + ) + self.assertIsNotNone(receiver_1) + self.assertIsNotNone(receiver_2) + self.assertIsNotNone(receiver_3) + + +class PyscriptTests(PlaywrightTestCase): + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.page.goto(f"http://{cls.host}:{cls._port}/pyscript/") + + def test_0_hello_world(self): + self.page.wait_for_selector("#hello-world-loading") + self.page.wait_for_selector("#hello-world") + + def test_custom_root(self): + self.page.wait_for_selector("#custom-root") + + def test_multifile(self): + self.page.wait_for_selector("#multifile-parent") + self.page.wait_for_selector("#multifile-child") + + def test_counter(self): + self.page.wait_for_selector("#counter") + self.page.wait_for_selector("#counter pre[data-value='0']") + self.page.wait_for_selector("#counter .plus").click(delay=CLICK_DELAY) + self.page.wait_for_selector("#counter pre[data-value='1']") + self.page.wait_for_selector("#counter .plus").click(delay=CLICK_DELAY) + self.page.wait_for_selector("#counter pre[data-value='2']") + self.page.wait_for_selector("#counter .minus").click(delay=CLICK_DELAY) + self.page.wait_for_selector("#counter pre[data-value='1']") + + def test_server_side_parent(self): + self.page.wait_for_selector("#parent") + self.page.wait_for_selector("#child") + self.page.wait_for_selector("#child pre[data-value='0']") + self.page.wait_for_selector("#child .plus").click(delay=CLICK_DELAY) + self.page.wait_for_selector("#child pre[data-value='1']") + self.page.wait_for_selector("#child .plus").click(delay=CLICK_DELAY) + self.page.wait_for_selector("#child pre[data-value='2']") + self.page.wait_for_selector("#child .minus").click(delay=CLICK_DELAY) + self.page.wait_for_selector("#child pre[data-value='1']") + + def test_server_side_parent_with_toggle(self): + self.page.wait_for_selector("#parent-toggle") + self.page.wait_for_selector("#parent-toggle button").click(delay=CLICK_DELAY) + self.page.wait_for_selector("#parent-toggle") + self.page.wait_for_selector("#parent-toggle pre[data-value='0']") + self.page.wait_for_selector("#parent-toggle .plus").click(delay=CLICK_DELAY) + self.page.wait_for_selector("#parent-toggle pre[data-value='1']") + self.page.wait_for_selector("#parent-toggle .plus").click(delay=CLICK_DELAY) + self.page.wait_for_selector("#parent-toggle pre[data-value='2']") + self.page.wait_for_selector("#parent-toggle .minus").click(delay=CLICK_DELAY) + self.page.wait_for_selector("#parent-toggle pre[data-value='1']") + + def test_javascript_module_execution_within_pyscript(self): + self.page.wait_for_selector("#moment[data-success=true]") + + +class DistributedComputingTests(PlaywrightTestCase): + + @classmethod + def setUpServer(cls): + super().setUpServer() + cls._server_process2 = cls.ProtocolServerProcess(cls.host, cls.get_application) + cls._server_process2.start() + cls._server_process2.ready.wait() + cls._port2 = cls._server_process2.port.value + + @classmethod + def tearDownServer(cls): + super().tearDownServer() + cls._server_process2.terminate() + cls._server_process2.join() + + def test_host_roundrobin(self): + """Verify if round-robin host selection is working.""" + self.page.goto( + f"{self.live_server_url}/roundrobin/{self._port}/{self._port2}/8" + ) + elem0 = self.page.locator(".custom_host-0") + elem1 = self.page.locator(".custom_host-1") + elem2 = self.page.locator(".custom_host-2") + elem3 = self.page.locator(".custom_host-3") + + elem0.wait_for() + elem1.wait_for() + elem2.wait_for() + elem3.wait_for() + + current_ports = { + elem0.get_attribute("data-port"), + elem1.get_attribute("data-port"), + elem2.get_attribute("data-port"), + elem3.get_attribute("data-port"), + } + correct_ports = { + str(self._port), + str(self._port2), + } + + # There should only be two ports in the set + self.assertEqual(current_ports, correct_ports) + self.assertEqual(len(current_ports), 2) + + def test_custom_host(self): + """Make sure that the component is rendered by a separate server.""" + self.page.goto(f"{self.live_server_url}/port/{self._port2}/") + elem = self.page.locator(".custom_host-0") + elem.wait_for() + self.assertIn( + f"Server Port: {self._port2}", + elem.text_content(), + ) + + def test_custom_host_wrong_port(self): + """Make sure that other ports are not rendering components.""" + tmp_sock = socket.socket() + tmp_sock.bind((self._server_process.host, 0)) + random_port = tmp_sock.getsockname()[1] + self.page.goto(f"{self.live_server_url}/port/{random_port}/") + with self.assertRaises(TimeoutError): + self.page.locator(".custom_host").wait_for(timeout=1000) + + +class OfflineTests(PlaywrightTestCase): + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.page.goto(f"http://{cls.host}:{cls._port}/offline/") + + def test_offline_components(self): + self.page.wait_for_selector("div:not([hidden]) > #online") + self.assertIsNotNone(self.page.query_selector("div[hidden] > #offline")) + self._server_process.terminate() + self._server_process.join() + self.page.wait_for_selector("div:not([hidden]) > #offline") + self.assertIsNotNone(self.page.query_selector("div[hidden] > #online")) diff --git a/tests/test_app/tests/utils.py b/tests/test_app/tests/utils.py new file mode 100644 index 00000000..89d52c35 --- /dev/null +++ b/tests/test_app/tests/utils.py @@ -0,0 +1,93 @@ +import asyncio +import os +import sys +from functools import partial + +from channels.testing import ChannelsLiveServerTestCase +from channels.testing.live import make_application +from django.core.exceptions import ImproperlyConfigured +from django.core.management import call_command +from django.db import connections +from django.test.utils import modify_settings +from playwright.sync_api import sync_playwright + +from reactpy_django.utils import strtobool + +GITHUB_ACTIONS = os.getenv("GITHUB_ACTIONS", "False") + + +class PlaywrightTestCase(ChannelsLiveServerTestCase): + + from reactpy_django import config + + databases = {"default"} + + @classmethod + def setUpClass(cls): + # Repurposed from ChannelsLiveServerTestCase._pre_setup + for connection in connections.all(): + if cls._is_in_memory_db(cls, connection): + raise ImproperlyConfigured( + "ChannelLiveServerTestCase can not be used with in memory databases" + ) + cls._live_server_modified_settings = modify_settings( + ALLOWED_HOSTS={"append": cls.host} + ) + cls._live_server_modified_settings.enable() + cls.get_application = partial( + make_application, + static_wrapper=cls.static_wrapper if cls.serve_static else None, + ) + cls.setUpServer() + + # Open a Playwright browser window + if sys.platform == "win32": + asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy()) + cls.playwright = sync_playwright().start() + headless = strtobool(os.environ.get("PLAYWRIGHT_HEADLESS", GITHUB_ACTIONS)) + cls.browser = cls.playwright.chromium.launch(headless=bool(headless)) + cls.page = cls.browser.new_page() + cls.page.set_default_timeout(5000) + + @classmethod + def setUpServer(cls): + cls._server_process = cls.ProtocolServerProcess(cls.host, cls.get_application) + cls._server_process.start() + cls._server_process.ready.wait() + cls._port = cls._server_process.port.value + + @classmethod + def tearDownClass(cls): + from reactpy_django import config + + # Close the Playwright browser + cls.playwright.stop() + + # Close the other server processes + cls.tearDownServer() + + # Repurposed from ChannelsLiveServerTestCase._post_teardown + cls._live_server_modified_settings.disable() + for db_name in ["default", config.REACTPY_DATABASE]: + call_command( + "flush", + verbosity=0, + interactive=False, + database=db_name, + reset_sequences=False, + ) + + @classmethod + def tearDownServer(cls): + cls._server_process.terminate() + cls._server_process.join() + + def _pre_setup(self): + """Handled manually in `setUpClass` to speed things up.""" + + def _post_teardown(self): + """Handled manually in `tearDownClass` to prevent TransactionTestCase from doing + database flushing. This is needed to prevent a `SynchronousOnlyOperation` from + occuring due to a bug within `ChannelsLiveServerTestCase`.""" + + def setUp(self): ...