jihoson/LazyEnter
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: jihoson:feat/add-codex-bridge-support
Branches Tags
jihoson:main jihoson:feat/unify-start-stop-args jihoson:fix/output-throttle jihoson:fix/slack-input-drop-logs jihoson:fix/codex-tui-noise-filter jihoson:chore/record-pr-workflow-safety jihoson:feat/add-codex-bridge-support jihoson:chore/gitea-workflow-notes jihoson:feat/tmux-attach-bridge jihoson:fix/slack-ansi-filter jihoson:feat/remoteenter-hardening jihoson:docs/contributor-guide-and-readme
..
compare: jihoson:feat/tmux-attach-bridge
Branches Tags
jihoson:main jihoson:feat/unify-start-stop-args jihoson:fix/output-throttle jihoson:fix/slack-input-drop-logs jihoson:fix/codex-tui-noise-filter jihoson:chore/record-pr-workflow-safety jihoson:feat/add-codex-bridge-support jihoson:chore/gitea-workflow-notes jihoson:feat/tmux-attach-bridge jihoson:fix/slack-ansi-filter jihoson:feat/remoteenter-hardening jihoson:docs/contributor-guide-and-readme

1 Commits
feat/add-c ... feat/tmux-

Author SHA1 Message Date
jihoson
119dd9a486 Reduce tmux TUI noise with pane capture 2026-02-17 04:13:05 +09:00
12 changed files with 147 additions and 304 deletions
Expand all files Collapse all files

1
.env.example
View File

@@ -8,7 +8,6 @@ SLACK_ALLOWED_CHANNEL_ID=C0XXXXXXXXX
# PTY / tmux 연결 설정 # PTY / tmux 연결 설정
TMUX_SESSION_NAME=claude TMUX_SESSION_NAME=claude
CODEX_TMUX_SESSION_NAME=codex
PTY_READ_TIMEOUT=5 PTY_READ_TIMEOUT=5
# 출력 버퍼 설정 # 출력 버퍼 설정

32
AGENTS.md
View File

@@ -42,35 +42,3 @@ Current git history is minimal (`Initial commit`), so enforce clear conventions
## Security & Configuration Tips ## Security & Configuration Tips
Do not commit secrets. Copy `.env.example` to `.env` locally. Do not commit secrets. Copy `.env.example` to `.env` locally.
Restrict Slack access with `SLACK_ALLOWED_USER_ID` and `SLACK_ALLOWED_CHANNEL_ID` before running in shared workspaces. Restrict Slack access with `SLACK_ALLOWED_USER_ID` and `SLACK_ALLOWED_CHANNEL_ID` before running in shared workspaces.
## Git & Gitea Workflow Notes
Use `tea` (Gitea CLI), not `gh`.
Hard rule:
- Never implement changes, stage files, or commit on `main`.
- Always create/use a feature branch first, and merge via PR.
- Check login:
- `tea login ls`
- `tea whoami`
- Create/update feature branch:
- `git checkout -b <branch>` (or `git checkout <branch>`)
- `git add <files>`
- `git commit -m "<message>"`
- Push to Gitea:
- Normal: `git push -u origin <branch>`
- If HTTP username prompt fails in this environment, use the token from `~/.config/tea/config.yml`:
- `TOKEN=$(sed -n 's/^[[:space:]]*token: //p' ~/.config/tea/config.yml | head -n1)`
- `git push "http://agentson:${TOKEN}@localhost:3000/jihoson/LazyEnter.git" <branch>`
- After token-based push, ensure tracking is on `origin/<branch>` (not token URL):
- `git fetch origin <branch>:refs/remotes/origin/<branch>`
- `git branch --set-upstream-to=origin/<branch> <branch>`
- Create PR on Gitea:
- `tea pr create --base main --head <branch> --title "<title>" --description "<body>"`
- Sync local main:
- `git checkout main`
- `git pull --ff-only`
Safety:
- Do not commit or print tokens in logs/docs.
- Keep unrelated local files (for example `uv.lock`) out of PRs unless intentionally changed.

25
README.md
View File

