mirror of
https://github.com/NCBM/plyngent.git
synced 2026-07-23 05:55:16 +08:00
core/tools+test: make process and path checks Windows-portable
Use sys.executable for process tests; normalize denylist/glob paths; skip readline when missing.
This commit is contained in:
@@ -1,5 +1,6 @@
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import os
|
||||||
import shlex
|
import shlex
|
||||||
from typing import TYPE_CHECKING, Any, cast, override
|
from typing import TYPE_CHECKING, Any, cast, override
|
||||||
|
|
||||||
@@ -835,7 +836,8 @@ async def handle_slash(state: ReplState, line: str) -> bool:
|
|||||||
if not body:
|
if not body:
|
||||||
return True
|
return True
|
||||||
try:
|
try:
|
||||||
args = shlex.split(body)
|
# Windows paths use backslashes; POSIX shlex would treat them as escapes.
|
||||||
|
args = shlex.split(body, posix=os.name != "nt")
|
||||||
except ValueError as exc:
|
except ValueError as exc:
|
||||||
click.echo(f"error: {exc}")
|
click.echo(f"error: {exc}")
|
||||||
return True
|
return True
|
||||||
|
|||||||
@@ -14,9 +14,9 @@ DEFAULT_MAX_MATCHES = 200
|
|||||||
|
|
||||||
def _rel(path: Path, root: Path) -> str:
|
def _rel(path: Path, root: Path) -> str:
|
||||||
try:
|
try:
|
||||||
return str(path.relative_to(root))
|
return path.relative_to(root).as_posix()
|
||||||
except ValueError:
|
except ValueError:
|
||||||
return str(path)
|
return path.as_posix()
|
||||||
|
|
||||||
|
|
||||||
def _hidden_or_vcs(parts: tuple[str, ...]) -> bool:
|
def _hidden_or_vcs(parts: tuple[str, ...]) -> bool:
|
||||||
|
|||||||
@@ -94,9 +94,10 @@ def resolve_path(path: str | Path) -> Path:
|
|||||||
except ValueError as exc:
|
except ValueError as exc:
|
||||||
msg = f"path escapes workspace root ({root}): {path}"
|
msg = f"path escapes workspace root ({root}): {path}"
|
||||||
raise WorkspaceError(msg) from exc
|
raise WorkspaceError(msg) from exc
|
||||||
resolved_str = str(resolved)
|
# Normalize separators so denylist entries like ``/secrets/`` match on Windows.
|
||||||
|
resolved_str = str(resolved).replace("\\", "/")
|
||||||
for pattern in _state.path_denylist:
|
for pattern in _state.path_denylist:
|
||||||
if pattern and pattern in resolved_str:
|
if pattern and pattern.replace("\\", "/") in resolved_str:
|
||||||
msg = f"path denied by policy (matched {pattern!r}): {path}"
|
msg = f"path denied by policy (matched {pattern!r}): {path}"
|
||||||
raise WorkspaceError(msg)
|
raise WorkspaceError(msg)
|
||||||
return resolved
|
return resolved
|
||||||
|
|||||||
@@ -1,5 +1,8 @@
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import importlib.util
|
||||||
|
|
||||||
|
import pytest
|
||||||
import tomlkit
|
import tomlkit
|
||||||
|
|
||||||
from plyngent.cli.readline_setup import (
|
from plyngent.cli.readline_setup import (
|
||||||
@@ -12,6 +15,11 @@ from plyngent.cli.state import ReplState
|
|||||||
from plyngent.config.models import ModelConfig, OpenAICompatibleProvider, OpenAIProvider
|
from plyngent.config.models import ModelConfig, OpenAICompatibleProvider, OpenAIProvider
|
||||||
from plyngent.config.store import ConfigStore
|
from plyngent.config.store import ConfigStore
|
||||||
|
|
||||||
|
pytestmark = pytest.mark.skipif(
|
||||||
|
importlib.util.find_spec("readline") is None,
|
||||||
|
reason="readline not available (e.g. Windows/Wine)",
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
def _minimal_state(tmp_path: object) -> ReplState:
|
def _minimal_state(tmp_path: object) -> ReplState:
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|||||||
@@ -1,6 +1,7 @@
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
|
import sys
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
from plyngent.tools.process import close_pty, open_pty, read_pty, run_command, write_pty
|
from plyngent.tools.process import close_pty, open_pty, read_pty, run_command, write_pty
|
||||||
@@ -26,9 +27,14 @@ def _field(text: str, name: str) -> str:
|
|||||||
raise AssertionError(msg)
|
raise AssertionError(msg)
|
||||||
|
|
||||||
|
|
||||||
|
def _py(code: str) -> list[str]:
|
||||||
|
"""Cross-platform argv that runs a short Python snippet."""
|
||||||
|
return [sys.executable, "-c", code]
|
||||||
|
|
||||||
|
|
||||||
async def test_run_command_echo(workspace: object) -> None:
|
async def test_run_command_echo(workspace: object) -> None:
|
||||||
del workspace
|
del workspace
|
||||||
out = await call_async(run_command, ["echo", "hi"])
|
out = await call_async(run_command, _py("print('hi')"))
|
||||||
assert "exit_code=0" in out
|
assert "exit_code=0" in out
|
||||||
assert "hi" in out
|
assert "hi" in out
|
||||||
|
|
||||||
@@ -44,13 +50,13 @@ async def test_run_command_cwd(workspace: object) -> None:
|
|||||||
sub = workspace / "sub"
|
sub = workspace / "sub"
|
||||||
sub.mkdir()
|
sub.mkdir()
|
||||||
_ = (sub / "f.txt").write_text("z", encoding="utf-8")
|
_ = (sub / "f.txt").write_text("z", encoding="utf-8")
|
||||||
out = await call_async(run_command, ["ls"], cwd="sub")
|
out = await call_async(run_command, _py("import os; print('\\n'.join(os.listdir()))"), cwd="sub")
|
||||||
assert "f.txt" in out
|
assert "f.txt" in out
|
||||||
|
|
||||||
|
|
||||||
async def test_run_command_timeout(workspace: object) -> None:
|
async def test_run_command_timeout(workspace: object) -> None:
|
||||||
del workspace
|
del workspace
|
||||||
out = await call_async(run_command, ["sleep", "5"], timeout_seconds=0.2)
|
out = await call_async(run_command, _py("import time; time.sleep(5)"), timeout_seconds=0.2)
|
||||||
assert "timed_out=true" in out
|
assert "timed_out=true" in out
|
||||||
|
|
||||||
|
|
||||||
@@ -59,7 +65,7 @@ async def test_run_command_timeout_keeps_partial_output(workspace: object) -> No
|
|||||||
# Print then sleep past the timeout so communicate has partial stdout after kill.
|
# Print then sleep past the timeout so communicate has partial stdout after kill.
|
||||||
out = await call_async(
|
out = await call_async(
|
||||||
run_command,
|
run_command,
|
||||||
["sh", "-c", "printf partial-out; sleep 5"],
|
_py("import sys, time; sys.stdout.write('partial-out'); sys.stdout.flush(); time.sleep(5)"),
|
||||||
timeout_seconds=0.3,
|
timeout_seconds=0.3,
|
||||||
)
|
)
|
||||||
assert "timed_out=true" in out
|
assert "timed_out=true" in out
|
||||||
@@ -68,7 +74,11 @@ async def test_run_command_timeout_keeps_partial_output(workspace: object) -> No
|
|||||||
|
|
||||||
async def test_run_command_stdin(workspace: object) -> None:
|
async def test_run_command_stdin(workspace: object) -> None:
|
||||||
del workspace
|
del workspace
|
||||||
out = await call_async(run_command, ["cat"], stdin="hello-stdin\n")
|
out = await call_async(
|
||||||
|
run_command,
|
||||||
|
_py("import sys; print(sys.stdin.read(), end='')"),
|
||||||
|
stdin="hello-stdin\n",
|
||||||
|
)
|
||||||
assert "exit_code=0" in out
|
assert "exit_code=0" in out
|
||||||
assert "hello-stdin" in out
|
assert "hello-stdin" in out
|
||||||
|
|
||||||
@@ -77,7 +87,7 @@ async def test_run_command_env(workspace: object) -> None:
|
|||||||
del workspace
|
del workspace
|
||||||
out = await call_async(
|
out = await call_async(
|
||||||
run_command,
|
run_command,
|
||||||
["sh", "-c", "printf '%s' \"$PLYNGENT_TEST_VAR\""],
|
_py("import os; print(os.environ.get('PLYNGENT_TEST_VAR', ''), end='')"),
|
||||||
env={"PLYNGENT_TEST_VAR": "from-env"},
|
env={"PLYNGENT_TEST_VAR": "from-env"},
|
||||||
)
|
)
|
||||||
assert "exit_code=0" in out
|
assert "exit_code=0" in out
|
||||||
@@ -87,7 +97,7 @@ async def test_run_command_env(workspace: object) -> None:
|
|||||||
async def test_pty_open_read_close(workspace: object) -> None:
|
async def test_pty_open_read_close(workspace: object) -> None:
|
||||||
del workspace
|
del workspace
|
||||||
try:
|
try:
|
||||||
opened = call_sync(open_pty, ["sleep", "30"])
|
opened = call_sync(open_pty, _py("import time; time.sleep(30)"))
|
||||||
assert "session_id=" in opened
|
assert "session_id=" in opened
|
||||||
session_id = _session_id(opened)
|
session_id = _session_id(opened)
|
||||||
data = await call_async(read_pty, session_id, timeout=0.05)
|
data = await call_async(read_pty, session_id, timeout=0.05)
|
||||||
@@ -109,7 +119,7 @@ def test_pty_denied(workspace: object) -> None:
|
|||||||
async def test_pty_echo_output(workspace: object) -> None:
|
async def test_pty_echo_output(workspace: object) -> None:
|
||||||
del workspace
|
del workspace
|
||||||
try:
|
try:
|
||||||
opened = call_sync(open_pty, ["/bin/echo", "hello-pty"])
|
opened = call_sync(open_pty, _py("print('hello-pty')"))
|
||||||
session_id = _session_id(opened)
|
session_id = _session_id(opened)
|
||||||
text = await call_async(read_pty, session_id, timeout=2.0, until="hello-pty")
|
text = await call_async(read_pty, session_id, timeout=2.0, until="hello-pty")
|
||||||
assert "hello-pty" in text
|
assert "hello-pty" in text
|
||||||
@@ -123,7 +133,19 @@ async def test_pty_echo_output(workspace: object) -> None:
|
|||||||
async def test_write_pty(workspace: object) -> None:
|
async def test_write_pty(workspace: object) -> None:
|
||||||
del workspace
|
del workspace
|
||||||
try:
|
try:
|
||||||
opened = call_sync(open_pty, ["cat"])
|
# Line-oriented echo (portable stand-in for ``cat``).
|
||||||
|
opened = call_sync(
|
||||||
|
open_pty,
|
||||||
|
_py(
|
||||||
|
"import sys\n"
|
||||||
|
"while True:\n"
|
||||||
|
" line = sys.stdin.readline()\n"
|
||||||
|
" if not line:\n"
|
||||||
|
" break\n"
|
||||||
|
" sys.stdout.write(line)\n"
|
||||||
|
" sys.stdout.flush()\n"
|
||||||
|
),
|
||||||
|
)
|
||||||
session_id = _session_id(opened)
|
session_id = _session_id(opened)
|
||||||
written = call_sync(write_pty, session_id, "pty-input\n")
|
written = call_sync(write_pty, session_id, "pty-input\n")
|
||||||
assert "wrote=" in written
|
assert "wrote=" in written
|
||||||
@@ -143,6 +165,10 @@ async def test_pty_exec_failure_surfaces(workspace: object) -> None:
|
|||||||
del workspace
|
del workspace
|
||||||
try:
|
try:
|
||||||
opened = call_sync(open_pty, ["definitely-not-a-real-binary-xyz"])
|
opened = call_sync(open_pty, ["definitely-not-a-real-binary-xyz"])
|
||||||
|
# Windows ConPTY may fail at open; POSIX may open then fail on exec.
|
||||||
|
if "error:" in opened and "session_id=" not in opened:
|
||||||
|
assert "not found" in opened.lower() or "failed" in opened.lower()
|
||||||
|
return
|
||||||
session_id = _session_id(opened)
|
session_id = _session_id(opened)
|
||||||
text = await call_async(read_pty, session_id, timeout=2.0)
|
text = await call_async(read_pty, session_id, timeout=2.0)
|
||||||
# marker and/or dead process with 127
|
# marker and/or dead process with 127
|
||||||
@@ -161,9 +187,9 @@ def test_pty_session_limit(workspace: object) -> None:
|
|||||||
try:
|
try:
|
||||||
PtyManager.set_limit_continue_hook(None)
|
PtyManager.set_limit_continue_hook(None)
|
||||||
PtyManager.configure(max_sessions=1)
|
PtyManager.configure(max_sessions=1)
|
||||||
first = call_sync(open_pty, ["sleep", "30"])
|
first = call_sync(open_pty, _py("import time; time.sleep(30)"))
|
||||||
assert "session_id=" in first
|
assert "session_id=" in first
|
||||||
second = call_sync(open_pty, ["sleep", "30"])
|
second = call_sync(open_pty, _py("import time; time.sleep(30)"))
|
||||||
assert "limit" in second
|
assert "limit" in second
|
||||||
finally:
|
finally:
|
||||||
PtyManager.close_all()
|
PtyManager.close_all()
|
||||||
@@ -177,8 +203,8 @@ def test_pty_session_limit_continue(workspace: object) -> None:
|
|||||||
try:
|
try:
|
||||||
PtyManager.configure(max_sessions=1)
|
PtyManager.configure(max_sessions=1)
|
||||||
PtyManager.set_limit_continue_hook(lambda _reason: True)
|
PtyManager.set_limit_continue_hook(lambda _reason: True)
|
||||||
first = call_sync(open_pty, ["sleep", "30"])
|
first = call_sync(open_pty, _py("import time; time.sleep(30)"))
|
||||||
second = call_sync(open_pty, ["sleep", "30"])
|
second = call_sync(open_pty, _py("import time; time.sleep(30)"))
|
||||||
assert "session_id=" in first
|
assert "session_id=" in first
|
||||||
assert "session_id=" in second
|
assert "session_id=" in second
|
||||||
assert PtyManager.max_sessions >= 2
|
assert PtyManager.max_sessions >= 2
|
||||||
@@ -194,7 +220,7 @@ async def test_pty_output_budget(workspace: object) -> None:
|
|||||||
try:
|
try:
|
||||||
PtyManager.set_limit_continue_hook(None)
|
PtyManager.set_limit_continue_hook(None)
|
||||||
PtyManager.configure(session_output_budget=64)
|
PtyManager.configure(session_output_budget=64)
|
||||||
opened = call_sync(open_pty, ["sh", "-c", "yes x | head -c 1000"])
|
opened = call_sync(open_pty, _py("print('x' * 1000)"))
|
||||||
session_id = _session_id(opened)
|
session_id = _session_id(opened)
|
||||||
# Drain until budget exhausted or process ends.
|
# Drain until budget exhausted or process ends.
|
||||||
budget_hit = False
|
budget_hit = False
|
||||||
@@ -223,7 +249,7 @@ async def test_pty_output_budget_is_per_session(workspace: object) -> None:
|
|||||||
PtyManager.configure(session_output_budget=1024)
|
PtyManager.configure(session_output_budget=1024)
|
||||||
class_budget = PtyManager.session_output_budget
|
class_budget = PtyManager.session_output_budget
|
||||||
PtyManager.set_limit_continue_hook(lambda _reason: True)
|
PtyManager.set_limit_continue_hook(lambda _reason: True)
|
||||||
opened = call_sync(open_pty, ["sh", "-c", "yes x | head -c 200"])
|
opened = call_sync(open_pty, _py("print('x' * 200)"))
|
||||||
session_id = _session_id(opened)
|
session_id = _session_id(opened)
|
||||||
session = PtyManager.get(session_id)
|
session = PtyManager.get(session_id)
|
||||||
assert session is not None
|
assert session is not None
|
||||||
@@ -246,7 +272,6 @@ async def test_pty_output_budget_is_per_session(workspace: object) -> None:
|
|||||||
def test_pty_master_not_inheritable(workspace: object) -> None:
|
def test_pty_master_not_inheritable(workspace: object) -> None:
|
||||||
del workspace
|
del workspace
|
||||||
import os
|
import os
|
||||||
import sys
|
|
||||||
|
|
||||||
if sys.platform == "win32":
|
if sys.platform == "win32":
|
||||||
import pytest
|
import pytest
|
||||||
@@ -254,7 +279,7 @@ def test_pty_master_not_inheritable(workspace: object) -> None:
|
|||||||
pytest.skip("master FD inheritance is POSIX-only")
|
pytest.skip("master FD inheritance is POSIX-only")
|
||||||
|
|
||||||
try:
|
try:
|
||||||
opened = call_sync(open_pty, ["sleep", "5"])
|
opened = call_sync(open_pty, _py("import time; time.sleep(5)"))
|
||||||
session_id = _session_id(opened)
|
session_id = _session_id(opened)
|
||||||
session = PtyManager.get(session_id)
|
session = PtyManager.get(session_id)
|
||||||
assert session is not None
|
assert session is not None
|
||||||
|
|||||||
@@ -1,9 +1,4 @@
|
|||||||
"""Stubs for pywinpty (ConPTY) matching winpty.PtyProcess runtime API."""
|
|
||||||
|
|
||||||
from __future__ import annotations
|
|
||||||
|
|
||||||
from collections.abc import Sequence
|
from collections.abc import Sequence
|
||||||
from typing import Any
|
|
||||||
|
|
||||||
class PtyProcess:
|
class PtyProcess:
|
||||||
exitstatus: int | None
|
exitstatus: int | None
|
||||||
|
|||||||
Reference in New Issue
Block a user