mirror/rio
1
0
Fork 0
mirror of https://github.com/rio-labs/rio.git synced 2026-05-02 17:09:18 -05:00
Code Issues Packages Projects Releases Wiki Activity

fix unit tests

Browse Source
This commit is contained in:
Aran-Fey
2024-05-20 00:01:39 +02:00
parent ce341c8049
commit 82d1fd5f64
22 changed files with 624 additions and 517 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rio/cli/run_project/uvicorn_worker.py
-14
View File
@@ -74,20 +74,6 @@ class UvicornWorker:
name="uvicorn serve",
)
# Uvicorn doesn't handle CancelledError properly, which results in ugly
# output in the console. This monkeypatch suppresses that.
original_receive = uvicorn.lifespan.on.LifespanOn.receive
async def patched_receive(self) -> Any:
try:
return await original_receive(self)
except asyncio.CancelledError:
return {
"type": "lifespan.shutdown",
}
uvicorn.lifespan.on.LifespanOn.receive = patched_receive
# Run the server
try:
await asyncio.shield(serve_task)
rio/components/button.py
+5 -5
View File
@@ -101,7 +101,7 @@ class Button(Component):
return rio.Button(
rio.Column(
rio.Text("Click repeatedly to fill up the progress bar"),
rio.ProgressBar(self.clicks/10, width=15, height=1),
rio.ProgressBar(self.clicks / 10, width=15, height=1),
),
on_press=self._on_button_press,
)
@@ -273,9 +273,9 @@ class IconButton(Component):
)
```
`IconButton`s are commonly used to trigger actions. You can easily achieve this by
adding a function call to `on_press`. You can use a function call to update
the banner text signaling that the button was pressed:
`IconButton`s are commonly used to trigger actions. You can easily achieve
this by adding a function call to `on_press`. You can use a function call to
update the banner text signaling that the button was pressed:
```python
class MyComponent(rio.Component):
@@ -288,7 +288,7 @@ class IconButton(Component):
return rio.Column(
rio.Banner(
text=self.banner_text,
style="material/info",
style="info",
),
rio.IconButton(
icon="material/castle",
rio/components/color_picker.py
+3 -3
View File
@@ -63,7 +63,7 @@ class ColorPicker(FundamentalComponent):
```python
def print_selected_color(event: rio.ColorChangeEvent):
print('You selected the color', event.color)
print("You selected the color", event.color)
rio.ColorPicker(
rio.Color.from_hex("#ff0000"),
@@ -84,9 +84,9 @@ class ColorPicker(FundamentalComponent):
color=self.bind().color,
),
rio.Icon(
'material/star',
"material/star",
fill=self.color,
)
),
)
```
"""
rio/components/list_items.py
+7 -5
View File
@@ -270,13 +270,14 @@ class CustomListItem(FundamentalComponent):
to the list item. You can add any component to the list item.
```python
import functools
class MyCustomListItemComponent(rio.Component):
# create a custom list item
# Create a custom list item
product: str
button_text: str
def build(self) -> rio.Component:
return rio.Row(
rio.Text(self.product),
rio.Spacer(),
@@ -300,13 +301,14 @@ class CustomListItem(FundamentalComponent):
for product in self.products:
list_items.append(
rio.CustomListItem(
# Use the `MyCustomListItem` component to create a custom list item
# Use the `MyCustomListItem` component to create a
# custom list item
content=MyCustomListItemComponent(
product=product, button_text="Click Me!"
),
key=product,
# Note the use of `functools.partial` to pass the product
# to the event handler.
# Note the use of `functools.partial` to pass the
# product to the event handler.
on_press=functools.partial(
self.on_press_heading_list_item,
product=product,
rio/components/scroll_target.py
+2 -2
View File
@@ -39,13 +39,13 @@ class ScrollTarget(FundamentalComponent):
```python
rio.ScrollTarget(
id="chapter-1",
content=rio.Text('Chapter 1', style='heading1'),
content=rio.Text("Chapter 1", style="heading1"),
)
```
## Metadata
experimental: True
`experimental`: True
"""
id: str
rio/components/theme_context_switcher.py
+3 -2
View File
@@ -25,6 +25,7 @@ class ThemeContextSwitcher(FundamentalComponent):
You can find more details on theming on the [theming how-to
page](https://rio.dev/docs/howto/theming).
## Attributes
`content`: The currently displayed component.
@@ -34,12 +35,12 @@ class ThemeContextSwitcher(FundamentalComponent):
## Examples
A minimal example of a `ThemeContextSwitcher` will be shown:
A minimal example of a `ThemeContextSwitcher`:
```python
rio.ThemeContextSwitcher(
content=rio.Button("Button"),
color="secondary"
color="secondary",
)
```
"""
rio/session.py
+8
View File
@@ -185,6 +185,11 @@ class Session(unicall.Unicall):
# were last saved.
self._last_settings_save_time: float = -float("inf")
# A dict of {build_function: BuildFailedComponent}. This is cleared at
# the start of every refresh, and tracks which build functions failed.
# Used for unit testing.
self._crashed_build_functions = dict[Callable, str]()
# Weak dictionaries to hold additional information about components.
# These are split in two to avoid the dictionaries keeping the
# components alive. Notice how both dictionaries are weak on the actual
@@ -873,6 +878,9 @@ window.history.{method}(null, "", {json.dumps(str(active_page_url))})
# For why this lock is here see its creation in `__init__`
async with self._refresh_lock:
# Clear the dict of crashed build functions
self._crashed_build_functions.clear()
while self._dirty_components:
# Refresh and get a set of all components which have been visited
(
rio/testing.py
+231
View File
@@ -0,0 +1,231 @@
import asyncio
import json
import types
from collections.abc import Callable, Iterable, Iterator, Mapping
from dataclasses import KW_ONLY, dataclass, field
from typing_extensions import Any, Self, TypeVar
import ordered_set
from uniserde import Jsonable, JsonDoc
import rio
from rio.app_server import AppServer
from rio.components.root_components import (
FundamentalRootComponent,
HighLevelRootComponent,
)
__all__ = ["TestClient"]
T = TypeVar("T")
C = TypeVar("C", bound=rio.Component)
@dataclass(repr=False, eq=False, match_args=False)
class TestClient:
build: Callable[[], rio.Component] = field(default=lambda: rio.Text("hi"))
_: KW_ONLY
app_name: str = "mock-app"
running_in_window: bool = False
user_settings: JsonDoc = field(default_factory=dict)
default_attachments: Iterable[object] = ()
use_ordered_dirty_set: bool = False
def __post_init__(self):
self._session: rio.Session | None = None
self._server_task: asyncio.Task | None = None
self._outgoing_messages = list[JsonDoc]()
self._responses = asyncio.Queue[JsonDoc]()
self._responses.put_nowait(
{
"websiteUrl": "https://unit.test",
"preferredLanguages": [],
"userSettings": self.user_settings,
"windowWidth": 1920,
"windowHeight": 1080,
"timezone": "America/New_York",
"decimalSeparator": ".",
"thousandsSeparator": ",",
"prefersLightTheme": True,
}
)
self._first_refresh_completed = asyncio.Event()
async def _send_message(self, message_text: str) -> None:
message = json.loads(message_text)
self._outgoing_messages.append(message)
if "id" in message:
self._responses.put_nowait(
{
"jsonrpc": "2.0",
"id": message["id"],
"result": None,
}
)
if message["method"] == "updateComponentStates":
self._first_refresh_completed.set()
async def _receive_message(self) -> Jsonable:
return await self._responses.get()
async def __aenter__(self) -> Self:
app = rio.App(
build=self.build,
name=self.app_name,
default_attachments=tuple(self.default_attachments),
)
app_server = AppServer(
app,
debug_mode=False,
running_in_window=self.running_in_window,
validator_factory=None,
internal_on_app_start=None,
)
# Emulate the process of creating a session as closely as possible
fake_request: Any = types.SimpleNamespace(
url="https://unit.test",
base_url="https://unit.test",
headers={"accept": "text/html"},
client=types.SimpleNamespace(host="localhost", port="12345"),
)
await app_server._serve_index(fake_request, "")
[[session_token, session]] = app_server._active_session_tokens.items()
self._session = session
if self.use_ordered_dirty_set:
session._dirty_components = ordered_set.OrderedSet(
session._dirty_components
) # type: ignore
fake_websocket: Any = types.SimpleNamespace(
client="1.2.3.4",
accept=lambda: _make_awaitable(),
send_text=self._send_message,
receive_json=self._receive_message,
)
test_task = asyncio.current_task()
assert test_task is not None
async def serve_websocket():
try:
await app_server._serve_websocket(fake_websocket, session_token)
except asyncio.CancelledError:
pass
except Exception as error:
test_task.cancel(
f"Exception in AppServer._serve_websocket: {error}"
)
else:
test_task.cancel(
"AppServer._serve_websocket exited unexpectedly. An exception"
" must have occurred in the `init_coro`."
)
self._server_task = asyncio.create_task(serve_websocket())
await self._first_refresh_completed.wait()
return self
async def __aexit__(self, *_) -> None:
if self._server_task is not None:
self._server_task.cancel()
@property
def _dirty_components(self) -> set[rio.Component]:
return set(self.session._dirty_components)
@property
def _last_updated_components(self) -> set[rio.Component]:
return set(self._last_component_state_changes)
@property
def _last_component_state_changes(
self,
) -> Mapping[rio.Component, Mapping[str, object]]:
for message in reversed(self._outgoing_messages):
if message["method"] == "updateComponentStates":
delta_states: dict = message["params"]["deltaStates"] # type: ignore
return {
self.session._weak_components_by_id[
int(component_id)
]: delta
for component_id, delta in delta_states.items()
if int(component_id) != self.session._root_component._id
}
return {}
def _get_build_output(
self,
component: rio.Component,
type_: type[C] | None = None,
) -> C:
result = self.session._weak_component_data_by_component[
component
].build_result
if type_ is not None:
assert (
type(result) is type_
), f"Expected {type_}, got {type(result)}"
return result # type: ignore
@property
def session(self) -> rio.Session:
assert self._session is not None
return self._session
@property
def crashed_build_functions(self) -> Mapping[Callable, str]:
return self.session._crashed_build_functions
@property
def root_component(self) -> rio.Component:
high_level_root = self.session._root_component
assert isinstance(
high_level_root, HighLevelRootComponent
), high_level_root
low_level_root = self.session._weak_component_data_by_component[
high_level_root
].build_result
assert isinstance(
low_level_root, FundamentalRootComponent
), low_level_root
scroll_container = low_level_root.content
assert isinstance(
scroll_container, rio.ScrollContainer
), scroll_container
return scroll_container.content
def get_components(self, component_type: type[C]) -> Iterator[C]:
root_component = self.root_component
for component in root_component._iter_component_tree():
if type(component) is component_type:
yield component # type: ignore
def get_component(self, component_type: type[C]) -> C:
try:
return next(self.get_components(component_type))
except StopIteration:
raise AssertionError(f"No component of type {component_type} found")
async def refresh(self) -> None:
await self.session._refresh()
async def _make_awaitable(value: T = None) -> T:
return value
rio/user_settings_module.py
+15 -1
View File
@@ -7,7 +7,7 @@ from typing import * # type: ignore
import uniserde
from typing_extensions import Self
from .dataclass import RioDataclassMeta
from .dataclass import RioDataclassMeta, all_class_fields
from . import inspection, session
__all__ = [
@@ -161,3 +161,17 @@ class UserSettings(metaclass=RioDataclassMeta):
if not TYPE_CHECKING:
__setattr__ = __setattr
def _equals(self, other: Self) -> bool:
if type(self) != type(other):
return False
fields_to_compare = (
all_class_fields(type(self)).keys()
- all_class_fields(__class__).keys()
)
for field_name in fields_to_compare:
if getattr(self, field_name) != getattr(other, field_name):
return False
return True
rio/utils.py
+15 -6
View File
@@ -333,10 +333,15 @@ def safe_build(build_function: Callable[[], rio.Component]) -> rio.Component:
# Screw circular imports
from rio.components.build_failed import BuildFailed
return BuildFailed(f"`{build_function_repr}` has crashed", repr(err))
build_failed_component = BuildFailed(
f"`{build_function_repr}` has crashed", repr(err)
)
else:
# Make sure the result meets expectations
if isinstance(build_result, rio.Component): # type: ignore (unnecessary isinstance)
# All is well
return build_result
# Make sure the result meets expectations
if not isinstance(build_result, rio.Component): # type: ignore (unnecessary isinstance)
build_function_repr = _repr_build_function(build_function)
rio._logger.error(
@@ -347,10 +352,14 @@ def safe_build(build_function: Callable[[], rio.Component]) -> rio.Component:
# Screw circular imports
from rio.components.build_failed import BuildFailed
return BuildFailed(
build_failed_component = BuildFailed(
f"`{build_function_repr}` has returned an invalid result",
f"Build functions must return instances of `rio.Component`, but the result was {build_result!r}",
)
# All is well
return build_result
# Save the error in the session, for testing purposes
build_failed_component.session._crashed_build_functions[build_function] = (
build_failed_component.error_details
)
return build_failed_component
tests/test_app_build.py
+8 -7
View File
@@ -1,14 +1,15 @@
from utils import create_mockapp
import rio
import rio.testing
async def test_fundamental_container_as_root():
def build():
return rio.Row(rio.Text("Hello"))
async with create_mockapp(build) as app:
row_component = app.get_component(rio.Row)
text_component = app.get_component(rio.Text)
async with rio.testing.TestClient(build) as test_client:
row_component = test_client.get_component(rio.Row)
text_component = test_client.get_component(rio.Text)
assert {row_component, text_component} <= app.last_updated_components
assert {
row_component,
text_component,
} <= test_client._last_updated_components
tests/test_attribute_bindings.py
+95 -61
View File
@@ -1,8 +1,6 @@
from typing import cast
from utils import create_mockapp
import rio
import rio.testing
from rio.state_properties import PleaseTurnThisIntoAnAttributeBinding
@@ -43,9 +41,9 @@ async def test_bindings_arent_created_too_early():
def build(self) -> rio.Component:
return IHaveACustomInit(text=self.bind().text)
async with create_mockapp(Container) as app:
root_component = app.get_component(Container)
child_component = app.get_component(IHaveACustomInit)
async with rio.testing.TestClient(Container) as test_client:
root_component = test_client.get_component(Container)
child_component = test_client.get_component(IHaveACustomInit)
assert child_component.text == "hi"
@@ -78,34 +76,40 @@ async def test_init_receives_attribute_bindings_as_input():
def build(self) -> rio.Component:
return Square(self.bind().size)
async with create_mockapp(lambda: Container(7)):
async with rio.testing.TestClient(lambda: Container(7)):
assert isinstance(size_value, PleaseTurnThisIntoAnAttributeBinding)
async def test_binding_assignment_on_child():
async with create_mockapp(Parent) as app:
root_component = app.get_component(Parent)
text_component = app.get_build_output(root_component, rio.Text)
async with rio.testing.TestClient(Parent) as test_client:
root_component = test_client.get_component(Parent)
text_component = test_client._get_build_output(root_component, rio.Text)
assert not app.dirty_components
assert not test_client._dirty_components
text_component.text = "Hello"
assert app.dirty_components == {root_component, text_component}
assert test_client._dirty_components == {
root_component,
text_component,
}
assert root_component.text == "Hello"
assert text_component.text == "Hello"
async def test_binding_assignment_on_parent():
async with create_mockapp(Parent) as app:
root_component = app.get_component(Parent)
text_component = app.get_build_output(root_component)
async with rio.testing.TestClient(Parent) as test_client:
root_component = test_client.get_component(Parent)
text_component = test_client._get_build_output(root_component)
assert not app.dirty_components
assert not test_client._dirty_components
root_component.text = "Hello"
assert app.dirty_components == {root_component, text_component}
assert test_client._dirty_components == {
root_component,
text_component,
}
assert root_component.text == "Hello"
assert text_component.text == "Hello"
@@ -120,85 +124,103 @@ async def test_binding_assignment_on_sibling():
rio.Text(self.bind().text),
)
async with create_mockapp(Root) as app:
root_component = app.get_component(Root)
async with rio.testing.TestClient(Root) as test_client:
root_component = test_client.get_component(Root)
text1, text2 = cast(
list[rio.Text],
app.get_build_output(root_component, rio.Column).children,
test_client._get_build_output(root_component, rio.Column).children,
)
assert not app.dirty_components
assert not test_client._dirty_components
text1.text = "Hello"
assert app.dirty_components == {root_component, text1, text2}
assert test_client._dirty_components == {
root_component,
text1,
text2,
}
assert root_component.text == "Hello"
assert text1.text == "Hello"
assert text2.text == "Hello"
async def test_binding_assignment_on_grandchild():
async with create_mockapp(Grandparent) as app:
root_component = app.get_component(Grandparent)
parent = cast(Parent, app.get_build_output(root_component))
text_component: rio.Text = app.get_build_output(parent)
async with rio.testing.TestClient(Grandparent) as test_client:
root_component = test_client.get_component(Grandparent)
parent = cast(Parent, test_client._get_build_output(root_component))
text_component: rio.Text = test_client._get_build_output(parent)
assert not app.dirty_components
assert not test_client._dirty_components
text_component.text = "Hello"
assert app.dirty_components == {root_component, parent, text_component}
assert test_client._dirty_components == {
root_component,
parent,
text_component,
}
assert root_component.text == "Hello"
assert parent.text == "Hello"
assert text_component.text == "Hello"
async def test_binding_assignment_on_middle():
async with create_mockapp(Grandparent) as app:
root_component = app.get_component(Grandparent)
parent: Parent = app.get_build_output(root_component)
text_component: rio.Text = app.get_build_output(parent)
async with rio.testing.TestClient(Grandparent) as test_client:
root_component = test_client.get_component(Grandparent)
parent: Parent = test_client._get_build_output(root_component)
text_component: rio.Text = test_client._get_build_output(parent)
assert not app.dirty_components
assert not test_client._dirty_components
parent.text = "Hello"
assert app.dirty_components == {root_component, parent, text_component}
assert test_client._dirty_components == {
root_component,
parent,
text_component,
}
assert root_component.text == "Hello"
assert parent.text == "Hello"
assert text_component.text == "Hello"
async def test_binding_assignment_on_child_after_reconciliation():
async with create_mockapp(Parent) as app:
root_component = app.get_component(Parent)
text_component: rio.Text = app.get_build_output(root_component)
async with rio.testing.TestClient(Parent) as test_client:
root_component = test_client.get_component(Parent)
text_component: rio.Text = test_client._get_build_output(root_component)
assert not app.dirty_components
assert not test_client._dirty_components
# Rebuild the root component, which reconciles the child
await root_component.force_refresh()
text_component.text = "Hello"
assert app.dirty_components == {root_component, text_component}
assert test_client._dirty_components == {
root_component,
text_component,
}
assert root_component.text == "Hello"
assert text_component.text == "Hello"
async def test_binding_assignment_on_parent_after_reconciliation():
async with create_mockapp(Parent) as app:
root_component = app.get_component(Parent)
text_component: rio.Text = app.get_build_output(root_component)
async with rio.testing.TestClient(Parent) as test_client:
root_component = test_client.get_component(Parent)
text_component: rio.Text = test_client._get_build_output(root_component)
assert not app.dirty_components
assert not test_client._dirty_components
# Rebuild the root component, which reconciles the child
await root_component.force_refresh()
root_component.text = "Hello"
assert app.dirty_components == {root_component, text_component}
assert test_client._dirty_components == {
root_component,
text_component,
}
assert root_component.text == "Hello"
assert text_component.text == "Hello"
@@ -213,56 +235,68 @@ async def test_binding_assignment_on_sibling_after_reconciliation():
rio.Text(self.bind().text),
)
async with create_mockapp(Root) as app:
root_component = app.get_component(Root)
text1, text2 = app.get_build_output(root_component).children
async with rio.testing.TestClient(Root) as test_client:
root_component = test_client.get_component(Root)
text1, text2 = test_client._get_build_output(root_component).children
assert not app.dirty_components
assert not test_client._dirty_components
# Rebuild the root component, which reconciles the children
await root_component.force_refresh()
text1.text = "Hello"
assert app.dirty_components == {root_component, text1, text2}
assert test_client._dirty_components == {
root_component,
text1,
text2,
}
assert root_component.text == "Hello"
assert text1.text == "Hello"
assert text2.text == "Hello"
async def test_binding_assignment_on_grandchild_after_reconciliation():
async with create_mockapp(Grandparent) as app:
root_component = app.get_component(Grandparent)
parent: Parent = app.get_build_output(root_component)
text_component: rio.Text = app.get_build_output(parent)
async with rio.testing.TestClient(Grandparent) as test_client:
root_component = test_client.get_component(Grandparent)
parent: Parent = test_client._get_build_output(root_component)
text_component: rio.Text = test_client._get_build_output(parent)
assert not app.dirty_components
assert not test_client._dirty_components
# Rebuild the root component, which reconciles the child
await root_component.force_refresh()
text_component.text = "Hello"
assert app.dirty_components == {root_component, parent, text_component}
assert test_client._dirty_components == {
root_component,
parent,
text_component,
}
assert root_component.text == "Hello"
assert parent.text == "Hello"
assert text_component.text == "Hello"
async def test_binding_assignment_on_middle_after_reconciliation():
async with create_mockapp(Grandparent) as app:
root_component = app.get_component(Grandparent)
parent: Parent = app.get_build_output(root_component)
text_component: rio.Text = app.get_build_output(parent)
async with rio.testing.TestClient(Grandparent) as test_client:
root_component = test_client.get_component(Grandparent)
parent: Parent = test_client._get_build_output(root_component)
text_component: rio.Text = test_client._get_build_output(parent)
assert not app.dirty_components
assert not test_client._dirty_components
# Rebuild the root component, which reconciles the child
await root_component.force_refresh()
parent.text = "Hello"
assert app.dirty_components == {root_component, parent, text_component}
assert test_client._dirty_components == {
root_component,
parent,
text_component,
}
assert root_component.text == "Hello"
assert parent.text == "Hello"
assert text_component.text == "Hello"
tests/test_custom_components.py
+4 -4
View File
@@ -1,8 +1,8 @@
import dataclasses
from utils import create_mockapp, enable_component_instantiation
from utils import enable_component_instantiation
import rio
import rio.testing
@enable_component_instantiation
@@ -29,6 +29,6 @@ async def test_post_init():
def build(self) -> rio.Component:
return rio.Text("hi")
async with create_mockapp(TestComponent) as app:
root_component = app.get_component(TestComponent)
async with rio.testing.TestClient(TestComponent) as test_client:
root_component = test_client.get_component(TestComponent)
assert root_component.post_init_called
tests/test_docstring_code_blocks.py
+62 -48
View File
@@ -1,81 +1,83 @@
import re
import subprocess
import tempfile
import textwrap
from typing import * # type: ignore
import black
import imy.docstrings
import pyright
import pytest
import rio
import rio.docs
CODE_BLOCK_PATTERN = re.compile(r"```.*?\n(.*?)\n```", re.DOTALL)
CODE_BLOCK_PATTERN = re.compile(r"```(.*?)```", re.DOTALL)
def all_components() -> list[Type[rio.Component]]:
"""
Iterates over all components that ship with Rio.
"""
to_do: Iterable[Type[rio.Component]] = [rio.Component]
result: list[Type[rio.Component]] = []
while to_do:
component = to_do.pop()
result.append(component)
to_do.extend(component.__subclasses__())
return result
all_documented_objects = [
obj for obj, _ in rio.docs.find_documented_objects(False)
]
all_documented_objects.sort(key=lambda obj: obj.__name__)
def get_code_blocks(comp: Type[rio.Component]) -> list[str]:
def get_code_blocks(obj: type | Callable) -> list[str]:
"""
Returns a list of all code blocks in the docstring of a component.
"""
docs = imy.docstrings.ClassDocs.from_class(comp)
docstring = obj.__doc__
# No docs?
if docs.details is None:
if not docstring:
return []
docstring = textwrap.dedent(docstring)
# Find any contained code blocks
result: list[str] = []
for match in CODE_BLOCK_PATTERN.finditer(docs.details):
block: str = match.group(0)
assert block.startswith("```")
assert block.endswith("```")
for match in CODE_BLOCK_PATTERN.finditer(docstring):
block: str = match.group(1)
# Split into language and source
linebreak = block.index("\n")
linebreak = block.find("\n")
assert linebreak != -1
first_line = block[3:linebreak].strip()
block = block[linebreak + 1 : -3]
language = block[:linebreak]
block = block[linebreak + 1 :]
# Make sure a language is specified
assert first_line, "The code block has no language specified"
assert language, "The code block has no language specified"
result.append(block)
return result
@pytest.mark.parametrize("comp", all_components())
def test_eval_code_block(comp: Type[rio.Component]) -> None:
@pytest.mark.parametrize("obj", all_documented_objects)
def test_eval_code_block(obj: type | Callable) -> None:
# Eval all code blocks and make sure they don't crash
for source in get_code_blocks(comp):
for source in get_code_blocks(obj):
compile(source, "<string>", "exec")
@pytest.mark.parametrize("comp", all_components())
def test_code_block_is_formatted(comp: Type[rio.Component]) -> None:
@pytest.mark.parametrize("obj", all_documented_objects)
def test_code_block_is_formatted(obj: type | Callable) -> None:
# Make sure all code blocks are formatted according to black
for source in get_code_blocks(comp):
for source in get_code_blocks(obj):
formatted_source = black.format_str(source, mode=black.FileMode())
# Black often inserts 2 empty lines between stuff, but that's really not
# necessary in docstrings. So we'll collapse those into a single empty
# line.
source = source.replace("\n\n\n", "\n\n")
formatted_source = formatted_source.replace("\n\n\n", "\n\n")
assert source == formatted_source
def _pyright_check_source(source: str) -> tuple[int, int]:
PYRIGHT_ERROR_OR_WARNING_REGEX = re.compile(
r".*\.py:\d+:\d+ - (?:error|warning): (.*)"
)
def _find_static_typing_errors(source: str) -> str:
"""
Run pyright on the given source and return the number of errors and
warnings.
@@ -97,19 +99,31 @@ def _pyright_check_source(source: str) -> tuple[int, int]:
result_out = result_out.decode()
# Find the number of errors and warnings
match = re.search(
r"(\d+) error(s)?, (\d+) warning(s)?, (\d+) information",
result_out,
)
assert match is not None, result_out
return int(match.group(1)), int(match.group(3))
lines = list[str]()
for line in result_out.splitlines():
match = PYRIGHT_ERROR_OR_WARNING_REGEX.match(line)
if match:
lines.append(match.group(1))
return "\n".join(lines)
@pytest.mark.parametrize("comp", all_components())
def test_analyze_code_block(comp: Type[rio.Component]) -> None:
@pytest.mark.parametrize("obj", all_documented_objects)
def test_analyze_code_block(obj: type | Callable) -> None:
# A lot of snippets are missing context, so it's only natural that pyright
# will find issues with the code. There isn't really anything we can do
# about it, so we'll just skip those object.
if obj in (rio.App, rio.Color, rio.UserSettings):
pytest.xfail()
# rio.Plot fails due to a weird pyright bug. It says "import cannot be
# resolved" for some reason.
if obj is rio.Plot:
pytest.xfail()
# Make sure pyright is happy with all code blocks
for source in get_code_blocks(comp):
n_errors, n_warnings = _pyright_check_source(source)
assert n_errors == 0, f"Found {n_errors} errors"
assert n_warnings == 0, f"Found {n_warnings} warnings"
for source in get_code_blocks(obj):
errors = _find_static_typing_errors(source)
assert not errors, errors
tests/test_events.py
+9 -11
View File
@@ -1,8 +1,6 @@
import asyncio
from utils import create_mockapp
import rio
import rio.testing
class ChildToggler(rio.Component):
@@ -39,18 +37,18 @@ async def test_mounted():
def build():
return ChildToggler(DemoComponent())
async with create_mockapp(build) as app:
root = app.get_component(ChildToggler)
async with rio.testing.TestClient(build) as test_client:
root = test_client.get_component(ChildToggler)
assert not mounted
assert not unmounted
root.toggle()
await app.refresh()
await test_client.refresh()
assert mounted
assert not unmounted
root.toggle()
await app.refresh()
await test_client.refresh()
assert unmounted
@@ -65,15 +63,15 @@ async def test_refresh_after_synchronous_mount_handler():
def build(self) -> rio.Component:
return rio.Switch(self.mounted)
async with create_mockapp(DemoComponent) as app:
demo_component = app.get_component(DemoComponent)
switch = app.get_component(rio.Switch)
async with rio.testing.TestClient(DemoComponent) as test_client:
demo_component = test_client.get_component(DemoComponent)
switch = test_client.get_component(rio.Switch)
# TODO: I don't know how we can wait for the refresh, so I'll just use a
# sleep()
await asyncio.sleep(0.5)
assert demo_component.mounted
last_component_state_changes = app.last_component_state_changes
last_component_state_changes = test_client._last_component_state_changes
assert switch in last_component_state_changes
assert last_component_state_changes[switch].get("is_on") is True
tests/test_reconciliation.py
+70 -63
View File
@@ -1,6 +1,4 @@
from utils import create_mockapp
import rio
import rio.testing
async def test_default_values_arent_considered_explicitly_set():
@@ -22,9 +20,11 @@ async def test_default_values_arent_considered_explicitly_set():
square_component = SquareComponent(self.text, size=10)
return rio.Container(square_component)
async with create_mockapp(lambda: RootComponent("Hello")) as app:
root_component = app.get_component(RootComponent)
square_component = app.get_component(SquareComponent)
async with rio.testing.TestClient(
lambda: RootComponent("Hello")
) as test_client:
root_component = test_client.get_component(RootComponent)
square_component = test_client.get_component(SquareComponent)
# Create a new SquareComponent with the default size. Since we aren't
# explicitly passing a size to the constructor, reconciliation should
@@ -41,20 +41,21 @@ async def test_reconcile_same_component_instance():
def build():
return rio.Container(rio.Text("Hello"))
async with create_mockapp(build) as app:
app.outgoing_messages.clear()
async with rio.testing.TestClient(build) as test_client:
test_client._outgoing_messages.clear()
root_component = app.get_component(rio.Container)
root_component = test_client.get_component(rio.Container)
await root_component.force_refresh()
# Nothing changed, so there's no need to send any data to JS. But in
# order to know that nothing changed, the framework would have to track
# every individual attribute of every component. Since we forced the
# root_component to refresh, it's reasonable to send that component's data to
# JS.
assert not app.outgoing_messages or app.last_updated_components == {
root_component
}
# root_component to refresh, it's reasonable to send that component's
# data to JS.
assert (
not test_client._outgoing_messages
or test_client._last_updated_components == {root_component}
)
async def test_reconcile_not_dirty_high_level_component():
@@ -85,14 +86,14 @@ async def test_reconcile_not_dirty_high_level_component():
def build(self):
return self.content
async with create_mockapp(HighLevelComponent1) as app:
root_component = app.get_component(HighLevelComponent1)
async with rio.testing.TestClient(HighLevelComponent1) as test_client:
root_component = test_client.get_component(HighLevelComponent1)
root_component.switch = True
await app.refresh()
await test_client.refresh()
assert any(
isinstance(component, rio.Switch)
for component in app.last_updated_components
for component in test_client._last_updated_components
)
@@ -115,8 +116,8 @@ async def test_reconcile_unusual_types():
def build(self):
return rio.Text(self.text)
async with create_mockapp(Container) as app:
root_component = app.get_component(Container)
async with rio.testing.TestClient(Container) as test_client:
root_component = test_client.get_component(Container)
# As long as this doesn't crash, it's fine
await root_component.force_refresh()
@@ -132,12 +133,12 @@ async def test_reconcile_by_key():
else:
return rio.Container(rio.Text("World", key="foo"))
async with create_mockapp(Toggler) as app:
root_component = app.get_component(Toggler)
text = app.get_component(rio.Text)
async with rio.testing.TestClient(Toggler) as test_client:
root_component = test_client.get_component(Toggler)
text = test_client.get_component(rio.Text)
root_component.toggle = True
await app.refresh()
await test_client.refresh()
assert text.text == "World"
@@ -152,12 +153,12 @@ async def test_key_prevents_structural_match():
else:
return rio.Text("World", key="foo")
async with create_mockapp(Toggler) as app:
root_component = app.get_component(Toggler)
text = app.get_component(rio.Text)
async with rio.testing.TestClient(Toggler) as test_client:
root_component = test_client.get_component(Toggler)
text = test_client.get_component(rio.Text)
root_component.toggle = True
await app.refresh()
await test_client.refresh()
assert text.text == "Hello"
@@ -169,12 +170,12 @@ async def test_key_interrupts_structure():
def build(self):
return rio.Container(rio.Text(self.key_), key=self.key_)
async with create_mockapp(Toggler) as app:
root_component = app.get_component(Toggler)
text = app.get_component(rio.Text)
async with rio.testing.TestClient(Toggler) as test_client:
root_component = test_client.get_component(Toggler)
text = test_client.get_component(rio.Text)
root_component.key_ = "123"
await app.refresh()
await test_client.refresh()
# The container's key changed, so even though the structure is the same,
# the old Text component should be unchanged.
@@ -194,12 +195,12 @@ async def test_structural_matching_inside_keyed_component():
rio.Container(rio.Text("C"), key="foo"),
)
async with create_mockapp(Toggler) as app:
root_component = app.get_component(Toggler)
text = app.get_component(rio.Text)
async with rio.testing.TestClient(Toggler) as test_client:
root_component = test_client.get_component(Toggler)
text = test_client.get_component(rio.Text)
root_component.toggle = True
await app.refresh()
await test_client.refresh()
# The container with key "foo" has moved. Make sure the structure inside
# of it was reconciled correctly.
@@ -225,12 +226,12 @@ async def test_key_matching_inside_keyed_component():
key="row",
)
async with create_mockapp(Toggler) as app:
root_component = app.get_component(Toggler)
text = app.get_component(rio.Text)
async with rio.testing.TestClient(Toggler) as test_client:
root_component = test_client.get_component(Toggler)
text = test_client.get_component(rio.Text)
root_component.toggle = True
await app.refresh()
await test_client.refresh()
# The container with key "foo" has moved. Make sure the structure inside
# of it was reconciled correctly.
@@ -253,12 +254,12 @@ async def test_same_key_on_different_component_type():
else:
return ComponentWithText("World", key="foo")
async with create_mockapp(Toggler) as app:
root_component = app.get_component(Toggler)
text = app.get_component(rio.Text)
async with rio.testing.TestClient(Toggler) as test_client:
root_component = test_client.get_component(Toggler)
text = test_client.get_component(rio.Text)
root_component.toggle = True
await app.refresh()
await test_client.refresh()
assert text.text == "Hello"
@@ -270,12 +271,12 @@ async def test_text_reconciliation():
def build(self) -> rio.Component:
return rio.Text(self.text)
async with create_mockapp(RootComponent) as app:
root = app.get_component(RootComponent)
text = app.get_component(rio.Text)
async with rio.testing.TestClient(RootComponent) as test_client:
root = test_client.get_component(RootComponent)
text = test_client.get_component(rio.Text)
root.text = "bar"
await app.refresh()
await test_client.refresh()
assert text.text == root.text
@@ -288,14 +289,14 @@ async def test_grid_reconciliation():
rows = [[rio.Text(f"Row {n}")] for n in range(self.num_rows)]
return rio.Grid(*rows)
async with create_mockapp(RootComponent) as app:
root = app.get_component(RootComponent)
grid = app.get_component(rio.Grid)
async with rio.testing.TestClient(RootComponent) as test_client:
root = test_client.get_component(RootComponent)
grid = test_client.get_component(rio.Grid)
root.num_rows += 1
await app.refresh()
await test_client.refresh()
assert {root, grid} < app.last_updated_components
assert {root, grid} < test_client._last_updated_components
assert len(grid._children) == root.num_rows
@@ -317,12 +318,12 @@ async def test_margin_reconciliation():
rio.Text("hi", margin=1),
)
async with create_mockapp(RootComponent) as app:
root = app.get_component(RootComponent)
texts = list(app.get_components(rio.Text))
async with rio.testing.TestClient(RootComponent) as test_client:
root = test_client.get_component(RootComponent)
texts = list(test_client.get_components(rio.Text))
root.switch = False
await app.refresh()
await test_client.refresh()
assert texts[0].margin_left == 1
assert texts[1].margin_right == 1
@@ -359,9 +360,11 @@ async def test_reconcile_instance_with_itself():
def build() -> rio.Component:
return Container(rio.Text("foo"))
async with create_mockapp(build, use_ordered_dirty_set=True) as app:
container = app.get_component(Container)
child = app.get_component(rio.Text)
async with rio.testing.TestClient(
build, use_ordered_dirty_set=True
) as test_client:
container = test_client.get_component(Container)
child = test_client.get_component(rio.Text)
# Change the child's state and make its parent rebuild
child.text = "bar"
@@ -369,7 +372,11 @@ async def test_reconcile_instance_with_itself():
# In order for the bug to occur, the parent has to be rebuilt before the
# child
assert list(app.session._dirty_components) == [child, container]
await app.refresh()
assert test_client._session is not None
assert list(test_client._session._dirty_components) == [
child,
container,
]
await test_client.refresh()
assert app.last_updated_components == {child, container}
assert test_client._last_updated_components == {child, container}
tests/test_refresh.py
+26 -25
View File
@@ -1,18 +1,16 @@
from utils import create_mockapp
import rio
import rio.testing
async def test_refresh_with_nothing_to_do():
def build():
return rio.Text("Hello")
async with create_mockapp(build) as app:
app.outgoing_messages.clear()
await app.refresh()
async with rio.testing.TestClient(build) as test_client:
test_client._outgoing_messages.clear()
await test_client.refresh()
assert not app.dirty_components
assert not app.last_updated_components
assert not test_client._dirty_components
assert not test_client._last_updated_components
async def test_refresh_with_clean_root_component():
@@ -20,13 +18,13 @@ async def test_refresh_with_clean_root_component():
text_component = rio.Text("Hello")
return rio.Container(text_component)
async with create_mockapp(build) as app:
text_component = app.get_component(rio.Text)
async with rio.testing.TestClient(build) as test_client:
text_component = test_client.get_component(rio.Text)
text_component.text = "World"
await app.refresh()
await test_client.refresh()
assert app.last_updated_components == {text_component}
assert test_client._last_updated_components == {text_component}
async def test_rebuild_component_with_dead_parent():
@@ -50,20 +48,20 @@ async def test_rebuild_component_with_dead_parent():
)
)
async with create_mockapp(build) as app:
async with rio.testing.TestClient(build) as test_client:
# Change the component's state, but also remove its parent from the
# component tree
root_component = app.get_component(RootComponent)
component = app.get_component(ComponentWithState)
progress_component = app.get_component(rio.ProgressCircle)
root_component = test_client.get_component(RootComponent)
component = test_client.get_component(ComponentWithState)
progress_component = test_client.get_component(rio.ProgressCircle)
component.state = "Hi"
root_component.content = progress_component
await app.refresh()
await test_client.refresh()
# Make sure no data for dead components was sent to JS
assert app.last_updated_components == {root_component}
assert test_client._last_updated_components == {root_component}
async def test_unmount_and_remount():
@@ -81,18 +79,21 @@ async def test_unmount_and_remount():
show_child=True,
)
async with create_mockapp(build) as app:
root_component = app.get_component(DemoComponent)
async with rio.testing.TestClient(build) as test_client:
root_component = test_client.get_component(DemoComponent)
child_component = root_component.content
row_component = app.get_component(rio.Row)
row_component = test_client.get_component(rio.Row)
root_component.show_child = False
await app.refresh()
assert app.last_updated_components == {root_component, row_component}
await test_client.refresh()
assert test_client._last_updated_components == {
root_component,
row_component,
}
root_component.show_child = True
await app.refresh()
assert app.last_updated_components == {
await test_client.refresh()
assert test_client._last_updated_components == {
root_component,
row_component,
child_component,
tests/test_session.py
+10 -13
View File
@@ -1,12 +1,11 @@
import pytest
from utils import create_mockapp
import rio
import rio.testing
async def test_session_attachments():
async with create_mockapp() as app:
session = app.session
async def test_client_attachments():
async with rio.testing.TestClient() as test_client:
session = test_client.session
list1 = ["foo", "bar"]
list2 = []
@@ -19,11 +18,9 @@ async def test_session_attachments():
async def test_access_nonexistent_session_attachment():
async with create_mockapp() as app:
session = app.session
async with rio.testing.TestClient() as test_client:
with pytest.raises(KeyError):
session[list]
test_client.session[list]
async def test_default_attachments():
@@ -33,13 +30,13 @@ async def test_default_attachments():
dict_attachment = {"foo": "bar"}
settings_attachment = Settings(3)
async with create_mockapp(
async with rio.testing.TestClient(
default_attachments=[dict_attachment, settings_attachment]
) as app:
session = app.session
) as test_client:
session = test_client.session
# Default attachments shouldn't be copied, unless they're UserSettings
assert session[dict] is dict_attachment
assert session[Settings] == settings_attachment
assert session[Settings] is not settings_attachment
assert session[Settings]._equals(settings_attachment)
tests/test_testing_tools.py
+30
View File
@@ -0,0 +1,30 @@
import rio.testing
async def test_crashed_build_functions_are_tracked():
def build() -> rio.Component:
return 3 # type: ignore
async with rio.testing.TestClient(build) as test_client:
assert len(test_client.crashed_build_functions) == 1
async def test_rebuild_resets_crashed_build_functions():
class CrashingComponent(rio.Component):
fail: bool = True
def build(self) -> rio.Component:
if self.fail:
raise RuntimeError
else:
return rio.Text("hi")
async with rio.testing.TestClient(CrashingComponent) as test_client:
assert len(test_client.crashed_build_functions) == 1
crashing_component = test_client.get_component(CrashingComponent)
crashing_component.fail = False
await test_client.refresh()
assert not test_client.crashed_build_functions
tests/test_user_settings.py
+9 -10
View File
@@ -1,8 +1,7 @@
import aiofiles
from uniserde import JsonDoc
from utils import create_mockapp
import rio
import rio.testing
class FakeFile:
@@ -35,13 +34,13 @@ async def test_load_settings():
"foo:bar": "baz",
}
async with create_mockapp(
async with rio.testing.TestClient(
running_in_window=False,
default_attachments=(RootSettings(), FooSettings()),
user_settings=user_settings,
) as app:
root_settings = app.session[RootSettings]
foo_settings = app.session[FooSettings]
) as test_client:
root_settings = test_client.session[RootSettings]
foo_settings = test_client.session[FooSettings]
assert root_settings.foo == "bar"
assert foo_settings.bar == "baz"
@@ -54,12 +53,12 @@ async def test_load_settings_file(monkeypatch):
lambda _: FakeFile('{"foo": "bar", "section:foo": {"bar": "baz"} }'),
)
async with create_mockapp(
async with rio.testing.TestClient(
running_in_window=True,
default_attachments=(RootSettings(), FooSettings()),
) as app:
root_settings = app.session[RootSettings]
foo_settings = app.session[FooSettings]
) as test_client:
root_settings = test_client.session[RootSettings]
foo_settings = test_client.session[FooSettings]
assert root_settings.foo == "bar"
assert foo_settings.bar == "baz"
tests/test_zzz_guardrails.py
+3 -18
View File
@@ -2,9 +2,9 @@
# that the remaining tests run without the monkeypatches applied.
import pytest
from utils import create_mockapp, enable_component_instantiation
from utils import enable_component_instantiation
import rio
import rio.testing
from rio.debug.monkeypatches import apply_monkeypatches
apply_monkeypatches()
@@ -62,21 +62,6 @@ def test_component_class_can_be_used_as_build_function(
_ = rio.PageView(fallback_build=component_cls)
@pytest.mark.parametrize(
"component_cls",
[
rio.Dropdown,
rio.Text,
],
)
@enable_component_instantiation
def test_component_class_cant_be_used_as_build_function(
component_cls: type[rio.Component],
):
with pytest.raises(Exception):
_ = rio.PageView(fallback_build=component_cls)
async def test_init_cannot_read_state_properties():
# Accessing state properties in `__init__` is not allowed because state
# bindings aren't initialized yet at that point. In development mode, trying
@@ -116,7 +101,7 @@ async def test_init_cannot_read_state_properties():
def build(self) -> rio.Component:
return IllegalComponent(17)
async with create_mockapp(Container):
async with rio.testing.TestClient(Container):
assert init_executed
assert accessing_foo_raised_exception
assert accessing_margin_top_raised_exception
tests/utils.py
+9 -219
View File
@@ -1,237 +1,24 @@
import asyncio
import contextlib
import functools
import inspect
import json
import types
from collections.abc import (
AsyncGenerator,
Callable,
Container,
Iterable,
Iterator,
Mapping,
)
from typing import Any, TypeVar
from typing import TypeVar
import ordered_set
from uniserde import Jsonable, JsonDoc
from uniserde import Jsonable
import rio.global_state
from rio.app_server import AppServer
from rio.components.root_components import (
FundamentalRootComponent,
HighLevelRootComponent,
)
__all__ = ["enable_component_instantiation"]
T = TypeVar("T")
C = TypeVar("C", bound=rio.Component)
async def _make_awaitable(value: T = None) -> T:
return value
class MockApp:
def __init__(
self,
session: rio.Session,
user_settings: JsonDoc = {},
) -> None:
self.session = session
self.outgoing_messages: list[JsonDoc] = []
self._first_refresh_completed = asyncio.Event()
self._responses = asyncio.Queue[JsonDoc]()
self._responses.put_nowait(
{
"websiteUrl": "https://unit.test",
"preferredLanguages": [],
"userSettings": user_settings,
"windowWidth": 1920,
"windowHeight": 1080,
"timezone": "America/New_York",
"decimalSeparator": ".",
"thousandsSeparator": ",",
"prefersLightTheme": True,
}
)
async def _send_message(self, message_text: str) -> None:
message = json.loads(message_text)
self.outgoing_messages.append(message)
if "id" in message:
self._responses.put_nowait(
{
"jsonrpc": "2.0",
"id": message["id"],
"result": None,
}
)
if message["method"] == "updateComponentStates":
self._first_refresh_completed.set()
async def _receive_message(self) -> Jsonable:
return await self._responses.get()
@property
def dirty_components(self) -> Container[rio.Component]:
return set(self.session._dirty_components)
@property
def last_updated_components(self) -> set[rio.Component]:
return set(self.last_component_state_changes)
@property
def last_component_state_changes(
self,
) -> Mapping[rio.Component, Mapping[str, object]]:
for message in reversed(self.outgoing_messages):
if message["method"] == "updateComponentStates":
delta_states: dict = message["params"]["deltaStates"] # type: ignore
return {
self.session._weak_components_by_id[
int(component_id)
]: delta
for component_id, delta in delta_states.items()
if int(component_id) != self.session._root_component._id
}
return {}
def get_root_component(self) -> rio.Component:
sess = self.session
high_level_root = sess._root_component
assert isinstance(
high_level_root, HighLevelRootComponent
), high_level_root
low_level_root = sess._weak_component_data_by_component[
high_level_root
].build_result
assert isinstance(
low_level_root, FundamentalRootComponent
), low_level_root
scroll_container = low_level_root.content
assert isinstance(
scroll_container, rio.ScrollContainer
), scroll_container
return scroll_container.content
def get_components(self, component_type: type[C]) -> Iterator[C]:
root_component = self.get_root_component()
for component in root_component._iter_component_tree():
if type(component) is component_type:
yield component # type: ignore
def get_component(self, component_type: type[C]) -> C:
try:
return next(self.get_components(component_type))
except StopIteration:
raise AssertionError(f"No component of type {component_type} found")
def get_build_output(
self,
component: rio.Component,
type_: type[C] | None = None,
) -> C:
result = self.session._weak_component_data_by_component[
component
].build_result
if type_ is not None:
assert (
type(result) is type_
), f"Expected {type_}, got {type(result)}"
return result # type: ignore
async def refresh(self) -> None:
await self.session._refresh()
@contextlib.asynccontextmanager
async def create_mockapp(
build: Callable[[], rio.Component] = lambda: rio.Text("hi"),
*,
app_name: str = "mock-app",
running_in_window: bool = False,
user_settings: JsonDoc = {},
default_attachments: Iterable[object] = (),
use_ordered_dirty_set: bool = False,
) -> AsyncGenerator[MockApp, None]:
app = rio.App(
build=build,
name=app_name,
default_attachments=tuple(default_attachments),
)
app_server = AppServer(
app,
debug_mode=False,
running_in_window=running_in_window,
validator_factory=None,
internal_on_app_start=None,
)
# Emulate the process of creating a session as closely as possible
fake_request: Any = types.SimpleNamespace(
url="https://unit.test",
base_url="https://unit.test",
headers={"accept": "text/html"},
)
await app_server._serve_index(fake_request, "")
[[session_token, session]] = app_server._active_session_tokens.items()
if use_ordered_dirty_set:
session._dirty_components = ordered_set.OrderedSet(
session._dirty_components
) # type: ignore
mock_app = MockApp(session, user_settings=user_settings)
fake_websocket: Any = types.SimpleNamespace(
client="1.2.3.4",
accept=lambda: _make_awaitable(),
send_text=mock_app._send_message,
receive_json=mock_app._receive_message,
)
test_task = asyncio.current_task()
assert test_task is not None
async def serve_websocket():
try:
await app_server._serve_websocket(fake_websocket, session_token)
except asyncio.CancelledError:
pass
except Exception as error:
test_task.cancel(
f"Exception in AppServer._serve_websocket: {error}"
)
else:
test_task.cancel(
"AppServer._serve_websocket exited unexpectedly. An exception"
" must have occurred in the `init_coro`."
)
server_task = asyncio.create_task(serve_websocket())
await mock_app._first_refresh_completed.wait()
try:
yield mock_app
finally:
server_task.cancel()
def enable_component_instantiation(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
@@ -254,6 +41,9 @@ def enable_component_instantiation(func):
session = rio.Session(
app_server,
"<a fake session token>",
"<a fake client ip>",
12345,
"<a fake user agent>",
)
session._decimal_separator = "."
session._thousands_separator = ","