Skip to content

Commit

Permalink
Refactor PlaywrightTestCase
Browse files Browse the repository at this point in the history
  • Loading branch information
@Archmonger
Archmonger committed Dec 13, 2024
1 parent c8d6cf9 commit 9f9f99d
Show file tree
Hide file tree
Showing 2 changed files with 102 additions and 72 deletions.
42 changes: 14 additions & 28 deletions tests/test_app/tests/test_components.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ class GenericComponentTests(PlaywrightTestCase):
@classmethod
def setUpClass(cls):
super().setUpClass()
cls.page.goto(f"http://{cls.host}:{cls._port}")
cls.page.goto(f"http://{cls.host}:{cls._port_0}")

def test_hello_world(self):
self.page.wait_for_selector("#hello-world")
Expand Down Expand Up @@ -288,7 +288,7 @@ class PrerenderTests(PlaywrightTestCase):
@classmethod
def setUpClass(cls):
super().setUpClass()
cls.page.goto(f"http://{cls.host}:{cls._port}/prerender/")
cls.page.goto(f"http://{cls.host}:{cls._port_0}/prerender/")

def test_prerender(self):
"""Verify if round-robin host selection is working."""
Expand Down Expand Up @@ -326,7 +326,7 @@ class ErrorTests(PlaywrightTestCase):
@classmethod
def setUpClass(cls):
super().setUpClass()
cls.page.goto(f"http://{cls.host}:{cls._port}/errors/")
cls.page.goto(f"http://{cls.host}:{cls._port_0}/errors/")

def test_component_does_not_exist_error(self):
broken_component = self.page.locator("#component_does_not_exist_error")
Expand Down Expand Up @@ -435,7 +435,7 @@ class ChannelLayersTests(PlaywrightTestCase):
@classmethod
def setUpClass(cls):
super().setUpClass()
cls.page.goto(f"http://{cls.host}:{cls._port}/channel-layers/")
cls.page.goto(f"http://{cls.host}:{cls._port_0}/channel-layers/")

def test_channel_layer_components(self):
sender = self.page.wait_for_selector("#sender")
Expand All @@ -459,7 +459,7 @@ class PyscriptTests(PlaywrightTestCase):
@classmethod
def setUpClass(cls):
super().setUpClass()
cls.page.goto(f"http://{cls.host}:{cls._port}/pyscript/")
cls.page.goto(f"http://{cls.host}:{cls._port_0}/pyscript/")

def test_0_hello_world(self):
self.page.wait_for_selector("#hello-world-loading")
Expand Down Expand Up @@ -510,23 +510,9 @@ def test_1_javascript_module_execution_within_pyscript(self):


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")
self.page.goto(f"{self.live_server_url}/roundrobin/{self._port_0}/{self._port_1}/8")
elem0 = self.page.locator(".custom_host-0")
elem1 = self.page.locator(".custom_host-1")
elem2 = self.page.locator(".custom_host-2")
Expand All @@ -544,8 +530,8 @@ def test_host_roundrobin(self):
elem3.get_attribute("data-port"),
}
correct_ports = {
str(self._port),
str(self._port2),
str(self._port_0),
str(self._port_1),
}

# There should only be two ports in the set
Expand All @@ -554,15 +540,15 @@ def test_host_roundrobin(self):

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}/")
self.page.goto(f"{self.live_server_url}/port/{self._port_1}/")
elem = self.page.locator(".custom_host-0")
elem.wait_for()
assert f"Server Port: {self._port2}" in elem.text_content()
assert f"Server Port: {self._port_1}" in 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))
tmp_sock.bind((self._server_process_0.host, 0))
random_port = tmp_sock.getsockname()[1]
self.page.goto(f"{self.live_server_url}/port/{random_port}/")
with pytest.raises(TimeoutError):
Expand All @@ -573,13 +559,13 @@ class OfflineTests(PlaywrightTestCase):
@classmethod
def setUpClass(cls):
super().setUpClass()
cls.page.goto(f"http://{cls.host}:{cls._port}/offline/")
cls.page.goto(f"http://{cls.host}:{cls._port_0}/offline/")

def test_offline_components(self):
self.page.wait_for_selector("div:not([hidden]) > #online")
assert self.page.query_selector("div[hidden] > #offline") is not None
self._server_process.terminate()
self._server_process.join()
self._server_process_0.terminate()
self._server_process_0.join()
self.page.wait_for_selector("div:not([hidden]) > #offline")
assert self.page.query_selector("div[hidden] > #online") is not None