@@ -1,15 +1,13 @@
# LazyEnter # LazyEnter
소파에서 엔터나 치자. 소파에서 엔터나 치자.
로컬 PC의 Claude/Codex CLI 세션을 Slack Socket Mode로 중계해 모바일에서도 원격으로 입력/승인을 처리하는 브릿지입니다. 로컬 PC의 Claude Code CLI 세션을 Slack Socket Mode로 중계해 모바일에서도 원격으로 입력/승인을 처리하는 브릿지입니다.
## 동작 방식 ## 동작 방식
1. 로컬에서 Claude 또는 Codex를 tmux 세션으로 미리 실행 1. 로컬에서 Claude를 tmux 세션으로 미리 실행 (`tmux new -s claude claude`)
- Claude: `tmux new -s claude claude` 2. Slack에서 `/start-claude` 실행
- Codex: `tmux new -s codex codex` 3. 브릿지가 기존 tmux 세션에 연결해 pane 출력/입력을 중계
2. Slack에서 `/start-claude` 또는 `/start-codex` 실행
3. 브릿지가 기존 tmux 세션에 attach
4. Slack 채널 메시지가 CLI 입력으로 전달됨 4. Slack 채널 메시지가 CLI 입력으로 전달됨
5. CLI 출력이 Slack으로 다시 전송됨 5. CLI 출력이 Slack으로 다시 전송됨
6. `/stop-claude`로 브릿지 연결 해제 (tmux 세션은 유지) 6. `/stop-claude`로 브릿지 연결 해제 (tmux 세션은 유지)
@@ -37,8 +35,7 @@ cp .env.example .env
- `SLACK_ALLOWED_CHANNEL_ID` - `SLACK_ALLOWED_CHANNEL_ID`
선택 환경 변수: 선택 환경 변수:
- `TMUX_SESSION_NAME` (기본: `claude`, `/start-claude` 대상) - `TMUX_SESSION_NAME` (기본: `claude`)
- `CODEX_TMUX_SESSION_NAME` (기본: `codex`, `/start-codex` 대상)
- `PTY_READ_TIMEOUT` (기본: `5`) - `PTY_READ_TIMEOUT` (기본: `5`)
- `OUTPUT_BUFFER_INTERVAL` (기본: `2.0`) - `OUTPUT_BUFFER_INTERVAL` (기본: `2.0`)
- `MAX_MESSAGE_LENGTH` (기본: `3000`) - `MAX_MESSAGE_LENGTH` (기본: `3000`)
@@ -55,18 +52,14 @@ cp .env.example .env
- Slash Commands 생성: - Slash Commands 생성:
- `/start-claude` - `/start-claude`
- `/stop-claude` - `/stop-claude`
- `/start-codex`
- `/stop-codex`
- 앱을 워크스페이스에 설치 후 토큰을 `.env`에 반영 - 앱을 워크스페이스에 설치 후 토큰을 `.env`에 반영
## 실행 ## 실행
먼저 로컬에서 원하는 tmux 세션을 실행: 먼저 로컬에서 Claude tmux 세션을 실행:
```bash ```bash
tmux new -s claude claude tmux new -s claude claude
# 또는
tmux new -s codex codex
``` ```
그 다음 브릿지 실행: 그 다음 브릿지 실행:
@@ -78,9 +71,9 @@ python -m lazy_enter
``` ```
실행 후 Slack의 허용된 채널에서: 실행 후 Slack의 허용된 채널에서:
- `/start-claude`, `/start-codex`: 기존 세션에 연결 - `/start-claude`: 기존 세션에 연결
- 일반 메시지 전송: 현재 연결된 CLI(Claude/Codex)로 입력 전달 - 일반 메시지 전송: Claude CLI로 입력 전달
- `/stop-claude`, `/stop-codex`: 브릿지 연결 해제 (세션 유지) - `/stop-claude`: 브릿지 연결 해제 (세션 유지)
## 테스트 및 품질 점검 ## 테스트 및 품질 점검

1
pyproject.toml
View File

