izackp/webterm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Bump minor version and update ghostty-web

Browse Source
This commit is contained in:
GitHub Copilot
2026-01-28 16:13:08 +00:00
parent 69f0e2748f
commit b4d7f2e98a
63 changed files with 581 additions and 1432 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/__init__.py
+1 -1
View File
@@ -1 +1 @@
"""Tests for textual-webterm."""
"""Tests for webterm."""
tests/conftest.py
+5 -5
View File
@@ -1,4 +1,4 @@
"""Pytest configuration and fixtures for textual-webterm tests."""
"""Pytest configuration and fixtures for webterm tests."""
from __future__ import annotations
@@ -8,10 +8,10 @@ from unittest.mock import AsyncMock, MagicMock
import pytest
from textual_webterm.config import App, Config
from textual_webterm.local_server import LocalServer
from textual_webterm.poller import Poller
from textual_webterm.session_manager import SessionManager
from webterm.config import App, Config
from webterm.local_server import LocalServer
from webterm.poller import Poller
from webterm.session_manager import SessionManager
if TYPE_CHECKING:
from collections.abc import AsyncGenerator, Generator
tests/test_app_session.py
-193
View File
@@ -1,193 +0,0 @@
"""Tests for app_session module."""
import asyncio
import contextlib
import json
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from textual_webterm.app_session import AppSession, ProcessState
class TestProcessState:
"""Tests for ProcessState enum."""
def test_process_states_exist(self):
"""Test that all process states exist."""
assert ProcessState.PENDING is not None
assert ProcessState.RUNNING is not None
assert ProcessState.CLOSED is not None
class TestAppSession:
"""Tests for AppSession class."""
@pytest.fixture
def mock_path(self, tmp_path):
"""Create a mock path."""
return tmp_path
def test_init(self, mock_path):
"""Test AppSession initialization."""
session = AppSession(mock_path, "python app.py", "test-session")
assert session.working_directory == mock_path
assert session.command == "python app.py"
assert session.session_id == "test-session"
assert session.state == ProcessState.PENDING
def test_init_with_devtools(self, mock_path):
"""Test AppSession with devtools enabled."""
session = AppSession(mock_path, "python app.py", "test-session", devtools=True)
assert session.devtools is True
@pytest.mark.asyncio
async def test_send_bytes_not_running(self, mock_path):
"""Test send_bytes when not running returns False."""
session = AppSession(mock_path, "python app.py", "test-session")
# Session not started, will return False gracefully
result = await session.send_bytes(b"test")
assert result is False
@pytest.mark.asyncio
async def test_send_meta(self, mock_path):
"""Test send_meta."""
session = AppSession(mock_path, "python app.py", "test-session")
session._process = MagicMock()
session._process.stdin = MagicMock()
session._process.stdin.write = MagicMock()
session._process.stdin.drain = AsyncMock()
await session.send_meta({"key": "value"})
# Should handle meta data
@pytest.mark.asyncio
async def test_set_terminal_size(self, mock_path):
"""Test set_terminal_size."""
session = AppSession(mock_path, "python app.py", "test-session")
session._process = MagicMock()
session._process.stdin = MagicMock()
session._process.stdin.write = MagicMock()
session._process.stdin.drain = AsyncMock()
# Should not raise
await session.set_terminal_size(100, 50)
@pytest.mark.asyncio
async def test_close_not_running(self, mock_path):
"""Test close when not running handles gracefully."""
session = AppSession(mock_path, "python app.py", "test-session")
# No process running, close should handle gracefully (not crash)
await session.close()
assert session.state == ProcessState.CLOSING
@pytest.mark.asyncio
async def test_wait_no_task(self, mock_path):
"""Test wait when no task."""
session = AppSession(mock_path, "python app.py", "test-session")
# Should not raise
await session.wait()
def test_state_transitions(self, mock_path):
"""Test state transition tracking."""
session = AppSession(mock_path, "python app.py", "test-session")
assert session.state == ProcessState.PENDING
# Manually set state for testing
session.state = ProcessState.RUNNING
assert session.state == ProcessState.RUNNING
session.state = ProcessState.CLOSED
assert session.state == ProcessState.CLOSED
class TestAppSessionConnector:
"""Tests for AppSession with connector."""
@pytest.fixture
def mock_connector(self):
"""Create a mock connector."""
connector = MagicMock()
connector.on_data = AsyncMock()
connector.on_meta = AsyncMock()
connector.on_binary_encoded_message = AsyncMock()
connector.on_close = AsyncMock()
return connector
@pytest.mark.asyncio
async def test_start_creates_task(self, tmp_path, mock_connector):
"""Test that start creates a task."""
session = AppSession(tmp_path, "echo test", "test-session")
with (
patch.object(session, "open", new_callable=AsyncMock),
patch.object(session, "run", new_callable=AsyncMock),
):
task = await session.start(mock_connector)
assert task is not None
# Cancel to clean up
task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await task
def test_encode_packet(self, tmp_path):
session = AppSession(tmp_path, "echo test", "sid")
packet = session.encode_packet(b"D", b"abc")
assert packet[:1] == b"D"
assert packet[1:5] == (3).to_bytes(4, "big")
assert packet[5:] == b"abc"
@pytest.mark.asyncio
async def test_send_bytes_handles_broken_pipe(self, tmp_path):
session = AppSession(tmp_path, "echo test", "sid")
stdin = MagicMock()
stdin.write = MagicMock(side_effect=BrokenPipeError())
stdin.drain = AsyncMock()
session._process = MagicMock(stdin=stdin)
assert await session.send_bytes(b"x") is False
@pytest.mark.asyncio
async def test_send_meta_encodes_json_and_writes(self, tmp_path):
session = AppSession(tmp_path, "echo test", "sid")
stdin = MagicMock()
stdin.write = MagicMock()
stdin.drain = AsyncMock()
session._process = MagicMock(stdin=stdin)
meta = {"type": "hello", "n": 1}
assert await session.send_meta(meta) is True
written = stdin.write.call_args.args[0]
assert written[:1] == b"M"
payload = written[5:]
assert json.loads(payload.decode("utf-8")) == meta
@pytest.mark.asyncio
async def test_open_sets_env_and_cwd(self, tmp_path, monkeypatch):
session = AppSession(tmp_path, "echo test", "sid", devtools=True)
fake_proc = MagicMock()
fake_proc.stdin = MagicMock()
fake_proc.stdout = MagicMock()
fake_proc.stderr = MagicMock()
async def fake_create(command, **kwargs):
assert command == "echo test"
assert kwargs["cwd"] == str(tmp_path)
env = kwargs["env"]
assert env["COLUMNS"] == "100"
assert env["ROWS"] == "40"
assert "TEXTUAL" in env
return fake_proc
monkeypatch.setattr(asyncio, "create_subprocess_shell", fake_create)
monkeypatch.setattr(session, "set_terminal_size", AsyncMock())
await session.open(width=100, height=40)
assert session._process is fake_proc
# run() packet decoding coverage is exercised in test_app_session_run_packets.py
tests/test_app_session_run_packets.py
-113
View File
@@ -1,113 +0,0 @@
import asyncio
import json
from unittest.mock import AsyncMock, MagicMock
import pytest
from textual_webterm.app_session import AppSession
@pytest.fixture
def mock_connector():
connector = MagicMock()
connector.on_data = AsyncMock()
connector.on_meta = AsyncMock()
connector.on_binary_encoded_message = AsyncMock()
connector.on_close = AsyncMock()
return connector
@pytest.mark.asyncio
async def test_run_decodes_packets_and_forwards(tmp_path, mock_connector, monkeypatch):
from textual_webterm import app_session
session = AppSession(tmp_path, "echo test", "sid")
session._connector = mock_connector
session.start_time = 0.0
stdin = MagicMock()
stdin.write = MagicMock()
stdin.drain = AsyncMock()
stdout = MagicMock()
# Provide a second empty line so AppSession's readiness loop terminates cleanly.
stdout.readline = AsyncMock(side_effect=[b"__GANGLION__\n", b""])
payload_data = b"hello"
payload_meta = json.dumps({"type": "custom", "x": 1}).encode("utf-8")
payload_meta_exit = json.dumps({"type": "exit"}).encode("utf-8")
payload_bin = b"\x00\x01"
read_parts = [
b"D",
len(payload_data).to_bytes(4, "big"),
payload_data,
b"M",
len(payload_meta).to_bytes(4, "big"),
payload_meta,
b"M",
len(payload_meta_exit).to_bytes(4, "big"),
payload_meta_exit,
b"P",
len(payload_bin).to_bytes(4, "big"),
payload_bin,
]
async def readexactly(n: int) -> bytes:
await asyncio.sleep(0)
if not read_parts:
raise asyncio.IncompleteReadError(partial=b"", expected=n)
part = read_parts.pop(0)
assert len(part) == n
return part
stdout.readexactly = AsyncMock(side_effect=readexactly)
stderr = MagicMock()
stderr.read = AsyncMock(return_value=b"")
session._process = MagicMock(stdin=stdin, stdout=stdout, stderr=stderr, returncode=0)
monkeypatch.setattr(app_session.constants, "DEBUG", False)
await session.run()
mock_connector.on_data.assert_awaited_once_with(payload_data)
mock_connector.on_meta.assert_awaited_once_with({"type": "custom", "x": 1})
mock_connector.on_binary_encoded_message.assert_awaited_once_with(payload_bin)
assert stdin.write.called
mock_connector.on_close.assert_awaited_once()
@pytest.mark.asyncio
async def test_run_payload_too_large_breaks_loop(tmp_path, mock_connector, monkeypatch):
from textual_webterm import app_session
session = AppSession(tmp_path, "echo test", "sid")
session._connector = mock_connector
session.start_time = 0.0
stdin = MagicMock()
stdin.write = MagicMock()
stdin.drain = AsyncMock()
stdout = MagicMock()
stdout.readline = AsyncMock(side_effect=[b"__GANGLION__\n", b""])
async def readexactly(n: int) -> bytes:
await asyncio.sleep(0)
if n == 1:
return b"D"
if n == 4:
return (app_session.MAX_PAYLOAD_SIZE + 1).to_bytes(4, "big")
raise asyncio.IncompleteReadError(partial=b"", expected=n)
stdout.readexactly = AsyncMock(side_effect=readexactly)
stderr = MagicMock()
stderr.read = AsyncMock(return_value=b"")
session._process = MagicMock(stdin=stdin, stdout=stdout, stderr=stderr, returncode=0)
monkeypatch.setattr(app_session.constants, "DEBUG", False)
await session.run()
mock_connector.on_close.assert_awaited_once()
tests/test_cli.py
+62 -127
View File
@@ -1,85 +1,8 @@
"""Tests for CLI module."""
from pathlib import Path
import click
import pytest
from click.testing import CliRunner
class TestParseAppPath:
"""Tests for parse_app_path function."""
def test_parse_module_class(self):
"""Test parsing module:class format."""
from textual_webterm.cli import parse_app_path
module, cls = parse_app_path("mymodule:MyClass")
assert module == "mymodule"
assert cls == "MyClass"
def test_parse_nested_module_class(self):
"""Test parsing nested.module:class format."""
from textual_webterm.cli import parse_app_path
module, cls = parse_app_path("my.nested.module:MyClass")
assert module == "my.nested.module"
assert cls == "MyClass"
def test_parse_file_path_class(self):
"""Test parsing file/path.py:class format."""
from textual_webterm.cli import parse_app_path
module, cls = parse_app_path("path/to/file.py:MyClass")
assert module == "path/to/file.py"
assert cls == "MyClass"
def test_parse_no_colon_raises(self):
"""Test that missing colon raises BadParameter."""
from textual_webterm.cli import parse_app_path
with pytest.raises(click.BadParameter) as exc_info:
parse_app_path("invalid_format")
assert "Expected format" in str(exc_info.value)
class TestLoadAppClass:
"""Tests for load_app_class function."""
def test_load_nonexistent_module(self):
"""Test loading from non-existent module raises."""
from textual_webterm.cli import load_app_class
with pytest.raises(click.BadParameter) as exc_info:
load_app_class("nonexistent_module_xyz:MyClass")
assert "Could not import" in str(exc_info.value)
def test_load_nonexistent_class(self):
"""Test loading non-existent class from existing module raises."""
from textual_webterm.cli import load_app_class
with pytest.raises(click.BadParameter) as exc_info:
load_app_class("os:NonExistentClass")
assert "has no attribute" in str(exc_info.value)
def test_load_existing_class(self):
"""Test loading an existing class from a module."""
from textual_webterm.cli import load_app_class
# Load Path from pathlib
cls = load_app_class("pathlib:Path")
assert cls is Path
def test_load_from_file_nonexistent(self):
"""Test loading from non-existent file raises."""
from textual_webterm.cli import load_app_class
with pytest.raises(click.BadParameter) as exc_info:
load_app_class("/nonexistent/path.py:MyClass")
assert (
"not found" in str(exc_info.value).lower()
or "does not exist" in str(exc_info.value).lower()
)
from webterm import cli
class TestCLI:
@@ -87,7 +10,7 @@ class TestCLI:
def test_cli_help(self):
"""Test CLI help output."""
from textual_webterm.cli import app as cli_app
cli_app = cli.app
runner = CliRunner()
result = runner.invoke(cli_app, ["--help"])
@@ -95,8 +18,6 @@ class TestCLI:
assert "terminal" in result.output.lower() or "command" in result.output.lower()
def test_cli_runs_terminal_command(self, monkeypatch):
from textual_webterm import cli
calls: dict[str, object] = {}
class FakeServer:
@@ -119,9 +40,6 @@ class TestCLI:
def test_cli_runs_default_shell(self, monkeypatch):
import os
from textual_webterm import cli
calls: dict[str, object] = {}
class FakeServer:
@@ -143,33 +61,18 @@ class TestCLI:
assert result.exit_code == 0
assert calls["terminal"][1] == os.environ["SHELL"]
def test_cli_app_module_validation_rejects(self):
from textual_webterm.cli import app as cli_app
runner = CliRunner()
result = runner.invoke(cli_app, ["--app", "os;rm -rf /:Fake"])
assert result.exit_code != 0
def test_cli_version(self):
"""Test CLI version output."""
from textual_webterm.cli import app as cli_app
cli_app = cli.app
runner = CliRunner()
result = runner.invoke(cli_app, ["--version"])
assert result.exit_code == 0
assert "version" in result.output
def test_cli_invalid_app_path(self):
"""Test CLI with invalid app path."""
from textual_webterm.cli import app as cli_app
runner = CliRunner()
result = runner.invoke(cli_app, ["--app", "invalid"])
assert result.exit_code != 0
def test_cli_port_option(self):
"""Test CLI port option parsing."""
from textual_webterm.cli import app as cli_app
cli_app = cli.app
runner = CliRunner()
result = runner.invoke(cli_app, ["--help"])
@@ -177,51 +80,83 @@ class TestCLI:
def test_cli_host_option(self):
"""Test CLI host option parsing."""
from textual_webterm.cli import app as cli_app
cli_app = cli.app
runner = CliRunner()
result = runner.invoke(cli_app, ["--help"])
assert "--host" in result.output or "-H" in result.output
class TestModuleValidation:
"""Tests for module/class name validation in CLI."""
def test_invalid_module_characters(self):
"""Test that invalid module names are rejected."""
from textual_webterm.cli import app as cli_app
runner = CliRunner()
# Module with shell characters should be rejected or fail gracefully
result = runner.invoke(cli_app, ["--app", "os; rm -rf /:Fake"])
# Should not succeed
assert result.exit_code != 0
def test_invalid_class_name(self):
"""Test that invalid class names are rejected."""
from textual_webterm.cli import app as cli_app
runner = CliRunner()
result = runner.invoke(cli_app, ["--app", "os:123invalid"])
assert result.exit_code != 0
class TestCLIOptions:
"""Tests for CLI option handling."""
def test_debug_option(self):
"""Test --debug option exists."""
from textual_webterm.cli import app as cli_app
cli_app = cli.app
runner = CliRunner()
result = runner.invoke(cli_app, ["--help"])
assert "--app" in result.output
assert "--docker-watch" in result.output
def test_no_run_option(self):
"""Test --no-run option exists."""
from textual_webterm.cli import app as cli_app
cli_app = cli.app
runner = CliRunner()
result = runner.invoke(cli_app, ["--help"])
# Check that basic options are documented
assert "port" in result.output.lower()
def test_package_version_fallback(monkeypatch):
def raise_missing(_name: str):
raise cli.PackageNotFoundError("webterm")
monkeypatch.setattr(cli, "version", raise_missing)
assert cli._package_version() == "0.0.0"
def test_cli_docker_watch_mode(monkeypatch):
calls: dict[str, object] = {}
class FakeServer:
def __init__(self, *_args, **_kwargs):
calls["init"] = True
def add_terminal(self, name, command, slug):
calls["terminal"] = (name, command, slug)
async def run(self):
calls["run"] = True
monkeypatch.setattr(cli, "LocalServer", FakeServer)
monkeypatch.setattr(cli.asyncio, "run", lambda _coro: calls.setdefault("run", True))
monkeypatch.setattr(cli.constants, "DEBUG", True)
runner = CliRunner()
result = runner.invoke(cli.app, ["--docker-watch"])
assert result.exit_code == 0
assert "terminal" not in calls
def test_cli_windows_branch(monkeypatch):
calls: dict[str, object] = {}
class FakeServer:
def __init__(self, *_args, **_kwargs):
calls["init"] = True
def add_terminal(self, name, command, slug):
calls["terminal"] = (name, command, slug)
async def run(self):
calls["run"] = True
monkeypatch.setattr(cli, "LocalServer", FakeServer)
monkeypatch.setattr(cli.constants, "WINDOWS", True)
monkeypatch.setattr(cli.asyncio, "run", lambda _coro: calls.setdefault("run", True))
runner = CliRunner()
result = runner.invoke(cli.app, ["--docker-watch"])
assert result.exit_code == 0
assert calls.get("run") is True
tests/test_cli_landing.py
+2 -2
View File
@@ -3,9 +3,10 @@ from pathlib import Path
from click.testing import CliRunner
from webterm import cli
def test_cli_landing_manifest_runs(monkeypatch, tmp_path: Path):
from textual_webterm import cli
manifest = tmp_path / "landing.yaml"
manifest.write_text(
@@ -39,7 +40,6 @@ def test_cli_landing_manifest_runs(monkeypatch, tmp_path: Path):
def test_cli_compose_manifest_runs(monkeypatch, tmp_path: Path):
from textual_webterm import cli
manifest = tmp_path / "compose.yaml"
manifest.write_text(
tests/test_cli_run_app.py
-55
View File
@@ -1,55 +0,0 @@
"""Extra CLI coverage tests for app execution paths."""
from __future__ import annotations
from click.testing import CliRunner
def test_cli_runs_app_from_file(monkeypatch, tmp_path):
from textual_webterm import cli
app_file = tmp_path / "myapp.py"
app_file.write_text(
"""
class MyApp:
TITLE = "MyApp"
def run(self):
return 0
""".lstrip()
)
calls: dict[str, object] = {}
class FakeServer:
def __init__(self, *_args, **_kwargs):
calls["init"] = True
def add_app(self, name, command, slug):
calls["app"] = (name, command, slug)
async def run(self):
calls["run"] = True
monkeypatch.setattr(cli, "LocalServer", FakeServer)
monkeypatch.setattr(cli.asyncio, "run", lambda _coro: None)
runner = CliRunner()
result = runner.invoke(cli.app, ["--app", f"{app_file}:MyApp"])
assert result.exit_code == 0
assert calls["app"][0] == "MyApp"
assert "python3" in calls["app"][1]
def test_load_app_class_from_file(tmp_path):
from textual_webterm.cli import load_app_class
app_file = tmp_path / "myapp2.py"
app_file.write_text(
"""
class MyApp2:
pass
""".lstrip()
)
cls = load_app_class(f"{app_file}:MyApp2")
assert cls.__name__ == "MyApp2"
tests/test_config.py
+16 -18
View File
@@ -2,7 +2,9 @@
from __future__ import annotations
from textual_webterm.config import App, Config
import pytest
from webterm.config import App, Config
class TestApp:
@@ -21,13 +23,12 @@ class TestApp:
assert app.terminal is True
assert app.command == "bash"
def test_create_textual_app(self) -> None:
"""Test creating a Textual app configuration."""
def test_create_terminal_app_defaults(self) -> None:
"""Test creating a terminal app configuration with defaults."""
app = App(
name="My App",
slug="my-app",
terminal=False,
command="python -m myapp",
command="bash",
)
assert app.terminal is False
@@ -53,7 +54,7 @@ class TestDefaultConfig:
def test_default_config_returns_config(self):
"""Test that default_config returns a Config object."""
from textual_webterm.config import default_config
from webterm.config import default_config
config = default_config()
assert config is not None
@@ -63,27 +64,24 @@ class TestDefaultConfig:
class TestLoadConfig:
"""Tests for load_config function."""
def test_load_config_parses_app_and_terminal(self, tmp_path):
from textual_webterm.config import load_config
def test_load_config_parses_terminal_only(self, tmp_path):
from webterm.config import load_config
config_path = tmp_path / "config.toml"
config_path.write_text(
"""
[app.demo]
command = "echo demo"
[terminal.shell]
command = "bash"
""".lstrip()
)
config = load_config(config_path)
assert len(config.apps) == 2
assert {a.name for a in config.apps} == {"demo", "shell"}
assert len(config.apps) == 1
assert {a.name for a in config.apps} == {"shell"}
assert any(a.terminal for a in config.apps)
def test_load_config_slugify_for_app(self, tmp_path):
from textual_webterm.config import load_config
def test_load_config_rejects_app_entries(self, tmp_path):
from webterm.config import load_config
config_path = tmp_path / "config.toml"
config_path.write_text(
@@ -92,11 +90,11 @@ command = "bash"
command = "echo hi"
""".lstrip()
)
config = load_config(config_path)
assert config.apps[0].slug
with pytest.raises(ValueError):
load_config(config_path)
def test_load_config_expands_vars(self, tmp_path, monkeypatch):
from textual_webterm.config import load_config
from webterm.config import load_config
monkeypatch.setenv("MY_CMD", "echo expanded")
config_path = tmp_path / "config.toml"
tests/test_config_manifest.py
+1 -1
View File
@@ -1,7 +1,7 @@
import tempfile
from pathlib import Path
from textual_webterm.config import load_compose_manifest, load_landing_yaml
from webterm.config import load_compose_manifest, load_landing_yaml
def test_load_landing_yaml_simple():
tests/test_constants.py
+4 -4
View File
@@ -4,7 +4,7 @@ from __future__ import annotations
def test_get_environ_bool(monkeypatch):
from textual_webterm.constants import get_environ_bool
from webterm.constants import get_environ_bool
monkeypatch.setenv("FLAG", "1")
assert get_environ_bool("FLAG") is True
@@ -14,21 +14,21 @@ def test_get_environ_bool(monkeypatch):
def test_get_environ_int_keyerror(monkeypatch):
from textual_webterm.constants import get_environ_int
from webterm.constants import get_environ_int
monkeypatch.delenv("INT", raising=False)
assert get_environ_int("INT", 7) == 7
def test_get_environ_int_valueerror(monkeypatch):
from textual_webterm.constants import get_environ_int
from webterm.constants import get_environ_int
monkeypatch.setenv("INT", "not-an-int")
assert get_environ_int("INT", 7) == 7
def test_get_environ_int_valid(monkeypatch):
from textual_webterm.constants import get_environ_int
from webterm.constants import get_environ_int
monkeypatch.setenv("INT", "42")
assert get_environ_int("INT", 7) == 42
tests/test_docker_stats.py
+7 -7
View File
@@ -4,7 +4,7 @@ from unittest.mock import MagicMock
import pytest
from textual_webterm.docker_stats import (
from webterm.docker_stats import (
STATS_HISTORY_SIZE,
DockerStatsCollector,
render_sparkline_svg,
@@ -177,8 +177,8 @@ class TestLocalServerSparklineEndpoint:
"""Missing container param returns 400."""
from aiohttp.web import HTTPBadRequest
from textual_webterm.config import Config
from textual_webterm.local_server import LocalServer
from webterm.config import Config
from webterm.local_server import LocalServer
server = LocalServer("./", Config(), compose_mode=True)
@@ -191,8 +191,8 @@ class TestLocalServerSparklineEndpoint:
@pytest.mark.asyncio
async def test_sparkline_endpoint_returns_svg(self):
"""Sparkline endpoint returns SVG."""
from textual_webterm.config import Config
from textual_webterm.local_server import LocalServer
from webterm.config import Config
from webterm.local_server import LocalServer
server = LocalServer("./", Config(), compose_mode=True)
@@ -206,8 +206,8 @@ class TestLocalServerSparklineEndpoint:
@pytest.mark.asyncio
async def test_sparkline_with_stats_collector(self):
"""Sparkline uses stats collector data when available."""
from textual_webterm.config import Config
from textual_webterm.local_server import LocalServer
from webterm.config import Config
from webterm.local_server import LocalServer
server = LocalServer("./", Config(), compose_mode=True)
server._docker_stats = MagicMock()
tests/test_docker_watcher.py
+3 -3
View File
@@ -6,7 +6,7 @@ from unittest.mock import AsyncMock, MagicMock
import pytest
from textual_webterm.docker_watcher import DEFAULT_COMMAND, LABEL_NAME, DockerWatcher
from webterm.docker_watcher import DEFAULT_COMMAND, LABEL_NAME, DockerWatcher
@pytest.fixture
@@ -262,6 +262,6 @@ async def test_watch_events_recovers_from_errors(docker_watcher, monkeypatch):
async def fake_sleep(_seconds):
return None
monkeypatch.setattr("textual_webterm.docker_watcher.asyncio.open_unix_connection", fail_once)
monkeypatch.setattr("textual_webterm.docker_watcher.asyncio.sleep", fake_sleep)
monkeypatch.setattr("webterm.docker_watcher.asyncio.open_unix_connection", fail_once)
monkeypatch.setattr("webterm.docker_watcher.asyncio.sleep", fake_sleep)
await docker_watcher._watch_events()
tests/test_exit_poller.py
+4 -4
View File
@@ -5,8 +5,8 @@ import pytest
@pytest.mark.asyncio
async def test_exit_poller_noop_when_idle_wait_zero(monkeypatch):
from textual_webterm import exit_poller
from textual_webterm.exit_poller import ExitPoller
from webterm import exit_poller
from webterm.exit_poller import ExitPoller
monkeypatch.setattr(exit_poller, "EXIT_POLL_RATE", 0.01)
@@ -32,8 +32,8 @@ async def test_exit_poller_noop_when_idle_wait_zero(monkeypatch):
@pytest.mark.asyncio
async def test_exit_poller_resets_idle_timer_when_session_appears(monkeypatch):
from textual_webterm import exit_poller
from textual_webterm.exit_poller import ExitPoller
from webterm import exit_poller
from webterm.exit_poller import ExitPoller
monkeypatch.setattr(exit_poller, "EXIT_POLL_RATE", 0.01)
tests/test_local_server.py
+2 -2
View File
@@ -2,8 +2,8 @@
from __future__ import annotations
from textual_webterm.config import App, Config
from textual_webterm.local_server import WEBTERM_STATIC_PATH, LocalServer
from webterm.config import App, Config
from webterm.local_server import WEBTERM_STATIC_PATH, LocalServer
class TestLocalServer:
tests/test_local_server_unit.py
+11 -11
View File
@@ -5,8 +5,8 @@ from unittest.mock import AsyncMock, MagicMock
import pytest
from aiohttp import web
from textual_webterm.config import App, Config
from textual_webterm.local_server import (
from webterm.config import App, Config
from webterm.local_server import (
LocalServer,
)
@@ -16,20 +16,20 @@ class TestGetStaticPath:
def test_static_path_exists(self):
"""Test that static path exists."""
from textual_webterm.local_server import WEBTERM_STATIC_PATH
from webterm.local_server import WEBTERM_STATIC_PATH
assert WEBTERM_STATIC_PATH is not None and WEBTERM_STATIC_PATH.exists()
def test_static_path_has_js(self):
"""Test that static path has JS directory."""
from textual_webterm.local_server import WEBTERM_STATIC_PATH
from webterm.local_server import WEBTERM_STATIC_PATH
assert WEBTERM_STATIC_PATH is not None
assert (WEBTERM_STATIC_PATH / "js").exists()
def test_static_path_has_wasm(self):
"""Test that static path has WASM file."""
from textual_webterm.local_server import WEBTERM_STATIC_PATH
from webterm.local_server import WEBTERM_STATIC_PATH
assert WEBTERM_STATIC_PATH is not None
assert (WEBTERM_STATIC_PATH / "js" / "ghostty-vt.wasm").exists()
@@ -66,8 +66,8 @@ class TestLocalServer:
assert server.session_manager is not None
def test_add_app(self, server):
"""Test adding an app."""
server.add_app("New App", "python app.py", "newapp")
"""Test adding a terminal app."""
server.add_app("New Terminal", "bash", "newapp")
assert "newapp" in server.session_manager.apps_by_slug
def test_add_terminal(self, server):
@@ -79,7 +79,7 @@ class TestLocalServer:
@pytest.mark.asyncio
async def test_create_terminal_session_uses_slug_and_starts_session(self, server, monkeypatch):
from textual_webterm import local_server
from webterm import local_server
monkeypatch.setattr(local_server, "generate", lambda: "fixed-session")
@@ -404,7 +404,7 @@ class TestLocalServerMoreCoverage:
assert server_with_no_apps.exit_event.is_set()
def test_add_terminal_windows_noop(self, server_with_no_apps, monkeypatch):
from textual_webterm import constants as constants_mod
from webterm import constants as constants_mod
monkeypatch.setattr(constants_mod, "WINDOWS", True)
server_with_no_apps.add_terminal("T", "cmd", "slug")
@@ -542,7 +542,7 @@ class TestLocalServerMoreCoverage:
coro.close()
return MagicMock()
monkeypatch.setattr("textual_webterm.local_server.asyncio.create_task", fake_create_task)
monkeypatch.setattr("webterm.local_server.asyncio.create_task", fake_create_task)
server_with_no_apps.on_keyboard_interrupt()
assert fake_loop.call_soon_threadsafe.called
@@ -556,7 +556,7 @@ class TestLocalServerMoreCoverage:
):
from unittest.mock import MagicMock
from textual_webterm import local_server
from webterm import local_server
# Create a mock path that returns False for exists()
fake_path = MagicMock()
tests/test_local_server_websocket_integration.py
+3 -3
View File
@@ -9,9 +9,9 @@ import pytest
from aiohttp import WSMsgType, web
from aiohttp.test_utils import TestClient, TestServer
from textual_webterm.config import App, Config
from textual_webterm.local_server import LocalServer
from textual_webterm.types import RouteKey, SessionID
from webterm.config import App, Config
from webterm.local_server import LocalServer
from webterm.types import RouteKey, SessionID
if TYPE_CHECKING:
from collections.abc import AsyncIterator
tests/test_modules.py
+5 -5
View File
@@ -6,7 +6,7 @@ class TestConstants:
def test_import(self):
"""Test module can be imported."""
from textual_webterm import constants
from webterm import constants
assert constants is not None
@@ -14,7 +14,7 @@ class TestConstants:
"""Test DEBUG constant exists and respects env var."""
import importlib
from textual_webterm import constants
from webterm import constants
assert hasattr(constants, "DEBUG")
assert isinstance(constants.DEBUG, bool)
@@ -33,7 +33,7 @@ class TestExitPoller:
def test_import(self):
"""Test module can be imported."""
from textual_webterm.exit_poller import ExitPoller
from webterm.exit_poller import ExitPoller
assert ExitPoller is not None
@@ -41,8 +41,8 @@ class TestExitPoller:
"""ExitPoller should call force_exit after idle_wait seconds with no sessions."""
import asyncio
from textual_webterm import exit_poller
from textual_webterm.exit_poller import ExitPoller
from webterm import exit_poller
from webterm.exit_poller import ExitPoller
# Speed up the poll loop for the unit test.
monkeypatch.setattr(exit_poller, "EXIT_POLL_RATE", 0.01)
tests/test_poller.py
+1 -1
View File
@@ -6,7 +6,7 @@ from unittest.mock import MagicMock, patch
import pytest
from textual_webterm.poller import Poller, Write
from webterm.poller import Poller, Write
class TestWrite:
tests/test_session.py
+4 -4
View File
@@ -4,8 +4,8 @@ from __future__ import annotations
import pytest
from textual_webterm.session import Session, SessionConnector
from textual_webterm.types import RouteKey, SessionID
from webterm.session import Session, SessionConnector
from webterm.types import RouteKey, SessionID
class TestSessionConnector:
@@ -68,14 +68,14 @@ class TestIdentity:
def test_generate_unique_ids(self) -> None:
"""Test that generated IDs are unique."""
from textual_webterm.identity import generate
from webterm.identity import generate
ids = [generate() for _ in range(100)]
assert len(set(ids)) == 100 # All unique
def test_generate_id_format(self) -> None:
"""Test that generated IDs have expected format."""
from textual_webterm.identity import generate
from webterm.identity import generate
id_ = generate()
assert isinstance(id_, str)
tests/test_session_manager.py
+4 -23
View File
@@ -5,9 +5,9 @@ from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from textual_webterm.config import App
from textual_webterm.session_manager import SessionManager
from textual_webterm.types import RouteKey, SessionID
from webterm.config import App
from webterm.session_manager import SessionManager
from webterm.types import RouteKey, SessionID
class TestSessionManager:
@@ -173,7 +173,7 @@ class TestSessionManager:
@pytest.mark.skipif(platform.system() == "Windows", reason="Terminal not supported on Windows")
async def test_new_terminal_session(self, mock_poller, mock_path):
"""Test creating a new terminal session."""
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
app = App(name="Terminal", slug="term", path="./", command="echo test", terminal=True)
manager = SessionManager(mock_poller, mock_path, [app])
@@ -190,25 +190,6 @@ class TestSessionManager:
assert SessionID("test-session") in manager.sessions
assert RouteKey("test-route") in manager.routes
@pytest.mark.asyncio
async def test_new_app_session(self, mock_poller, mock_path):
"""Test creating a new app session."""
from textual_webterm.app_session import AppSession
app = App(name="App", slug="app", path="./", command="python app.py", terminal=False)
manager = SessionManager(mock_poller, mock_path, [app])
with patch.object(AppSession, "open", new_callable=AsyncMock):
result = await manager.new_session(
"app",
SessionID("test-session"),
RouteKey("test-route"),
)
assert result is not None
assert isinstance(result, AppSession)
class TestSessionManagerRoutes:
"""Tests for SessionManager route handling."""
tests/test_slugify.py
+1 -1
View File
@@ -1,6 +1,6 @@
"""Tests for slugify module."""
from textual_webterm.slugify import slugify
from webterm.slugify import slugify
class TestSlugify:
tests/test_svg_exporter.py
+1 -1
View File
@@ -4,7 +4,7 @@ from __future__ import annotations
import pytest
from textual_webterm.svg_exporter import (
from webterm.svg_exporter import (
ANSI_COLORS,
DEFAULT_BG,
DEFAULT_FG,
tests/test_terminal_session.py
+66 -57
View File
@@ -21,19 +21,19 @@ class TestTerminalSession:
def test_import(self):
"""Test that module can be imported."""
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
assert TerminalSession is not None
def test_replay_buffer_size(self):
"""Test replay buffer size constant."""
from textual_webterm.terminal_session import REPLAY_BUFFER_SIZE
from webterm.terminal_session import REPLAY_BUFFER_SIZE
assert REPLAY_BUFFER_SIZE == 256 * 1024 # 64KB
def test_init(self):
"""Test TerminalSession initialization."""
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
mock_poller = MagicMock()
session = TerminalSession(mock_poller, "test-session", "bash")
@@ -46,17 +46,26 @@ class TestTerminalSession:
def test_init_default_shell(self):
"""Test that default shell is used when command is empty."""
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
mock_poller = MagicMock()
with patch.dict(os.environ, {"SHELL": "/bin/zsh"}):
session = TerminalSession(mock_poller, "test-session", "")
assert session.command == "/bin/zsh"
def test_package_version_fallback(self):
from webterm.terminal_session import TerminalSession
with (
patch("webterm.terminal_session.version", side_effect=RuntimeError()),
patch("webterm.terminal_session.PackageNotFoundError", RuntimeError),
):
assert TerminalSession._package_version() == "0.0.0"
@pytest.mark.asyncio
async def test_replay_buffer_add(self):
"""Test adding data to replay buffer."""
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
mock_poller = MagicMock()
session = TerminalSession(mock_poller, "test-session", "bash")
@@ -68,7 +77,7 @@ class TestTerminalSession:
@pytest.mark.asyncio
async def test_replay_buffer_multiple_adds(self):
"""Test adding multiple chunks to replay buffer."""
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
mock_poller = MagicMock()
session = TerminalSession(mock_poller, "test-session", "bash")
@@ -80,7 +89,7 @@ class TestTerminalSession:
@pytest.mark.asyncio
async def test_replay_buffer_overflow(self):
"""Test that replay buffer trims old data when exceeding limit."""
from textual_webterm.terminal_session import (
from webterm.terminal_session import (
REPLAY_BUFFER_SIZE,
TerminalSession,
)
@@ -99,7 +108,7 @@ class TestTerminalSession:
@pytest.mark.asyncio
async def test_screen_state_updates_with_data(self):
"""Test that pyte screen updates when data is received."""
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
mock_poller = MagicMock()
session = TerminalSession(mock_poller, "test-session", "bash")
@@ -114,7 +123,7 @@ class TestTerminalSession:
@pytest.mark.asyncio
async def test_screen_handles_cursor_positioning(self):
"""Test that pyte screen correctly handles cursor positioning (tmux-style)."""
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
mock_poller = MagicMock()
session = TerminalSession(mock_poller, "test-session", "bash")
@@ -133,7 +142,7 @@ class TestTerminalSession:
@pytest.mark.asyncio
async def test_get_screen_state_returns_dirty_flag(self):
"""Test that get_screen_state returns has_changes flag based on pyte dirty tracking."""
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
mock_poller = MagicMock()
session = TerminalSession(mock_poller, "test-session", "bash")
@@ -158,7 +167,7 @@ class TestTerminalSession:
def test_update_connector(self):
"""Test updating connector."""
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
mock_poller = MagicMock()
session = TerminalSession(mock_poller, "test-session", "bash")
@@ -169,7 +178,7 @@ class TestTerminalSession:
def test_is_running_not_started(self):
"""Test is_running when session not started."""
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
mock_poller = MagicMock()
session = TerminalSession(mock_poller, "test-session", "bash")
@@ -179,7 +188,7 @@ class TestTerminalSession:
@pytest.mark.asyncio
async def test_send_bytes_no_fd(self):
"""Test send_bytes returns False when no master_fd."""
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
mock_poller = MagicMock()
session = TerminalSession(mock_poller, "test-session", "bash")
@@ -190,7 +199,7 @@ class TestTerminalSession:
@pytest.mark.asyncio
async def test_send_meta(self):
"""Test send_meta returns True."""
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
mock_poller = MagicMock()
session = TerminalSession(mock_poller, "test-session", "bash")
@@ -201,7 +210,7 @@ class TestTerminalSession:
@pytest.mark.asyncio
async def test_close_no_pid(self):
"""Test close when no pid."""
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
mock_poller = MagicMock()
session = TerminalSession(mock_poller, "test-session", "bash")
@@ -212,7 +221,7 @@ class TestTerminalSession:
@pytest.mark.asyncio
async def test_wait_no_task(self):
"""Test wait when no task."""
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
mock_poller = MagicMock()
session = TerminalSession(mock_poller, "test-session", "bash")
@@ -222,7 +231,7 @@ class TestTerminalSession:
def test_repr(self):
"""Test repr output."""
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
mock_poller = MagicMock()
session = TerminalSession(mock_poller, "test-session", "bash")
@@ -233,7 +242,7 @@ class TestTerminalSession:
@pytest.mark.asyncio
async def test_open_uses_shlex_split_and_execvp_with_args(self):
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
mock_poller = MagicMock()
command = 'echo "hello world"'
@@ -241,15 +250,15 @@ class TestTerminalSession:
with (
patch(
"textual_webterm.terminal_session.pty.fork", return_value=(pty.CHILD, 123)
"webterm.terminal_session.pty.fork", return_value=(pty.CHILD, 123)
) as mock_fork,
patch("textual_webterm.terminal_session.version", return_value="0.0.0"),
patch("textual_webterm.terminal_session.shlex.split", wraps=shlex.split) as mock_split,
patch("webterm.terminal_session.version", return_value="0.0.0"),
patch("webterm.terminal_session.shlex.split", wraps=shlex.split) as mock_split,
patch(
"textual_webterm.terminal_session.os.execvp", side_effect=OSError()
"webterm.terminal_session.os.execvp", side_effect=OSError()
) as mock_execvp,
patch(
"textual_webterm.terminal_session.os._exit", side_effect=SystemExit(1)
"webterm.terminal_session.os._exit", side_effect=SystemExit(1)
) as mock_exit,
pytest.raises(SystemExit),
):
@@ -262,13 +271,13 @@ class TestTerminalSession:
@pytest.mark.asyncio
async def test_open_parent_branch_sets_fd_and_pid(self):
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
poller = MagicMock()
session = TerminalSession(poller, "sid", "bash")
with (
patch("textual_webterm.terminal_session.pty.fork", return_value=(1234, 99)),
patch("webterm.terminal_session.pty.fork", return_value=(1234, 99)),
patch.object(session, "_set_terminal_size") as set_size,
):
await session.open(width=80, height=24)
@@ -279,16 +288,16 @@ class TestTerminalSession:
@pytest.mark.asyncio
async def test_open_bad_command_exits(self):
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
poller = MagicMock()
session = TerminalSession(poller, "sid", "bad")
with (
patch("textual_webterm.terminal_session.pty.fork", return_value=(pty.CHILD, 123)),
patch("textual_webterm.terminal_session.shlex.split", side_effect=ValueError("bad")),
patch("webterm.terminal_session.pty.fork", return_value=(pty.CHILD, 123)),
patch("webterm.terminal_session.shlex.split", side_effect=ValueError("bad")),
patch(
"textual_webterm.terminal_session.os._exit", side_effect=SystemExit(1)
"webterm.terminal_session.os._exit", side_effect=SystemExit(1)
) as mock_exit,
pytest.raises(SystemExit),
):
@@ -298,7 +307,7 @@ class TestTerminalSession:
@pytest.mark.asyncio
async def test_get_screen_lines_strips(self):
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
poller = MagicMock()
session = TerminalSession(poller, "sid", "bash")
@@ -319,7 +328,7 @@ class TestTerminalSession:
@pytest.mark.asyncio
async def test_get_screen_state_no_changes(self):
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
poller = MagicMock()
session = TerminalSession(poller, "sid", "bash")
@@ -352,7 +361,7 @@ class TestTerminalSession:
@pytest.mark.asyncio
async def test_get_screen_state_clears_dirty(self):
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
poller = MagicMock()
session = TerminalSession(poller, "sid", "bash")
@@ -389,7 +398,7 @@ class TestTerminalSession:
@pytest.mark.asyncio
async def test_get_screen_has_changes_reads_dirty(self):
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
poller = MagicMock()
session = TerminalSession(poller, "sid", "bash")
@@ -414,7 +423,7 @@ class TestTerminalSession:
@pytest.mark.asyncio
async def test_send_bytes_handles_closed_fd(self):
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
poller = MagicMock()
poller.write = AsyncMock(side_effect=KeyError)
@@ -426,7 +435,7 @@ class TestTerminalSession:
@pytest.mark.asyncio
async def test_run_reads_from_poller_and_closes(self):
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
queue: asyncio.Queue[bytes | None] = asyncio.Queue()
await queue.put(b"hello")
@@ -444,7 +453,7 @@ class TestTerminalSession:
session.master_fd = 10
session._connector = connector
with patch("textual_webterm.terminal_session.os.close") as mock_close:
with patch("webterm.terminal_session.os.close") as mock_close:
await session.run()
connector.on_data.assert_awaited_once_with(b"hello")
@@ -454,7 +463,7 @@ class TestTerminalSession:
@pytest.mark.asyncio
async def test_start_updates_connector_when_already_running(self):
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
poller = MagicMock()
session = TerminalSession(poller, "sid", "bash")
@@ -472,7 +481,7 @@ class TestTerminalSession:
@pytest.mark.asyncio
async def test_send_bytes_writes_via_poller(self):
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
poller = MagicMock()
poller.write = AsyncMock()
@@ -485,15 +494,15 @@ class TestTerminalSession:
@pytest.mark.asyncio
async def test_open_set_terminal_size_oserror_closes_fd_and_clears_master_fd(self):
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
poller = MagicMock()
session = TerminalSession(poller, "sid", "bash")
with (
patch("textual_webterm.terminal_session.pty.fork", return_value=(1234, 99)),
patch("webterm.terminal_session.pty.fork", return_value=(1234, 99)),
patch.object(session, "_set_terminal_size", side_effect=OSError("bad")),
patch("textual_webterm.terminal_session.os.close") as mock_close,
patch("webterm.terminal_session.os.close") as mock_close,
pytest.raises(OSError),
):
await session.open(width=80, height=24)
@@ -503,7 +512,7 @@ class TestTerminalSession:
@pytest.mark.asyncio
async def test_set_terminal_size_uses_executor(self):
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
poller = MagicMock()
session = TerminalSession(poller, "sid", "bash")
@@ -516,20 +525,20 @@ class TestTerminalSession:
run_in_executor.assert_awaited_once_with(None, session._set_terminal_size, 80, 24)
def test__set_terminal_size_calls_ioctl(self):
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
poller = MagicMock()
session = TerminalSession(poller, "sid", "bash")
session.master_fd = 10
with patch("textual_webterm.terminal_session.fcntl.ioctl") as mock_ioctl:
with patch("webterm.terminal_session.fcntl.ioctl") as mock_ioctl:
session._set_terminal_size(80, 24)
assert mock_ioctl.called
@pytest.mark.asyncio
async def test_start_creates_task_when_not_running(self):
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
poller = MagicMock()
session = TerminalSession(poller, "sid", "bash")
@@ -547,7 +556,7 @@ class TestTerminalSession:
@pytest.mark.asyncio
async def test_run_without_connector_still_closes(self):
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
queue: asyncio.Queue[bytes | None] = asyncio.Queue()
await queue.put(b"hello")
@@ -561,7 +570,7 @@ class TestTerminalSession:
session.master_fd = 10
session._connector = None
with patch("textual_webterm.terminal_session.os.close") as mock_close:
with patch("webterm.terminal_session.os.close") as mock_close:
await session.run()
poller.remove_file.assert_called_once_with(10)
@@ -569,7 +578,7 @@ class TestTerminalSession:
@pytest.mark.asyncio
async def test_run_oserror_still_closes(self):
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
queue = MagicMock()
queue.get = AsyncMock(side_effect=OSError("boom"))
@@ -582,7 +591,7 @@ class TestTerminalSession:
session.master_fd = 10
session._connector = None
with patch("textual_webterm.terminal_session.os.close") as mock_close:
with patch("webterm.terminal_session.os.close") as mock_close:
await session.run()
poller.remove_file.assert_called_once_with(10)
@@ -590,26 +599,26 @@ class TestTerminalSession:
@pytest.mark.asyncio
async def test_close_process_lookup_error_is_ignored(self):
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
poller = MagicMock()
session = TerminalSession(poller, "sid", "bash")
session.pid = 123
with patch("textual_webterm.terminal_session.os.kill", side_effect=ProcessLookupError()):
with patch("webterm.terminal_session.os.kill", side_effect=ProcessLookupError()):
await session.close()
@pytest.mark.asyncio
async def test_close_logs_warning_on_unexpected_exception(self):
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
poller = MagicMock()
session = TerminalSession(poller, "sid", "bash")
session.pid = 123
with (
patch("textual_webterm.terminal_session.os.kill", side_effect=RuntimeError("x")),
patch("textual_webterm.terminal_session.log.warning") as warn,
patch("webterm.terminal_session.os.kill", side_effect=RuntimeError("x")),
patch("webterm.terminal_session.log.warning") as warn,
):
await session.close()
@@ -617,7 +626,7 @@ class TestTerminalSession:
@pytest.mark.asyncio
async def test_wait_suppresses_cancelled_error(self):
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
poller = MagicMock()
session = TerminalSession(poller, "sid", "bash")
@@ -629,7 +638,7 @@ class TestTerminalSession:
await session.wait()
def test_is_running_false_when_kill_fails(self):
from textual_webterm.terminal_session import TerminalSession
from webterm.terminal_session import TerminalSession
poller = MagicMock()
session = TerminalSession(poller, "sid", "bash")
@@ -637,5 +646,5 @@ class TestTerminalSession:
session._task = MagicMock()
session.pid = 123
with patch("textual_webterm.terminal_session.os.kill", side_effect=OSError()):
with patch("webterm.terminal_session.os.kill", side_effect=OSError()):
assert session.is_running() is False
tests/test_two_way_dict.py
+1 -1
View File
@@ -4,7 +4,7 @@ from __future__ import annotations
import pytest
from textual_webterm._two_way_dict import TwoWayDict
from webterm._two_way_dict import TwoWayDict
class TestTwoWayDict: