mirror of
https://github.com/Textualize/textual.git
synced 2025-10-17 02:38:12 +03:00
cab4925eaa
* New CLI runner * runner functionality * Add port * use env for port * changelog * test fix * flush * remove constant * comment * tidy docs * docstrings * punctuation * docstring * fix test * snapshot * Update src/textual/cli/cli.py Co-authored-by: Rodrigo Girão Serrão <5621605+rodrigogiraoserrao@users.noreply.github.com> * guard against bad imports * guard againsts screenshot * always print return * docstrings * docstrings --------- Co-authored-by: Rodrigo Girão Serrão <5621605+rodrigogiraoserrao@users.noreply.github.com>
108 lines
3.7 KiB
Python
108 lines
3.7 KiB
Python
import json
|
|
import types
|
|
from asyncio import Queue
|
|
from datetime import datetime
|
|
|
|
import msgpack
|
|
import time_machine
|
|
from aiohttp.web_ws import WebSocketResponse
|
|
from rich.console import ConsoleDimensions
|
|
from rich.panel import Panel
|
|
|
|
from tests.utilities.render import wait_for_predicate
|
|
from textual.constants import DEVTOOLS_PORT
|
|
from textual.devtools.client import DevtoolsClient
|
|
from textual.devtools.redirect_output import DevtoolsLog
|
|
|
|
CALLER_LINENO = 123
|
|
CALLER_PATH = "a/b/c.py"
|
|
CALLER = types.SimpleNamespace(filename=CALLER_PATH, lineno=CALLER_LINENO)
|
|
TIMESTAMP = 1649166819
|
|
|
|
|
|
def test_devtools_client_initialize_defaults():
|
|
devtools = DevtoolsClient()
|
|
assert devtools.url == f"ws://127.0.0.1:{DEVTOOLS_PORT}"
|
|
|
|
|
|
async def test_devtools_client_is_connected(devtools):
|
|
assert devtools.is_connected
|
|
|
|
|
|
@time_machine.travel(datetime.utcfromtimestamp(TIMESTAMP))
|
|
async def test_devtools_log_places_encodes_and_queues_message(devtools):
|
|
await devtools._stop_log_queue_processing()
|
|
devtools.log(DevtoolsLog("Hello, world!", CALLER))
|
|
queued_log = await devtools.log_queue.get()
|
|
queued_log_data = msgpack.unpackb(queued_log)
|
|
print(repr(queued_log_data))
|
|
|
|
|
|
@time_machine.travel(datetime.utcfromtimestamp(TIMESTAMP))
|
|
async def test_devtools_log_places_encodes_and_queues_many_logs_as_string(devtools):
|
|
await devtools._stop_log_queue_processing()
|
|
devtools.log(DevtoolsLog(("hello", "world"), CALLER))
|
|
queued_log = await devtools.log_queue.get()
|
|
queued_log_data = msgpack.unpackb(queued_log)
|
|
print(repr(queued_log_data))
|
|
assert queued_log_data == {
|
|
"type": "client_log",
|
|
"payload": {
|
|
"group": 0,
|
|
"verbosity": 0,
|
|
"timestamp": 1649166819,
|
|
"path": "a/b/c.py",
|
|
"line_number": 123,
|
|
"segments": b"\x80\x04\x95@\x00\x00\x00\x00\x00\x00\x00]\x94(\x8c\x0crich.segment\x94\x8c\x07Segment\x94\x93\x94\x8c\x0bhello world\x94NN\x87\x94\x81\x94h\x03\x8c\x01\n\x94NN\x87\x94\x81\x94e.",
|
|
},
|
|
}
|
|
|
|
|
|
async def test_devtools_log_spillover(devtools):
|
|
# Give the devtools an intentionally small max queue size
|
|
await devtools._stop_log_queue_processing()
|
|
devtools.log_queue = Queue(maxsize=2)
|
|
|
|
# Force spillover of 2
|
|
devtools.log(DevtoolsLog((Panel("hello, world"),), CALLER))
|
|
devtools.log(DevtoolsLog("second message", CALLER))
|
|
devtools.log(DevtoolsLog("third message", CALLER)) # Discarded by rate-limiting
|
|
devtools.log(DevtoolsLog("fourth message", CALLER)) # Discarded by rate-limiting
|
|
|
|
assert devtools.spillover == 2
|
|
|
|
# Consume log queue
|
|
while not devtools.log_queue.empty():
|
|
await devtools.log_queue.get()
|
|
|
|
# Add another message now that we're under spillover threshold
|
|
devtools.log(DevtoolsLog("another message", CALLER))
|
|
await devtools.log_queue.get()
|
|
|
|
# Ensure we're informing the server of spillover rate-limiting
|
|
spillover_message = await devtools.log_queue.get()
|
|
assert json.loads(spillover_message) == {
|
|
"type": "client_spillover",
|
|
"payload": {"spillover": 2},
|
|
}
|
|
|
|
|
|
async def test_devtools_client_update_console_dimensions(devtools, server):
|
|
"""Sending new server info through websocket from server to client should (eventually)
|
|
result in the dimensions of the devtools client console being updated to match.
|
|
"""
|
|
server_to_client: WebSocketResponse = next(
|
|
iter(server.app["service"].clients)
|
|
).websocket
|
|
server_info = {
|
|
"type": "server_info",
|
|
"payload": {
|
|
"width": 123,
|
|
"height": 456,
|
|
},
|
|
}
|
|
await server_to_client.send_json(server_info)
|
|
await wait_for_predicate(
|
|
lambda: devtools.console.size == ConsoleDimensions(123, 456)
|
|
)
|