@@ -11,7 +11,6 @@ authors = [
dependencies = [ dependencies = [
"slack-bolt>=1.21.0", "slack-bolt>=1.21.0",
"slack-sdk>=3.33.0", "slack-sdk>=3.33.0",
"pexpect>=4.9.0",
"python-dotenv>=1.0.0", "python-dotenv>=1.0.0",
] ]

2
src/lazy_enter/__init__.py
View File

@@ -1,3 +1,3 @@
"""LazyEnter - Slack을 통한 Claude/Codex 원격 제어 브릿지.""" """LazyEnter - Slack을 통한 Claude Code 원격 제어 브릿지."""
__version__ = "0.1.0" __version__ = "0.1.0"

40
src/lazy_enter/bridge.py
View File

@@ -25,7 +25,6 @@ class Bridge:
self._output_thread: threading.Thread | None = None self._output_thread: threading.Thread | None = None
self._running = False self._running = False
self._channel: str = self.config.allowed_channel_id self._channel: str = self.config.allowed_channel_id
self._active_target: str | None = None
self._last_sent_output: str = "" self._last_sent_output: str = ""
self._last_input_at = time.monotonic() self._last_input_at = time.monotonic()
self._last_output_at = time.monotonic() self._last_output_at = time.monotonic()
@@ -66,30 +65,15 @@ class Bridge:
) )
return any(pattern in normalized for pattern in blocked_patterns) return any(pattern in normalized for pattern in blocked_patterns)
@staticmethod def _handle_command(self, command: str, channel: str) -> None:
def _display_name(target: str) -> str:
if target == "codex":
return "Codex"
return "Claude"
def _session_name_for_target(self, target: str) -> str:
if target == "codex":
return self.config.codex_tmux_session_name
return self.config.tmux_session_name
def _handle_command(self, action: str, target: str, channel: str) -> None:
"""슬래시 커맨드를 처리한다.""" """슬래시 커맨드를 처리한다."""
if target not in {"claude", "codex"}: if command == "start":
self.slack.send_message(channel, ":warning: 지원하지 않는 대상입니다.") self._start_session(channel)
return elif command == "stop":
if action == "start":
self._start_session(channel, target)
elif action == "stop":
self._stop_session(channel) self._stop_session(channel)
def _start_session(self, channel: str, target: str) -> None: def _start_session(self, channel: str) -> None:
"""지정한 대상의 tmux 세션에 연결한다.""" """기존 Claude tmux 세션에 연결한다."""
if self.pty and self.pty.is_alive: if self.pty and self.pty.is_alive:
self.slack.send_message( self.slack.send_message(
channel, ":information_source: 이미 세션에 연결되어 있습니다." channel, ":information_source: 이미 세션에 연결되어 있습니다."
@@ -97,12 +81,8 @@ class Bridge:
return return
self._channel = channel self._channel = channel
self._active_target = target
self._last_sent_output = "" self._last_sent_output = ""
self.pty = PtyManager( self.pty = PtyManager(self.config.tmux_session_name)
self._session_name_for_target(target),
cli_name=target,
)
try: try:
self.pty.start() self.pty.start()
except RuntimeError as exc: except RuntimeError as exc:
@@ -118,10 +98,7 @@ class Bridge:
self._output_thread = threading.Thread(target=self._poll_output, daemon=True) self._output_thread = threading.Thread(target=self._poll_output, daemon=True)
self._output_thread.start() self._output_thread.start()
display_name = self._display_name(target) self.slack.send_message(channel, ":link: Claude 세션에 연결되었습니다.")
self.slack.send_message(
channel, f":link: {display_name} 세션에 연결되었습니다."
)
def _stop_session(self, channel: str) -> None: def _stop_session(self, channel: str) -> None:
"""브릿지 연결만 해제한다.""" """브릿지 연결만 해제한다."""
@@ -129,7 +106,6 @@ class Bridge:
if self.pty: if self.pty:
self.pty.stop() self.pty.stop()
self.pty = None self.pty = None
self._active_target = None
self._last_sent_output = "" self._last_sent_output = ""
self.slack.send_message(channel, ":electric_plug: 세션 연결이 해제되었습니다.") self.slack.send_message(channel, ":electric_plug: 세션 연결이 해제되었습니다.")

1
src/lazy_enter/config.py
View File

@@ -20,7 +20,6 @@ class Config:
# PTY / tmux attach # PTY / tmux attach
tmux_session_name: str = os.getenv("TMUX_SESSION_NAME", "claude") tmux_session_name: str = os.getenv("TMUX_SESSION_NAME", "claude")
codex_tmux_session_name: str = os.getenv("CODEX_TMUX_SESSION_NAME", "codex")
pty_read_timeout: int = int(os.getenv("PTY_READ_TIMEOUT", "5")) pty_read_timeout: int = int(os.getenv("PTY_READ_TIMEOUT", "5"))
# Buffer # Buffer

105
src/lazy_enter/pty_manager.py
View File

@@ -1,74 +1,105 @@
"""기존 tmux 세션에 attach하여 CLI 입출력을 중계한다.""" """기존 tmux 세션의 pane을 캡처/입력하여 CLI 입출력을 중계한다."""
from __future__ import annotations from __future__ import annotations
import logging import logging
import subprocess import subprocess
import pexpect
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
class PtyManager: class PtyManager:
"""기존 tmux 세션에 attach하고 입출력을 제어한다.""" """기존 tmux 세션의 pane을 읽고 입력을 보낸다."""
def __init__(self, session_name: str = "claude", cli_name: str = "claude") -> None: def __init__(self, session_name: str = "claude") -> None:
self.session_name = session_name self.session_name = session_name
self.cli_name = cli_name self._connected = False
self._process: pexpect.spawn | None = None self._last_capture: list[str] = []
@property @property
def is_alive(self) -> bool: def is_alive(self) -> bool:
return self._process is not None and self._process.isalive() return self._connected and self._session_exists()
def _ensure_session_exists(self) -> None: def _run_tmux(self, args: list[str]) -> subprocess.CompletedProcess[str]:
"""attach 대상 tmux 세션 존재 여부를 검증한다.""" return subprocess.run(
result = subprocess.run( ["tmux", *args],
["tmux", "has-session", "-t", self.session_name],
check=False, check=False,
capture_output=True, capture_output=True,
text=True, text=True,
) )
def _session_exists(self) -> bool:
return self._run_tmux(["has-session", "-t", self.session_name]).returncode == 0
def _ensure_session_exists(self) -> None:
"""대상 tmux 세션 존재 여부를 검증한다."""
result = self._run_tmux(["has-session", "-t", self.session_name])
if result.returncode != 0: if result.returncode != 0:
raise RuntimeError( raise RuntimeError(
"tmux 세션이 없습니다. 먼저 실행하세요: " "tmux 세션이 없습니다. 먼저 실행하세요: "
f"tmux new -s {self.session_name} {self.cli_name}" f"tmux new -s {self.session_name} claude"
) )
def start(self) -> None: def _capture_lines(self) -> list[str]:
"""기존 tmux 세션에 attach한다.""" """pane의 최근 출력 스냅샷을 줄 단위로 가져온다."""
self._ensure_session_exists() result = self._run_tmux(
logger.info("tmux 세션 attach: %s", self.session_name) ["capture-pane", "-p", "-t", self.session_name, "-S", "-200"]
self._process = pexpect.spawn(
"tmux",
["attach-session", "-t", self.session_name],
encoding="utf-8",
timeout=None,
) )
if result.returncode != 0:
return []
return result.stdout.splitlines()
@staticmethod
def _diff_lines(previous: list[str], current: list[str]) -> list[str]:
"""이전 스냅샷 대비 새로 추가된 줄만 계산한다."""
if not previous:
return []
prefix = 0
for prev, curr in zip(previous, current):
if prev != curr:
break
prefix += 1
if prefix > 0:
return current[prefix:]
# 화면 스크롤/리프레시로 앞부분이 달라진 경우 tail overlap으로 보정.
max_overlap = min(len(previous), len(current))
for overlap in range(max_overlap, 0, -1):
if previous[-overlap:] == current[:overlap]:
return current[overlap:]
return current
def start(self) -> None:
"""기존 tmux 세션에 연결한다."""
self._ensure_session_exists()
logger.info("tmux 세션 연결: %s", self.session_name)
self._connected = True
self._last_capture = self._capture_lines()
def send(self, text: str) -> None: def send(self, text: str) -> None:
"""프로세스에 텍스트 입력을 전달한다.""" """세션에 텍스트 입력을 전달한다."""
if not self.is_alive: if not self.is_alive:
raise RuntimeError("프로세스가 실행 중이 아닙니다.") raise RuntimeError("프로세스가 실행 중이 아닙니다.")
assert self._process is not None
logger.debug("입력 전송: %s", text) logger.debug("입력 전송: %s", text)
self._process.sendline(text) result = self._run_tmux(["send-keys", "-t", self.session_name, text, "C-m"])
if result.returncode != 0:
raise RuntimeError("tmux 입력 전송에 실패했습니다.")
def read_output(self, timeout: int = 5) -> str: def read_output(self, timeout: int = 5) -> str:
"""프로세스의 출력을 읽는다.""" """세션 pane 출력을 읽는다."""
_ = timeout
if not self.is_alive: if not self.is_alive:
raise RuntimeError("프로세스가 실행 중이 아닙니다.") raise RuntimeError("프로세스가 실행 중이 아닙니다.")
assert self._process is not None current = self._capture_lines()
try: delta_lines = self._diff_lines(self._last_capture, current)
self._process.expect(pexpect.TIMEOUT, timeout=timeout) self._last_capture = current
except pexpect.TIMEOUT: return "\n".join(delta_lines).strip()
pass
return self._process.before or ""
def stop(self) -> None: def stop(self) -> None:
"""attach 연결만 종료한다(원격 세션은 유지).""" """브릿지 연결만 종료한다(원격 세션은 유지)."""
if self._process is not None: if self._connected:
logger.info("tmux attach 종료") logger.info("tmux 세션 연결 해제")
self._process.close(force=True) self._connected = False
self._process = None self._last_capture = []

43
src/lazy_enter/slack_handler.py
View File

@@ -4,7 +4,6 @@ from __future__ import annotations
import logging import logging
import time import time
from collections.abc import Callable
from slack_bolt import App from slack_bolt import App
from slack_bolt.adapter.socket_mode import SocketModeHandler from slack_bolt.adapter.socket_mode import SocketModeHandler
@@ -23,8 +22,8 @@ class SlackHandler:
self.app = App(token=config.slack_bot_token) self.app = App(token=config.slack_bot_token)
self._handler: SocketModeHandler | None = None self._handler: SocketModeHandler | None = None
self._stop_requested = False self._stop_requested = False
self._on_message_callback: Callable[[str, str], None] | None = None self._on_message_callback: callable | None = None
self._on_command_callback: Callable[[str, str, str], None] | None = None self._on_command_callback: callable | None = None
self._register_listeners() self._register_listeners()
@@ -47,7 +46,7 @@ class SlackHandler:
"""Slack 이벤트 리스너를 등록한다.""" """Slack 이벤트 리스너를 등록한다."""
@self.app.event("message") @self.app.event("message")
def handle_message(event: dict, say: Callable[..., None]) -> None: def handle_message(event: dict, say: callable) -> None:
user_id = event.get("user", "") user_id = event.get("user", "")
channel_id = event.get("channel", "") channel_id = event.get("channel", "")
text = event.get("text", "") text = event.get("text", "")
@@ -59,7 +58,7 @@ class SlackHandler:
self._on_message_callback(text, channel_id) self._on_message_callback(text, channel_id)
@self.app.command("/start-claude") @self.app.command("/start-claude")
def handle_start_claude(ack: Callable[..., None], body: dict) -> None: def handle_start(ack: callable, body: dict) -> None:
ack() ack()
user_id = body.get("user_id", "") user_id = body.get("user_id", "")
channel_id = body.get("channel_id", "") channel_id = body.get("channel_id", "")
@@ -68,10 +67,10 @@ class SlackHandler:
return return
if self._on_command_callback: if self._on_command_callback:
self._on_command_callback("start", "claude", channel_id) self._on_command_callback("start", channel_id)
@self.app.command("/stop-claude") @self.app.command("/stop-claude")
def handle_stop_claude(ack: Callable[..., None], body: dict) -> None: def handle_stop(ack: callable, body: dict) -> None:
ack() ack()
user_id = body.get("user_id", "") user_id = body.get("user_id", "")
channel_id = body.get("channel_id", "") channel_id = body.get("channel_id", "")
@@ -80,37 +79,13 @@ class SlackHandler:
return return
if self._on_command_callback: if self._on_command_callback:
self._on_command_callback("stop", "claude", channel_id) self._on_command_callback("stop", channel_id)
@self.app.command("/start-codex") def on_message(self, callback: callable) -> None:
def handle_start_codex(ack: Callable[..., None], body: dict) -> None:
ack()
user_id = body.get("user_id", "")
channel_id = body.get("channel_id", "")
if not self._is_authorized(user_id, channel_id):
return
if self._on_command_callback:
self._on_command_callback("start", "codex", channel_id)
@self.app.command("/stop-codex")
def handle_stop_codex(ack: Callable[..., None], body: dict) -> None:
ack()
user_id = body.get("user_id", "")
channel_id = body.get("channel_id", "")
if not self._is_authorized(user_id, channel_id):
return
if self._on_command_callback:
self._on_command_callback("stop", "codex", channel_id)
def on_message(self, callback: Callable[[str, str], None]) -> None:
"""메시지 수신 콜백을 등록한다.""" """메시지 수신 콜백을 등록한다."""
self._on_message_callback = callback self._on_message_callback = callback
def on_command(self, callback: Callable[[str, str, str], None]) -> None: def on_command(self, callback: callable) -> None:
"""슬래시 커맨드 콜백을 등록한다.""" """슬래시 커맨드 콜백을 등록한다."""
self._on_command_callback = callback self._on_command_callback = callback

115
tests/test_bridge.py
View File

@@ -1,115 +0,0 @@
"""Bridge 명령 라우팅 테스트."""
from __future__ import annotations
import sys
import types
slack_bolt = types.ModuleType("slack_bolt")
slack_bolt.App = object
socket_mode = types.ModuleType("slack_bolt.adapter.socket_mode")
socket_mode.SocketModeHandler = object
slack_sdk = types.ModuleType("slack_sdk")
slack_sdk.WebClient = object
sys.modules.setdefault("slack_bolt", slack_bolt)
sys.modules.setdefault("slack_bolt.adapter.socket_mode", socket_mode)
sys.modules.setdefault("slack_sdk", slack_sdk)
class FakeSlackHandler:
def __init__(self, _config) -> None:
self.message_callback = None
self.command_callback = None
self.sent_messages: list[tuple[str, str]] = []
def on_message(self, callback) -> None:
self.message_callback = callback
def on_command(self, callback) -> None:
self.command_callback = callback
def send_message(self, channel: str, text: str) -> None:
self.sent_messages.append((channel, text))
def start(self) -> None:
return None
def stop(self) -> None:
return None
class FakePtyManager:
instances: list[FakePtyManager] = []
def __init__(self, session_name: str, cli_name: str = "claude") -> None:
self.session_name = session_name
self.cli_name = cli_name
self._alive = False
FakePtyManager.instances.append(self)
@property
def is_alive(self) -> bool:
return self._alive
def start(self) -> None:
self._alive = True
def stop(self) -> None:
self._alive = False
def send(self, _text: str) -> None:
return None
def read_output(self, timeout: int = 5) -> str:
return ""
def _make_bridge(monkeypatch):
from lazy_enter.bridge import Bridge
from lazy_enter.config import Config
monkeypatch.setattr("lazy_enter.bridge.SlackHandler", FakeSlackHandler)
monkeypatch.setattr("lazy_enter.bridge.PtyManager", FakePtyManager)
config = Config()
config.tmux_session_name = "claude-room"
config.codex_tmux_session_name = "codex-room"
return Bridge(config)
def test_start_claude_routes_to_claude_session(monkeypatch) -> None:
FakePtyManager.instances.clear()
bridge = _make_bridge(monkeypatch)
bridge._handle_command("start", "claude", "C1")
assert FakePtyManager.instances[-1].session_name == "claude-room"
assert FakePtyManager.instances[-1].cli_name == "claude"
assert bridge.slack.sent_messages[-1] == (
"C1",
":link: Claude 세션에 연결되었습니다.",
)
def test_start_codex_routes_to_codex_session(monkeypatch) -> None:
FakePtyManager.instances.clear()
bridge = _make_bridge(monkeypatch)
bridge._handle_command("start", "codex", "C1")
assert FakePtyManager.instances[-1].session_name == "codex-room"
assert FakePtyManager.instances[-1].cli_name == "codex"
assert bridge.slack.sent_messages[-1] == (
"C1",
":link: Codex 세션에 연결되었습니다.",
)
def test_unknown_target_is_rejected(monkeypatch) -> None:
bridge = _make_bridge(monkeypatch)
bridge._handle_command("start", "unknown", "C1")
assert bridge.slack.sent_messages[-1] == (
"C1",
":warning: 지원하지 않는 대상입니다.",
)

1
tests/test_config.py
View File

@@ -8,7 +8,6 @@ from lazy_enter.config import Config
def test_config_defaults(): def test_config_defaults():
config = Config() config = Config()
assert config.tmux_session_name == "claude" assert config.tmux_session_name == "claude"
assert config.codex_tmux_session_name == "codex"
assert config.pty_read_timeout == 5 assert config.pty_read_timeout == 5
assert config.output_buffer_interval == 2.0 assert config.output_buffer_interval == 2.0
assert config.max_message_length == 3000 assert config.max_message_length == 3000

85
tests/test_pty_manager.py
View File

@@ -3,29 +3,17 @@
from __future__ import annotations from __future__ import annotations
from collections.abc import Sequence from collections.abc import Sequence
from dataclasses import dataclass
import pytest import pytest
from lazy_enter.pty_manager import PtyManager from lazy_enter.pty_manager import PtyManager
class FakeSpawn: @dataclass
def __init__(self, *_args: object, **_kwargs: object) -> None: class Result:
self.before = "" returncode: int
self._alive = True stdout: str = ""
def isalive(self) -> bool:
return self._alive
def sendline(self, _text: str) -> None:
return None
def expect(self, *_args: object, **_kwargs: object) -> None:
return None
def close(self, force: bool = False) -> None:
assert force is True
self._alive = False
def test_initial_state(): def test_initial_state():
@@ -45,10 +33,7 @@ def test_start_raises_when_tmux_session_missing(monkeypatch: pytest.MonkeyPatch)
assert capture_output is True assert capture_output is True
assert text is True assert text is True
class Result: return Result(returncode=1)
returncode = 1
return Result()
monkeypatch.setattr("lazy_enter.pty_manager.subprocess.run", fake_run) monkeypatch.setattr("lazy_enter.pty_manager.subprocess.run", fake_run)
@@ -57,9 +42,47 @@ def test_start_raises_when_tmux_session_missing(monkeypatch: pytest.MonkeyPatch)
pty.start() pty.start()
def test_start_and_stop_attach(monkeypatch: pytest.MonkeyPatch): def test_start_send_read_and_stop(monkeypatch: pytest.MonkeyPatch):
calls: list[list[str]] = []
def fake_run( def fake_run(
_cmd: Sequence[str], cmd: Sequence[str],
check: bool,
capture_output: bool,
text: bool,
):
calls.append(list(cmd))
assert check is False
assert capture_output is True
assert text is True
if cmd[:4] == ["tmux", "has-session", "-t", "claude"]:
return Result(returncode=0)
if cmd[:7] == ["tmux", "capture-pane", "-p", "-t", "claude", "-S", "-200"]:
# 첫 캡처는 baseline, 두 번째 캡처에서 한 줄 증가.
if calls.count(list(cmd)) == 1:
return Result(returncode=0, stdout="line1\nline2\n")
return Result(returncode=0, stdout="line1\nline2\nline3\n")
if cmd[:6] == ["tmux", "send-keys", "-t", "claude", "hello", "C-m"]:
return Result(returncode=0)
return Result(returncode=1)
monkeypatch.setattr("lazy_enter.pty_manager.subprocess.run", fake_run)
pty = PtyManager("claude")
pty.start()
assert pty.is_alive
pty.send("hello")
assert pty.read_output() == "line3"
pty.stop()
assert not pty.is_alive
def test_read_output_returns_empty_when_unchanged(monkeypatch: pytest.MonkeyPatch):
def fake_run(
cmd: Sequence[str],
check: bool, check: bool,
capture_output: bool, capture_output: bool,
text: bool, text: bool,
@@ -67,18 +90,14 @@ def test_start_and_stop_attach(monkeypatch: pytest.MonkeyPatch):
assert check is False assert check is False
assert capture_output is True assert capture_output is True
assert text is True assert text is True
if cmd[:4] == ["tmux", "has-session", "-t", "claude"]:
class Result: return Result(returncode=0)
returncode = 0 if cmd[:7] == ["tmux", "capture-pane", "-p", "-t", "claude", "-S", "-200"]:
return Result(returncode=0, stdout="same\n")
return Result() return Result(returncode=1)
monkeypatch.setattr("lazy_enter.pty_manager.subprocess.run", fake_run) monkeypatch.setattr("lazy_enter.pty_manager.subprocess.run", fake_run)
monkeypatch.setattr("lazy_enter.pty_manager.pexpect.spawn", FakeSpawn)
pty = PtyManager("claude") pty = PtyManager("claude")
pty.start() pty.start()
assert pty.is_alive assert pty.read_output() == ""
pty.stop()
assert not pty.is_alive