diff --git a/.github/copilot-instructions.md b/.github/copilot-instructions.md
index 8599a1e..10905b2 100644
--- a/.github/copilot-instructions.md
+++ b/.github/copilot-instructions.md
@@ -1,4 +1,51 @@
-Use the Makefile to run linting, testing, and builds.
-Aim for good test coverage.
-Review tests periodically with a view to consolidate/parameterize and remove redundancy.
-Debug issues systematically. Search for and review documentation as needed.
+# Copilot Development Instructions
+
+## Makefile Usage (MANDATORY)
+
+Always use the Makefile for development tasks. Never run raw `pytest`, `ruff`, or `bun` commands directly.
+
+### Quick Reference
+
+| Task | Command | Description |
+|------|---------|-------------|
+| Run tests | `make test` | Run pytest |
+| Run linter | `make lint` | Run ruff linter |
+| Format code | `make format` | Auto-format with ruff |
+| Full check | `make check` | Run lint + coverage |
+| Coverage report | `make coverage` | Run pytest with coverage |
+| Build frontend | `make build` | TypeScript typecheck + bundle |
+| Quick frontend build | `make build-fast` | Bundle without typecheck |
+| Watch mode | `make bundle-watch` | Frontend dev with auto-rebuild |
+| Install dev deps | `make install-dev` | Install package + dev dependencies |
+| Clean build | `make build-all` | Full reproducible build from scratch |
+| See all targets | `make help` | Show all available commands |
+
+### Development Workflow
+
+1. **Before making changes**: Run `make check` to establish baseline
+2. **After making changes**: Run `make check` to verify no regressions
+3. **Frontend changes**: Use `make build` or `make build-fast`
+4. **Full rebuild**: Use `make build-all` (cleans everything first)
+
+### Clean Targets
+
+- `make clean` - Remove Python cache files only
+- `make bundle-clean` - Remove frontend build artifacts (node_modules, etc.)
+- `make clean-all` - Remove everything
+
+### Version Management
+
+- `make bump-patch` - Bump patch version and create git tag
+
+## Testing Guidelines
+
+- Aim for good test coverage
+- Review tests periodically to consolidate/parameterize and remove redundancy
+- Use fixtures from `tests/conftest.py` instead of duplicating setup code
+- Prefer parameterized tests for similar test cases
+
+## Code Style
+
+- Do not use heredocs or random shell commands
+- Prefer `make` and ecosystem tools (pip, bun) over manual operations
+- Debug issues systematically - search for and review documentation as needed
diff --git a/Makefile b/Makefile
index 3a68a30..d13539e 100644
--- a/Makefile
+++ b/Makefile
@@ -1,4 +1,4 @@
-.PHONY: help install install-dev lint format test coverage check clean clean-all build build-all bundle bundle-watch bundle-clean typecheck bump-patch
+.PHONY: help install install-dev lint format test coverage check clean clean-all build build-all build-fast bundle bundle-watch bundle-clean typecheck bump-patch push
PYTHON ?= python3
PIP ?= $(PYTHON) -m pip
@@ -6,92 +6,63 @@ PIP ?= $(PYTHON) -m pip
# Static assets
STATIC_JS_DIR = src/webterm/static/js
TERMINAL_JS = $(STATIC_JS_DIR)/terminal.js
-TERMINAL_TS = $(STATIC_JS_DIR)/terminal.ts
GHOSTTY_WASM = $(STATIC_JS_DIR)/ghostty-vt.wasm
-help:
- @echo "Build targets:"
- @echo " build-all - Full reproducible build (clean + deps + bundle + install)"
- @echo " build - Build frontend (typecheck + bundle)"
- @echo " build-fast - Build frontend without typecheck"
- @echo " bundle - Alias for build"
- @echo " bundle-watch - Watch mode for development"
- @echo " typecheck - Run TypeScript type checking"
- @echo ""
- @echo "Python targets:"
- @echo " install - Install package in editable mode"
- @echo " install-dev - Install with dev dependencies"
- @echo " lint - Run ruff linter"
- @echo " format - Format code with ruff"
- @echo " test - Run pytest"
- @echo " coverage - Run pytest with coverage"
- @echo " check - Run lint + coverage"
- @echo ""
- @echo "Clean targets:"
- @echo " clean - Remove Python cache files"
- @echo " bundle-clean - Remove frontend build artifacts"
- @echo " clean-all - Remove everything (clean + bundle-clean)"
+help: ## Show this help
+ @grep -E '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) | awk 'BEGIN {FS = ":.*?## "}; {printf " \033[36m%-14s\033[0m %s\n", $$1, $$2}'
# =============================================================================
# Full reproducible build
# =============================================================================
-build-all: clean-all node_modules build install-dev check
+build-all: clean-all node_modules build install-dev check ## Full reproducible build (clean + deps + bundle + install)
@echo "Build complete!"
# =============================================================================
# Python targets
# =============================================================================
-install:
+install: ## Install package in editable mode
$(PIP) install -e .
-install-dev:
- $(PIP) install -e .
+install-dev: install ## Install with dev dependencies
$(PIP) install pytest pytest-asyncio pytest-cov pytest-timeout ruff
-lint:
+lint: ## Run ruff linter
ruff check src tests
-format:
+format: ## Format code with ruff
ruff format src tests
-test:
+test: ## Run pytest
pytest
-coverage:
+coverage: ## Run pytest with coverage
pytest --cov=src/webterm --cov-report=term-missing
-check: lint coverage
+check: lint coverage ## Run lint + coverage
# =============================================================================
# Frontend build targets (requires Bun: https://bun.sh)
-# All frontend commands MUST go through bun run to ensure consistency
# =============================================================================
-# Install node dependencies (creates bun.lock if missing)
node_modules: package.json
bun install
@touch node_modules
-# TypeScript type checking
-typecheck: node_modules
+typecheck: node_modules ## Run TypeScript type checking
bun run typecheck
-# Main build target - typecheck + bundle + copy WASM
-build: node_modules
+build: node_modules ## Build frontend (typecheck + bundle)
bun run build
-# Fast build without typecheck (for rapid iteration)
-build-fast: node_modules
+build-fast: node_modules ## Build frontend without typecheck
bun run build:fast
@test -f $(GHOSTTY_WASM) || bun run copy-wasm
-# Alias for build
-bundle: build
+bundle: build ## Alias for build
-# Watch mode for development
-bundle-watch: node_modules
+bundle-watch: node_modules ## Watch mode for frontend development
@test -f $(GHOSTTY_WASM) || bun run copy-wasm
bun run watch
@@ -99,24 +70,29 @@ bundle-watch: node_modules
# Clean targets
# =============================================================================
-clean:
+clean: ## Remove Python cache files
rm -rf .pytest_cache .coverage htmlcov .ruff_cache __pycache__ src/**/__pycache__
-bundle-clean:
+bundle-clean: ## Remove frontend build artifacts
rm -rf node_modules bun.lock $(TERMINAL_JS) $(GHOSTTY_WASM)
-clean-all: clean bundle-clean
+clean-all: clean bundle-clean ## Remove everything (clean + bundle-clean)
# =============================================================================
# Version management
# =============================================================================
-# Bump patch version (e.g., 0.5.3 -> 0.5.4)
-bump-patch:
+bump-patch: ## Bump patch version and create git tag
@OLD=$$(grep -Po '(?<=^version = ")[^"]+' pyproject.toml); \
MAJOR=$$(echo $$OLD | cut -d. -f1); \
MINOR=$$(echo $$OLD | cut -d. -f2); \
PATCH=$$(echo $$OLD | cut -d. -f3); \
NEW="$$MAJOR.$$MINOR.$$((PATCH + 1))"; \
sed -i "s/^version = \"$$OLD\"/version = \"$$NEW\"/" pyproject.toml; \
- echo "Bumped version: $$OLD -> $$NEW"
+ git add pyproject.toml; \
+ git commit -m "Bump version to $$NEW"; \
+ git tag "v$$NEW"; \
+ echo "Bumped version: $$OLD -> $$NEW (tagged v$$NEW)"
+
+push: ## Push commits and tags to origin
+ git push origin main --tags
diff --git a/src/webterm/local_server.py b/src/webterm/local_server.py
index de41ef4..abc9ae3 100644
--- a/src/webterm/local_server.py
+++ b/src/webterm/local_server.py
@@ -554,18 +554,18 @@ class LocalServer:
if cached_response is not None:
return cached_response
- has_changes = await session_process.get_screen_has_changes() # type: ignore[union-attr]
- if not has_changes and cached is not None:
- cached_response = self._get_cached_screenshot_response(request, route_key)
- if cached_response is not None:
- return cached_response
-
+ # Use non-mutating snapshot method to avoid affecting terminal state
(
screen_width,
screen_height,
screen_buffer,
- _,
- ) = await session_process.get_screen_state() # type: ignore[union-attr]
+ has_changes,
+ ) = await session_process.get_screen_snapshot() # type: ignore[union-attr]
+
+ if not has_changes and cached is not None:
+ cached_response = self._get_cached_screenshot_response(request, route_key)
+ if cached_response is not None:
+ return cached_response
def _render_svg() -> str:
# Use custom SVG exporter - simpler and more reliable than Rich
diff --git a/src/webterm/terminal_session.py b/src/webterm/terminal_session.py
index fef6020..fd950a9 100644
--- a/src/webterm/terminal_session.py
+++ b/src/webterm/terminal_session.py
@@ -57,6 +57,9 @@ class TerminalSession(Session):
# Track last known terminal size for reconnection
self._last_width = DEFAULT_SCREEN_WIDTH
self._last_height = DEFAULT_SCREEN_HEIGHT
+ # Change counter for reliable activity detection (monotonically increasing)
+ self._change_counter = 0
+ self._last_snapshot_counter = 0
super().__init__()
def __repr__(self) -> str:
@@ -146,6 +149,8 @@ class TerminalSession(Session):
await loop.run_in_executor(None, self._set_terminal_size, width, height)
# Resize pyte screen to match
self._screen.resize(height, width)
+ # Increment change counter since resize changes screen content
+ self._change_counter += 1
async def force_redraw(self) -> None:
"""Force a terminal redraw by re-sending current size."""
@@ -169,6 +174,9 @@ class TerminalSession(Session):
try:
text = data.decode("utf-8", errors="replace")
self._stream.feed(text)
+ # Increment change counter when screen is modified
+ if self._screen.dirty:
+ self._change_counter += 1
except Exception:
# Don't let pyte errors crash the session
pass
@@ -188,14 +196,60 @@ class TerminalSession(Session):
return [line.rstrip() for line in self._screen.display]
async def get_screen_has_changes(self) -> bool:
- """Check if the screen has changed since the last snapshot."""
- await self._sync_pyte_to_pty()
+ """Check if the screen has changed since the last snapshot.
+
+ This is a non-mutating read-only check that compares the change counter.
+ """
async with self._screen_lock:
- return len(self._screen.dirty) > 0
+ return self._change_counter > self._last_snapshot_counter
+
+ async def get_screen_snapshot(self) -> tuple[int, int, list, bool]:
+ """Get a read-only snapshot of the current screen state for screenshots.
+
+ This method does NOT mutate terminal state - safe for dashboard screenshots.
+
+ Returns:
+ Tuple of (width, height, buffer, has_changes) where:
+ - width: screen width in columns
+ - height: screen height in rows
+ - buffer: list of rows, each containing character data with styling
+ - has_changes: True if screen has changed since last snapshot
+ """
+ async with self._screen_lock:
+ width = self._screen.columns
+ height = self._screen.lines
+ has_changes = self._change_counter > self._last_snapshot_counter
+ # Update the snapshot counter to track what we've seen
+ self._last_snapshot_counter = self._change_counter
+ # Snapshot buffer cells quickly to minimize lock hold time
+ snapshot = [
+ [self._screen.buffer[row][col] for col in range(width)] for row in range(height)
+ ]
+
+ buffer = []
+ for row_data in snapshot:
+ row_chars = []
+ for char in row_data:
+ row_chars.append(
+ {
+ "data": char.data if char.data else " ",
+ "fg": char.fg,
+ "bg": char.bg,
+ "bold": char.bold,
+ "italics": char.italics,
+ "underscore": char.underscore,
+ "reverse": char.reverse,
+ }
+ )
+ buffer.append(row_chars)
+ return (width, height, buffer, has_changes)
async def get_screen_state(self) -> tuple[int, int, list, bool]:
"""Get the current screen state including dimensions and character buffer.
+ Note: This method syncs pyte to PTY size and clears dirty flags.
+ For read-only screenshot access, use get_screen_snapshot() instead.
+
Returns:
Tuple of (width, height, buffer, has_changes) where:
- width: screen width in columns
diff --git a/tests/conftest.py b/tests/conftest.py
index f721668..40b6b67 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -86,6 +86,7 @@ def mock_session():
session = MagicMock()
session.get_screen_has_changes = AsyncMock(return_value=False)
session.get_screen_state = AsyncMock(return_value=(80, 24, [], True))
+ session.get_screen_snapshot = AsyncMock(return_value=(80, 24, [], True))
return session
@@ -95,6 +96,54 @@ def poller() -> Poller:
return Poller()
+@pytest.fixture
+def mock_poller() -> MagicMock:
+ """Create a mock Poller for unit tests."""
+ return MagicMock()
+
+
+class DummyAsyncLock:
+ """A dummy async context manager for replacing locks in tests."""
+
+ async def __aenter__(self):
+ return None
+
+ async def __aexit__(self, exc_type, exc, tb):
+ return False
+
+
+@pytest.fixture
+def dummy_lock() -> DummyAsyncLock:
+ """Create a dummy async lock for tests."""
+ return DummyAsyncLock()
+
+
+@pytest.fixture
+def mock_screen_char():
+ """Factory for creating mock pyte screen characters."""
+
+ def _make(
+ data: str = " ",
+ fg: int = 0,
+ bg: int = 0,
+ bold: bool = False,
+ italics: bool = False,
+ underscore: bool = False,
+ reverse: bool = False,
+ ) -> MagicMock:
+ char = MagicMock()
+ char.data = data
+ char.fg = fg
+ char.bg = bg
+ char.bold = bold
+ char.italics = italics
+ char.underscore = underscore
+ char.reverse = reverse
+ return char
+
+ return _make
+
+
@pytest.fixture
def session_manager(poller: Poller, tmp_path: Path, sample_terminal_app: App) -> SessionManager:
"""Create a SessionManager instance."""
diff --git a/tests/test_local_server_unit.py b/tests/test_local_server_unit.py
index d661b0c..95a68a8 100644
--- a/tests/test_local_server_unit.py
+++ b/tests/test_local_server_unit.py
@@ -181,7 +181,7 @@ class TestLocalServerHelpers:
request.query = {"route_key": "rk"}
screen_buffer = screen_buffer_factory(["hello", ""])
- mock_session.get_screen_state = AsyncMock(return_value=(80, 2, screen_buffer, True))
+ mock_session.get_screen_snapshot = AsyncMock(return_value=(80, 2, screen_buffer, True))
monkeypatch.setattr(
server.session_manager, "get_session_by_route_key", lambda _rk: mock_session
@@ -203,7 +203,7 @@ class TestLocalServerHelpers:
request.query = {"route_key": "known"}
screen_buffer = screen_buffer_factory(["world", ""])
- mock_session.get_screen_state = AsyncMock(return_value=(80, 2, screen_buffer, True))
+ mock_session.get_screen_snapshot = AsyncMock(return_value=(80, 2, screen_buffer, True))
# Pretend app exists for slug "known"
server.session_manager.apps_by_slug["known"] = App(
@@ -639,7 +639,7 @@ class TestLocalServerMoreCoverage:
async def test_handle_screenshot_uses_cached_when_no_changes(
self, server_with_no_apps, monkeypatch, mock_request, mock_session
):
- mock_session.get_screen_state = AsyncMock(return_value=(80, 24, [], False))
+ mock_session.get_screen_snapshot = AsyncMock(return_value=(80, 24, [], False))
monkeypatch.setattr(
server_with_no_apps.session_manager,
"get_session_by_route_key",
@@ -655,18 +655,17 @@ class TestLocalServerMoreCoverage:
resp = await server_with_no_apps._handle_screenshot(request)
assert resp.text == ""
- mock_session.get_screen_state.assert_not_awaited()
@pytest.mark.asyncio
async def test_handle_screenshot_uses_screen_state(
self, server_with_no_apps, monkeypatch, screen_buffer_factory, mock_request, mock_session
):
- """Test that screenshot uses get_screen_state for rendering."""
+ """Test that screenshot uses get_screen_snapshot for rendering."""
request = mock_request
request.query = {"route_key": "rk"}
screen_buffer = screen_buffer_factory(["line1", "line2"])
- mock_session.get_screen_state = AsyncMock(return_value=(80, 2, screen_buffer, True))
+ mock_session.get_screen_snapshot = AsyncMock(return_value=(80, 2, screen_buffer, True))
monkeypatch.setattr(
server_with_no_apps.session_manager,
"get_session_by_route_key",
@@ -678,7 +677,7 @@ class TestLocalServerMoreCoverage:
resp = await server_with_no_apps._handle_screenshot(request)
assert resp.content_type == "image/svg+xml"
assert " initial_counter
+
+ @pytest.mark.asyncio
+ async def test_set_terminal_size_increments_change_counter(self, terminal_session, mock_poller):
+ """Test that set_terminal_size increments change counter."""
+ terminal_session.master_fd = 10
+ initial_counter = terminal_session._change_counter
+
+ loop = asyncio.get_running_loop()
+ with patch.object(loop, "run_in_executor", new=AsyncMock()):
+ await terminal_session.set_terminal_size(100, 50)
+
+ assert terminal_session._change_counter == initial_counter + 1
+
+ @pytest.mark.asyncio
+ async def test_get_screen_has_changes_uses_change_counter(self, terminal_session, dummy_lock):
+ """Test that get_screen_has_changes uses the change counter."""
+ terminal_session._screen_lock = dummy_lock
+
+ # Initially no changes
+ terminal_session._change_counter = 0
+ terminal_session._last_snapshot_counter = 0
+ assert await terminal_session.get_screen_has_changes() is False
+
+ # After screen update increments counter
+ terminal_session._change_counter = 1
+ assert await terminal_session.get_screen_has_changes() is True
+
+ # After snapshot resets detection
+ terminal_session._last_snapshot_counter = 1
+ assert await terminal_session.get_screen_has_changes() is False
+
+ @pytest.mark.asyncio
+ async def test_send_bytes_handles_closed_fd(self, mock_poller):
+ mock_poller.write = AsyncMock(side_effect=KeyError)
+ session = TerminalSession(mock_poller, "sid", "bash")
session.master_fd = 10
ok = await session.send_bytes(b"test")
assert ok is False
@pytest.mark.asyncio
- async def test_run_reads_from_poller_and_closes(self):
- from webterm.terminal_session import TerminalSession
-
+ async def test_run_reads_from_poller_and_closes(self, mock_poller):
queue: asyncio.Queue[bytes | None] = asyncio.Queue()
await queue.put(b"hello")
await queue.put(None)
- poller = MagicMock()
- poller.add_file = MagicMock(return_value=queue)
- poller.remove_file = MagicMock()
+ mock_poller.add_file = MagicMock(return_value=queue)
+ mock_poller.remove_file = MagicMock()
connector = MagicMock()
connector.on_data = AsyncMock()
connector.on_close = AsyncMock()
- session = TerminalSession(poller, "sid", "bash")
+ session = TerminalSession(mock_poller, "sid", "bash")
session.master_fd = 10
session._connector = connector
@@ -458,15 +381,12 @@ class TestTerminalSession:
connector.on_data.assert_awaited_once_with(b"hello")
connector.on_close.assert_awaited_once()
- poller.remove_file.assert_called_once_with(10)
+ mock_poller.remove_file.assert_called_once_with(10)
mock_close.assert_called_once_with(10)
@pytest.mark.asyncio
- async def test_start_updates_connector_when_already_running(self):
- from webterm.terminal_session import TerminalSession
-
- poller = MagicMock()
- session = TerminalSession(poller, "sid", "bash")
+ async def test_start_updates_connector_when_already_running(self, mock_poller):
+ session = TerminalSession(mock_poller, "sid", "bash")
session.master_fd = 10
existing = asyncio.create_task(asyncio.sleep(0))
@@ -480,24 +400,18 @@ class TestTerminalSession:
await existing
@pytest.mark.asyncio
- async def test_send_bytes_writes_via_poller(self):
- from webterm.terminal_session import TerminalSession
+ async def test_send_bytes_writes_via_poller(self, mock_poller):
+ mock_poller.write = AsyncMock()
- poller = MagicMock()
- poller.write = AsyncMock()
-
- session = TerminalSession(poller, "sid", "bash")
+ session = TerminalSession(mock_poller, "sid", "bash")
session.master_fd = 10
assert await session.send_bytes(b"x") is True
- poller.write.assert_awaited_once_with(10, b"x")
+ mock_poller.write.assert_awaited_once_with(10, b"x")
@pytest.mark.asyncio
- async def test_open_set_terminal_size_oserror_closes_fd_and_clears_master_fd(self):
- from webterm.terminal_session import TerminalSession
-
- poller = MagicMock()
- session = TerminalSession(poller, "sid", "bash")
+ async def test_open_set_terminal_size_oserror_closes_fd_and_clears_master_fd(self, mock_poller):
+ session = TerminalSession(mock_poller, "sid", "bash")
with (
patch("webterm.terminal_session.pty.fork", return_value=(1234, 99)),
@@ -511,11 +425,8 @@ class TestTerminalSession:
assert session.master_fd is None
@pytest.mark.asyncio
- async def test_set_terminal_size_uses_executor(self):
- from webterm.terminal_session import TerminalSession
-
- poller = MagicMock()
- session = TerminalSession(poller, "sid", "bash")
+ async def test_set_terminal_size_uses_executor(self, mock_poller):
+ session = TerminalSession(mock_poller, "sid", "bash")
session.master_fd = 10
loop = asyncio.get_running_loop()
@@ -524,11 +435,8 @@ 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 webterm.terminal_session import TerminalSession
-
- poller = MagicMock()
- session = TerminalSession(poller, "sid", "bash")
+ def test__set_terminal_size_calls_ioctl(self, mock_poller):
+ session = TerminalSession(mock_poller, "sid", "bash")
session.master_fd = 10
with patch("webterm.terminal_session.fcntl.ioctl") as mock_ioctl:
@@ -537,11 +445,8 @@ class TestTerminalSession:
assert mock_ioctl.called
@pytest.mark.asyncio
- async def test_start_creates_task_when_not_running(self):
- from webterm.terminal_session import TerminalSession
-
- poller = MagicMock()
- session = TerminalSession(poller, "sid", "bash")
+ async def test_start_creates_task_when_not_running(self, mock_poller):
+ session = TerminalSession(mock_poller, "sid", "bash")
session.master_fd = 10
session.run = AsyncMock() # type: ignore[method-assign]
@@ -555,65 +460,53 @@ class TestTerminalSession:
session.run.assert_awaited_once()
@pytest.mark.asyncio
- async def test_run_without_connector_still_closes(self):
- from webterm.terminal_session import TerminalSession
-
+ async def test_run_without_connector_still_closes(self, mock_poller):
queue: asyncio.Queue[bytes | None] = asyncio.Queue()
await queue.put(b"hello")
await queue.put(None)
- poller = MagicMock()
- poller.add_file = MagicMock(return_value=queue)
- poller.remove_file = MagicMock()
+ mock_poller.add_file = MagicMock(return_value=queue)
+ mock_poller.remove_file = MagicMock()
- session = TerminalSession(poller, "sid", "bash")
+ session = TerminalSession(mock_poller, "sid", "bash")
session.master_fd = 10
session._connector = None
with patch("webterm.terminal_session.os.close") as mock_close:
await session.run()
- poller.remove_file.assert_called_once_with(10)
+ mock_poller.remove_file.assert_called_once_with(10)
mock_close.assert_called_once_with(10)
@pytest.mark.asyncio
- async def test_run_oserror_still_closes(self):
- from webterm.terminal_session import TerminalSession
-
+ async def test_run_oserror_still_closes(self, mock_poller):
queue = MagicMock()
queue.get = AsyncMock(side_effect=OSError("boom"))
- poller = MagicMock()
- poller.add_file = MagicMock(return_value=queue)
- poller.remove_file = MagicMock()
+ mock_poller.add_file = MagicMock(return_value=queue)
+ mock_poller.remove_file = MagicMock()
- session = TerminalSession(poller, "sid", "bash")
+ session = TerminalSession(mock_poller, "sid", "bash")
session.master_fd = 10
session._connector = None
with patch("webterm.terminal_session.os.close") as mock_close:
await session.run()
- poller.remove_file.assert_called_once_with(10)
+ mock_poller.remove_file.assert_called_once_with(10)
mock_close.assert_called_once_with(10)
@pytest.mark.asyncio
- async def test_close_process_lookup_error_is_ignored(self):
- from webterm.terminal_session import TerminalSession
-
- poller = MagicMock()
- session = TerminalSession(poller, "sid", "bash")
+ async def test_close_process_lookup_error_is_ignored(self, mock_poller):
+ session = TerminalSession(mock_poller, "sid", "bash")
session.pid = 123
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 webterm.terminal_session import TerminalSession
-
- poller = MagicMock()
- session = TerminalSession(poller, "sid", "bash")
+ async def test_close_logs_warning_on_unexpected_exception(self, mock_poller):
+ session = TerminalSession(mock_poller, "sid", "bash")
session.pid = 123
with (
@@ -625,11 +518,8 @@ class TestTerminalSession:
assert warn.called
@pytest.mark.asyncio
- async def test_wait_suppresses_cancelled_error(self):
- from webterm.terminal_session import TerminalSession
-
- poller = MagicMock()
- session = TerminalSession(poller, "sid", "bash")
+ async def test_wait_suppresses_cancelled_error(self, mock_poller):
+ session = TerminalSession(mock_poller, "sid", "bash")
task = asyncio.create_task(asyncio.sleep(10))
task.cancel()
@@ -637,11 +527,8 @@ class TestTerminalSession:
await session.wait()
- def test_is_running_false_when_kill_fails(self):
- from webterm.terminal_session import TerminalSession
-
- poller = MagicMock()
- session = TerminalSession(poller, "sid", "bash")
+ def test_is_running_false_when_kill_fails(self, mock_poller):
+ session = TerminalSession(mock_poller, "sid", "bash")
session.master_fd = 10
session._task = MagicMock()
session.pid = 123