2023-09-10 05:22:53 +03:00
|
|
|
"""Record interactions and save as an automated test."""
|
|
|
|
|
2023-09-09 22:25:51 +03:00
|
|
|
import os
|
2023-09-10 04:27:23 +03:00
|
|
|
from typing import Any, Callable
|
2023-09-10 07:00:55 +03:00
|
|
|
|
2023-09-10 06:28:43 +03:00
|
|
|
from rich.text import Text
|
2023-09-09 21:49:00 +03:00
|
|
|
from textual.css.query import NoMatches, TooManyMatches
|
|
|
|
from textual.dom import DOMNode
|
2023-09-10 04:27:23 +03:00
|
|
|
from textual.errors import NoWidget
|
2023-09-09 21:49:00 +03:00
|
|
|
from textual.events import Event, Key, MouseDown, MouseMove, MouseUp
|
2023-09-09 22:46:35 +03:00
|
|
|
from textual.geometry import Offset
|
2023-09-09 23:26:06 +03:00
|
|
|
from textual.pilot import Pilot
|
2023-09-09 21:49:00 +03:00
|
|
|
from textual.screen import Screen
|
2023-09-10 07:00:55 +03:00
|
|
|
|
2023-09-09 21:49:00 +03:00
|
|
|
from textual_paint.paint import PaintApp
|
|
|
|
|
2023-09-10 07:00:55 +03:00
|
|
|
|
2023-09-09 22:25:51 +03:00
|
|
|
def unique_file(path: str) -> str:
|
2023-09-10 07:14:20 +03:00
|
|
|
"""Return a path that doesn't exist yet, by appending a number to the filename."""
|
2023-09-09 22:25:51 +03:00
|
|
|
filename, extension = os.path.splitext(path)
|
|
|
|
counter = 1
|
2023-09-09 21:49:00 +03:00
|
|
|
|
2023-09-09 22:25:51 +03:00
|
|
|
while os.path.exists(path):
|
|
|
|
path = f"{filename}_{counter}{extension}"
|
|
|
|
counter += 1
|
2023-09-09 21:49:00 +03:00
|
|
|
|
2023-09-09 22:25:51 +03:00
|
|
|
return path
|
|
|
|
|
2023-09-09 23:47:35 +03:00
|
|
|
def indent(text: str, spaces: int) -> str:
|
2023-09-10 07:14:20 +03:00
|
|
|
"""Return the text indented by the given number of spaces (including the first line)."""
|
2023-09-09 23:47:35 +03:00
|
|
|
return "\n".join(" " * spaces + line for line in text.splitlines())
|
|
|
|
|
2023-09-10 05:22:56 +03:00
|
|
|
async def async_exec(code: str, **kwargs: object) -> object:
|
2023-09-10 07:14:20 +03:00
|
|
|
"""Execute the given code in an async function and return the result. Keyword arguments are made available as variables."""
|
2023-09-10 05:22:56 +03:00
|
|
|
# This dict will be used for passing variables to the `exec`ed code
|
|
|
|
# as well as retrieving the function defined by the code.
|
|
|
|
scope = kwargs
|
2023-09-09 22:25:51 +03:00
|
|
|
|
2023-09-10 05:22:56 +03:00
|
|
|
# Make an async function with the code and `exec` it
|
|
|
|
exec(f"async def async_exec_code():\n{indent(code, 4)}", scope)
|
|
|
|
|
|
|
|
# Get `async_exec_code` from the scope, call it and return the result
|
|
|
|
return await scope['async_exec_code']() # type: ignore
|
2023-09-09 22:25:51 +03:00
|
|
|
|
|
|
|
def get_selector(target: DOMNode) -> tuple[str, int|None]:
|
2023-09-09 21:49:00 +03:00
|
|
|
"""Return a selector that can be used to find the widget."""
|
2023-09-10 05:22:56 +03:00
|
|
|
app = target.app
|
2023-09-09 21:49:00 +03:00
|
|
|
widget = target
|
|
|
|
if widget.id:
|
2023-09-09 22:25:51 +03:00
|
|
|
return f"#{widget.id}", None
|
2023-09-09 21:49:00 +03:00
|
|
|
selector = widget.css_identifier
|
|
|
|
while widget.parent and not isinstance(widget.parent, Screen):
|
|
|
|
widget = widget.parent
|
|
|
|
if widget.id:
|
|
|
|
selector = f"#{widget.id} {selector}"
|
|
|
|
break
|
|
|
|
else:
|
|
|
|
selector = f"{widget.css_identifier} {selector}"
|
|
|
|
try:
|
|
|
|
query_result = app.query_one(selector)
|
|
|
|
except TooManyMatches:
|
2023-09-10 04:48:52 +03:00
|
|
|
return selector, app.query(selector).nodes.index(target) # type: ignore
|
2023-09-09 22:25:51 +03:00
|
|
|
# smarter differentiators would be nice, like tooltip or text content,
|
|
|
|
# but at least with indices, you'll know when you changed the tab order
|
2023-09-09 21:49:00 +03:00
|
|
|
except NoMatches:
|
|
|
|
raise Exception(f"Selector {selector!r} didn't match the target widget ({target!r})")
|
|
|
|
if query_result is not target:
|
|
|
|
raise Exception(f"Selector {selector!r} matched a different widget than the target ({query_result!r} rather than {target!r})")
|
|
|
|
|
2023-09-09 22:25:51 +03:00
|
|
|
return selector, None
|
2023-09-09 21:49:00 +03:00
|
|
|
|
2023-09-10 06:28:56 +03:00
|
|
|
original_on_event = PaintApp.on_event
|
|
|
|
|
2023-09-10 05:22:56 +03:00
|
|
|
class PilotRecorder():
|
2023-09-10 07:14:20 +03:00
|
|
|
"""Record (and undo and replay) interactions with an app, and save as a test."""
|
2023-09-10 05:22:56 +03:00
|
|
|
def __init__(self) -> None:
|
|
|
|
self.app: PaintApp | None = None
|
|
|
|
self.steps: list[tuple[Event, Offset, str, int|None]] = []
|
|
|
|
self.replaying: bool = False
|
|
|
|
self.output_file = unique_file("tests/test_paint_something.py")
|
|
|
|
self.next_after_exit: Callable[[], None] | None = None
|
|
|
|
|
|
|
|
recorder = self
|
|
|
|
async def on_event(self: PaintApp, event: Event) -> None:
|
2023-09-10 07:00:09 +03:00
|
|
|
# - Record before the event is handled, so a clicked widget that removes itself,
|
|
|
|
# such as an OK button in a dialog, will still be in the DOM when we record it.
|
|
|
|
# - Every event seems to be received twice, once with _forwarded set and once without.
|
|
|
|
# I don't claim to understand the forwarding scheme, but ignoring either
|
|
|
|
# the forwarded or the un-forwarded events seems workable.
|
2023-09-10 06:42:30 +03:00
|
|
|
if not event._forwarded:
|
2023-09-10 07:14:20 +03:00
|
|
|
recorder.handle_event(event)
|
2023-09-10 05:58:54 +03:00
|
|
|
await original_on_event(self, event)
|
2023-09-10 05:22:56 +03:00
|
|
|
self.app_on_event = on_event
|
2023-09-10 07:00:55 +03:00
|
|
|
|
2023-09-10 07:14:20 +03:00
|
|
|
def handle_event(self, event: Event) -> None:
|
|
|
|
"""Record the event as a step, or handle certain key presses as commands."""
|
2023-09-10 05:29:54 +03:00
|
|
|
assert self.app is not None, "app should be set if we're recording an event from it"
|
2023-09-12 00:59:54 +03:00
|
|
|
# Handling any event means including it in the undo stack right now.
|
|
|
|
# Don't want to undo a single mouse-move, especially when it doesn't do anything yet.
|
|
|
|
# if isinstance(event, (MouseDown, MouseMove, MouseUp)):
|
|
|
|
if isinstance(event, MouseDown):
|
2023-09-10 06:28:56 +03:00
|
|
|
if self.replaying:
|
|
|
|
return
|
2023-09-10 05:22:56 +03:00
|
|
|
try:
|
|
|
|
widget, _ = self.app.get_widget_at(*event.screen_offset)
|
|
|
|
except NoWidget:
|
|
|
|
return
|
|
|
|
offset = event.screen_offset - widget.region.offset
|
|
|
|
self.steps.append((event, offset, *get_selector(widget)))
|
2023-09-12 00:57:01 +03:00
|
|
|
self.steps_changed()
|
2023-09-09 23:26:06 +03:00
|
|
|
elif isinstance(event, Key):
|
2023-09-10 05:22:56 +03:00
|
|
|
if event.key == "ctrl+z" and self.steps:
|
|
|
|
self.steps.pop()
|
2023-09-12 00:57:01 +03:00
|
|
|
self.steps_changed()
|
2023-09-10 05:22:56 +03:00
|
|
|
self.run() # restart the app to replay up to this point
|
|
|
|
elif event.key == "ctrl+c":
|
|
|
|
self.save_replay()
|
2023-09-10 06:28:43 +03:00
|
|
|
self.app.exit(None, Text("Saved test recording to " + self.output_file))
|
2023-09-10 05:22:56 +03:00
|
|
|
else:
|
2023-09-10 06:28:56 +03:00
|
|
|
if self.replaying:
|
|
|
|
return
|
2023-09-10 05:22:56 +03:00
|
|
|
self.steps.append((event, Offset(), "", None))
|
2023-09-12 00:57:01 +03:00
|
|
|
self.steps_changed()
|
|
|
|
|
|
|
|
def steps_changed(self) -> None:
|
2023-09-10 07:14:20 +03:00
|
|
|
"""Save the steps any time they change."""
|
2023-09-12 00:57:01 +03:00
|
|
|
# Could implement a debug view of the steps, but just saving to the file is good enough for now.
|
|
|
|
self.save_replay()
|
2023-09-09 23:26:06 +03:00
|
|
|
|
2023-09-10 05:22:56 +03:00
|
|
|
async def replay_steps(self, pilot: Pilot[Any]) -> None:
|
2023-09-10 07:14:20 +03:00
|
|
|
"""Replay the recorded steps, in the current app instance."""
|
2023-09-10 05:22:56 +03:00
|
|
|
if not self.steps:
|
|
|
|
return
|
|
|
|
self.replaying = True
|
|
|
|
await async_exec(self.get_replay_code(), pilot=pilot, Offset=Offset)
|
|
|
|
self.replaying = False
|
|
|
|
|
|
|
|
def run(self) -> None:
|
2023-09-10 07:14:20 +03:00
|
|
|
"""Start the app, or restart it to replay the recorded steps."""
|
2023-09-10 05:22:56 +03:00
|
|
|
def startup_and_replay() -> None:
|
2023-09-10 07:14:20 +03:00
|
|
|
"""Start the app, hook its events, and replay steps if there are any."""
|
2023-09-10 05:22:56 +03:00
|
|
|
self.next_after_exit = None # important to allowing you to exit; don't keep launching the app
|
|
|
|
self.app = PaintApp()
|
|
|
|
self.app.on_event = self.app_on_event.__get__(self.app)
|
|
|
|
self.app.run(auto_pilot=self.replay_steps)
|
|
|
|
# run is blocking, so this will happen after the app exits
|
|
|
|
if self.next_after_exit:
|
|
|
|
self.next_after_exit()
|
|
|
|
if self.app is not None:
|
|
|
|
# exit can't be awaited, because it stops the whole event loop (eventually)
|
|
|
|
# but we need to wait for the event loop to stop before we can start a new app
|
|
|
|
self.next_after_exit = startup_and_replay
|
|
|
|
self.app.exit()
|
|
|
|
else:
|
|
|
|
startup_and_replay()
|
|
|
|
|
|
|
|
def get_replay_code(self) -> str:
|
2023-09-10 07:14:20 +03:00
|
|
|
"""Return code to replay the recorded steps."""
|
2023-09-10 05:22:56 +03:00
|
|
|
steps_code = ""
|
|
|
|
for event, offset, selector, index in self.steps:
|
|
|
|
if isinstance(event, MouseDown):
|
|
|
|
if index is None:
|
|
|
|
steps_code += f"await pilot.click({selector!r}, offset=Offset({offset.x}, {offset.y}))\n"
|
|
|
|
else:
|
|
|
|
steps_code += f"widget = pilot.app.query({selector!r})[{index!r}]\n"
|
|
|
|
# can't pass a widget to pilot.click, only a selector, or None
|
|
|
|
steps_code += f"await pilot.click(offset=Offset({offset.x}, {offset.y}) + widget.region.offset)\n"
|
|
|
|
elif isinstance(event, MouseMove):
|
|
|
|
# TODO: generate code for drags (but not extraneous mouse movement)
|
|
|
|
pass
|
|
|
|
elif isinstance(event, MouseUp):
|
|
|
|
pass
|
|
|
|
elif isinstance(event, Key):
|
|
|
|
steps_code += f"await pilot.press({event.key!r})\n"
|
|
|
|
else:
|
|
|
|
raise Exception(f"Unexpected event type {type(event)}")
|
|
|
|
return steps_code or "pass"
|
2023-09-09 22:25:51 +03:00
|
|
|
|
2023-09-10 05:22:56 +03:00
|
|
|
def save_replay(self) -> None:
|
2023-09-10 07:14:20 +03:00
|
|
|
"""Save the recorded steps as a test file."""
|
2023-09-10 05:22:56 +03:00
|
|
|
assert self.app is not None, "app should be set by now"
|
2023-09-09 21:49:00 +03:00
|
|
|
|
2023-09-10 05:22:56 +03:00
|
|
|
script = f"""\
|
2023-09-09 21:49:00 +03:00
|
|
|
from pathlib import Path, PurePath
|
|
|
|
from typing import Awaitable, Callable, Iterable, Protocol
|
|
|
|
|
|
|
|
import pytest
|
|
|
|
from textual.geometry import Offset
|
|
|
|
from textual.pilot import Pilot
|
|
|
|
from textual.widgets import Input
|
|
|
|
|
|
|
|
class SnapCompareType(Protocol):
|
|
|
|
\"\"\"Type of the function returned by the snap_compare fixture.\"\"\"
|
|
|
|
def __call__(
|
|
|
|
self,
|
|
|
|
app_path: str | PurePath,
|
|
|
|
press: Iterable[str] = (),
|
|
|
|
terminal_size: tuple[int, int] = (80, 24),
|
|
|
|
run_before: Callable[[Pilot], Awaitable[None] | None] | None = None, # type: ignore
|
|
|
|
) -> bool:
|
|
|
|
...
|
|
|
|
|
|
|
|
# Relative paths are treated as relative to this file, when using snap_compare.
|
|
|
|
PAINT = Path("../src/textual_paint/paint.py")
|
|
|
|
|
|
|
|
# Prevent flaky tests due to timing issues.
|
|
|
|
Input.cursor_blink = False # type: ignore
|
|
|
|
|
|
|
|
def test_paint_something(snap_compare: SnapCompareType):
|
|
|
|
async def test_paint_something_steps(pilot: Pilot[None]):
|
2023-09-10 05:22:56 +03:00
|
|
|
{indent(self.get_replay_code(), 8)}
|
2023-09-09 21:49:00 +03:00
|
|
|
|
2023-09-10 05:22:56 +03:00
|
|
|
assert snap_compare(PAINT, run_before=test_paint_something_steps, terminal_size=({self.app.size.width}, {self.app.size.height}))
|
2023-09-09 21:49:00 +03:00
|
|
|
"""
|
2023-09-10 05:22:56 +03:00
|
|
|
with open(self.output_file, "w") as f:
|
|
|
|
f.write(script)
|
2023-09-09 21:49:00 +03:00
|
|
|
|
|
|
|
if __name__ == "__main__":
|
2023-09-10 05:22:56 +03:00
|
|
|
recorder = PilotRecorder()
|
|
|
|
recorder.run()
|