Expand Down
132 changes: 88 additions & 44 deletions tests/test_app/tests/utils.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,12 +2,12 @@
import asyncio
import os
import sys
from functools import partial
from typing import Callable
from collections.abc import Iterable
from typing import TYPE_CHECKING, Any, Callable

import decorator
from channels.routing import get_default_application
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
Expand All @@ -16,30 +16,97 @@

from reactpy_django.utils import str_to_bool

if TYPE_CHECKING:
from daphne.testing import DaphneProcess

GITHUB_ACTIONS = os.getenv("GITHUB_ACTIONS", "False")


class PlaywrightTestCase(ChannelsLiveServerTestCase):
from reactpy_django import config

databases = {"default"}

total_servers = 3
_server_process_0: "DaphneProcess"
_server_process_1: "DaphneProcess"
_server_process_2: "DaphneProcess"
_server_process_3: "DaphneProcess"
_port_0: int
_port_1: int
_port_2: int
_port_3: int

####################################################
# Overrides for ChannelsLiveServerTestCase methods #
####################################################
@classmethod
def setUpClass(cls):
# Repurposed from ChannelsLiveServerTestCase._pre_setup
for connection in connections.all():
if cls._is_in_memory_db(cls, connection):
if connection.vendor == "sqlite" and connection.is_in_memory_db():
msg = "ChannelLiveServerTestCase can not be used with in memory databases"
raise ImproperlyConfigured(msg)
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()
cls.get_application = get_default_application

# Start the Django webserver(s)
for i in range(cls.total_servers):
cls.start_django_webserver(i)

# Wipe the databases
from reactpy_django import config

cls.flush_databases({"default", config.REACTPY_DATABASE})

# Open a Playwright browser window
cls.start_playwright_client()

@classmethod
def tearDownClass(cls):
# Close the Playwright browser
cls.shutdown_playwright_client()

# Shutdown the Django webserver
for i in range(cls.total_servers):
cls.shutdown_django_webserver(i)
cls._live_server_modified_settings.disable()

# Wipe the databases
from reactpy_django import config

cls.flush_databases({"default", config.REACTPY_DATABASE})

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 in between tests. This also fixes a `SynchronousOnlyOperation` caused
by a bug within `ChannelsLiveServerTestCase`."""

@property
def live_server_url(self):
"""Provides the URL to the FIRST SPAWNED Django webserver."""
return f"http://{self.host}:{self._port_0}"

#########################
# Custom helper methods #
#########################
@classmethod
def start_django_webserver(cls, num=0):
setattr(cls, f"_server_process_{num}", cls.ProtocolServerProcess(cls.host, cls.get_application))
server_process: DaphneProcess = getattr(cls, f"_server_process_{num}")
server_process.start()
server_process.ready.wait()
setattr(cls, f"_port_{num}", server_process.port.value)

@classmethod
def shutdown_django_webserver(cls, num=0):
server_process: DaphneProcess = getattr(cls, f"_server_process_{num}")
server_process.terminate()
server_process.join()

@classmethod
def start_playwright_client(cls):
if sys.platform == "win32":
asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())
cls.playwright = sync_playwright().start()
Expand All @@ -49,26 +116,13 @@ def setUpClass(cls):
cls.page.set_default_timeout(10000)

@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
def shutdown_playwright_client(cls):
cls.browser.close()
cls.playwright.stop()

# Close the other server processes
cls.tearDownServer()

# Repurposed from ChannelsLiveServerTestCase._post_teardown
cls._live_server_modified_settings.disable()
# Using set to prevent duplicates
for db_name in {"default", config.REACTPY_DATABASE}: # noqa: PLC0208
@staticmethod
def flush_databases(db_names: Iterable[Any]):
for db_name in db_names:
call_command(
"flush",
verbosity=0,
Expand All @@ -77,26 +131,16 @@ def tearDownClass(cls):
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
occurring due to a bug within `ChannelsLiveServerTestCase`."""

def navigate_to_page(path: str, *, server_num=0):
"""Decorator to make sure the browser is on a specific page before running a test."""

def navigate_to_page(path: str):
def _decorator(func: Callable):
@decorator.decorator
def _wrapper(func: Callable, self: PlaywrightTestCase, *args, **kwargs):
if self.page.url != path:
self.page.goto(f"http://{self.host}:{self._port}/{path.lstrip('/')}")
_port = getattr(self, f"_port_{server_num}")
self.page.goto(f"http://{self.host}:{_port}/{path.lstrip('/')}")
return func(self, *args, **kwargs)

return _wrapper(func)
Expand Down

0 comments on commit 9f9f99d

Please sign in to comment.