343 lines
12 KiB
Python
343 lines
12 KiB
Python
"""Unit tests for shellbound (no network required)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
|
|
from shellbound import attach
|
|
from shellbound import modes
|
|
from shellbound.personas import (
|
|
AmbiguousPersona,
|
|
can_expected_mode,
|
|
default_persona_name,
|
|
resolve_mode,
|
|
resolve_name,
|
|
shellbound_personas,
|
|
)
|
|
|
|
RAW = [
|
|
{"name": "shellbound_shell", "applies_to_kinds": ["interactive"]},
|
|
{"name": "shellbound_answer", "applies_to_kinds": ["interactive"]},
|
|
{"name": "shellbound_creative", "applies_to_kinds": ["interactive"]},
|
|
{"name": "shellbound_expert", "applies_to_kinds": ["interactive"]},
|
|
{"name": "shellbound_explain", "applies_to_kinds": ["interactive"]},
|
|
{"name": "shellbound_poetic", "applies_to_kinds": ["interactive"]},
|
|
{"name": "engineer", "applies_to_kinds": ["interactive"], "is_default": True},
|
|
{"name": "orchestrator", "applies_to_kinds": ["coordinator"]},
|
|
]
|
|
|
|
|
|
def test_shellbound_filter():
|
|
slugs = {p["name"] for p in shellbound_personas(RAW)}
|
|
assert slugs == {
|
|
"shellbound_shell",
|
|
"shellbound_answer",
|
|
"shellbound_creative",
|
|
"shellbound_expert",
|
|
"shellbound_explain",
|
|
"shellbound_poetic",
|
|
}
|
|
|
|
|
|
def test_unique_abbreviations():
|
|
assert resolve_mode(RAW, "s")["name"] == "shellbound_shell"
|
|
assert resolve_mode(RAW, "a")["name"] == "shellbound_answer"
|
|
assert resolve_mode(RAW, "c")["name"] == "shellbound_creative"
|
|
assert resolve_mode(RAW, "p")["name"] == "shellbound_poetic"
|
|
assert resolve_mode(RAW, "sh")["name"] == "shellbound_shell"
|
|
assert resolve_mode(RAW, "expe")["name"] == "shellbound_expert"
|
|
assert resolve_mode(RAW, "expl")["name"] == "shellbound_explain"
|
|
assert resolve_mode(RAW, "shellbound_shell")["name"] == "shellbound_shell"
|
|
assert resolve_mode(RAW, "SHELLBOUND_ANSWER")["name"] == "shellbound_answer"
|
|
|
|
|
|
def test_ambiguous_abbreviation():
|
|
try:
|
|
resolve_mode(RAW, "e")
|
|
except AmbiguousPersona as exc:
|
|
assert set(exc.candidates) == {"shellbound_expert", "shellbound_explain"}
|
|
else:
|
|
raise AssertionError("expected AmbiguousPersona")
|
|
|
|
|
|
def test_no_match_is_query():
|
|
assert resolve_mode(RAW, "zzz") is None
|
|
assert can_expected_mode("zzz") is False
|
|
assert can_expected_mode("s") is True
|
|
|
|
|
|
def test_resolve_name():
|
|
assert resolve_name(RAW, "engineer")["name"] == "engineer"
|
|
assert resolve_name(RAW, "shellbound_poetic")["name"] == "shellbound_poetic"
|
|
try:
|
|
resolve_name(RAW, "orchestrator") # coordinator kind, not interactive
|
|
except ValueError:
|
|
pass
|
|
else:
|
|
raise AssertionError("orchestrator should be rejected (non-interactive)")
|
|
|
|
|
|
def test_default_persona():
|
|
assert default_persona_name(RAW) == "shellbound_answer"
|
|
|
|
|
|
def test_extract_commands_fenced():
|
|
raw = '```\nls -la\n# hide me\necho hi\n```\n'
|
|
assert modes.extract_commands(raw) == ["ls -la", "echo hi"]
|
|
|
|
|
|
def test_extract_commands_plain():
|
|
assert modes.extract_commands("whoami\necho ok") == ["whoami", "echo ok"]
|
|
assert modes.extract_commands("# comment only") == []
|
|
|
|
|
|
def test_extract_commands_dollar_prompt():
|
|
assert modes.extract_commands("$ git status") == ["git status"]
|
|
|
|
|
|
def test_minimal_uid():
|
|
assert modes.minimal_uid("abc12345", ["abc99999", "abc12346"]) == "abc12345"
|
|
assert modes.minimal_uid("aaa11111", []) == "aaa1"
|
|
|
|
|
|
def test_session_name():
|
|
assert modes.session_name(" why is the sky blue? ") == "sb: why is the sky blue?"
|
|
|
|
|
|
def test_expert_prompts():
|
|
task = modes.expert_brief_task("hello")
|
|
assert task.startswith("Stage 1 - Produce an Expert Brief")
|
|
assert "USER PROMPT:" in task and "hello" in task
|
|
assert "prompt writing agent" not in task
|
|
retry = modes.expert_retry_task("hello", "META")
|
|
assert retry.startswith("Stage 2 - Adopt the following Expert Persona")
|
|
assert "EXPERT PERSONA" in retry and "META" in retry
|
|
assert "USER PROMPT:" in retry and "hello" in retry
|
|
|
|
|
|
def _ws(ws_id, persona, state, updated):
|
|
ns = type("W", (), {})()
|
|
ns.ws_id = ws_id
|
|
ns.persona = persona
|
|
ns.state = state
|
|
ns.updated = updated
|
|
return ns
|
|
|
|
|
|
def test_select_stale_keeps_newest_n():
|
|
from shellbound.client import select_stale_workspaces
|
|
|
|
workstreams = [
|
|
_ws("aaaa", "shellbound_answer", "idle", "2026-09-05T10:00:00"),
|
|
_ws("bbbb", "shellbound_answer", "idle", "2026-09-05T11:00:00"),
|
|
_ws("cccc", "engineer", "idle", "2026-09-05T12:00:00"), # not shellbound_
|
|
_ws("dddd", "shellbound_shell", "idle", "2026-09-05T09:00:00"),
|
|
]
|
|
stale = select_stale_workspaces(workstreams, keep=2)
|
|
# newest two shellbound_* kept (bbbb, aaaa); only the oldest is stale
|
|
assert [w.ws_id for w in stale] == ["dddd"]
|
|
|
|
|
|
def test_select_stale_excludes_closed_and_exclude_id():
|
|
from shellbound.client import select_stale_workspaces
|
|
|
|
workstreams = [
|
|
_ws("aaaa", "shellbound_answer", "closed", "2026-09-05T10:00:00"),
|
|
_ws("bbbb", "shellbound_answer", "idle", "2026-09-05T11:00:00"),
|
|
_ws("cccc", "shellbound_shell", "idle", "2026-09-05T09:00:00"),
|
|
_ws("dddd", "shellbound_shell", "idle", "2026-09-05T08:00:00"),
|
|
]
|
|
# bbbb is the newest but excluded (it's the session just used); only
|
|
# cccc remains inside keep=1, so the stale set is just dddd.
|
|
stale = select_stale_workspaces(workstreams, keep=1, exclude_id="bbbb")
|
|
assert [w.ws_id for w in stale] == ["dddd"]
|
|
|
|
|
|
def test_select_stale_keep_zero():
|
|
from shellbound.client import select_stale_workspaces
|
|
|
|
workstreams = [_ws("aaaa", "shellbound_answer", "idle", "2026-09-05T10:00:00")]
|
|
assert [w.ws_id for w in select_stale_workspaces(workstreams, keep=0)] == ["aaaa"]
|
|
|
|
|
|
def test_config_prunes_node_base_template(tmp_path, monkeypatch):
|
|
from shellbound import config as c
|
|
|
|
cfg_file = tmp_path / "config.json"
|
|
cfg_file.write_text(
|
|
json.dumps({"gateway": "https://x:1", "token": "t", "node_base_template": "{gateway}/bad"})
|
|
)
|
|
monkeypatch.setattr(c, "CONFIG_FILE", cfg_file)
|
|
cfg = c.Config.load()
|
|
assert "node_base_template" not in cfg
|
|
assert cfg["gateway"] == "https://x:1"
|
|
assert cfg.keep_workspaces == 5
|
|
|
|
|
|
def test_config_keep_workspaces_environ(tmp_path, monkeypatch):
|
|
from shellbound import config as c
|
|
|
|
monkeypatch.setattr(c, "CONFIG_FILE", tmp_path / "missing.json")
|
|
monkeypatch.setenv("SHELLBOUND_KEEP_WORKSPACES", "3")
|
|
assert c.Config.load().keep_workspaces == 3
|
|
|
|
|
|
def test_config_keep_workspaces_edge_values():
|
|
from shellbound.config import Config
|
|
|
|
assert Config({"keep_workspaces": "oops"}).keep_workspaces == 5
|
|
assert Config({}).keep_workspaces == 5
|
|
assert Config({"keep_workspaces": 0}).keep_workspaces == 0
|
|
|
|
|
|
def test_normalize_session_and_close():
|
|
from shellbound.cli import _normalize_session
|
|
|
|
assert _normalize_session(["--session", "abc", "hi"]) == ["--session=abc", "hi"]
|
|
assert _normalize_session(["--session"]) == ["--session=__list__"]
|
|
assert _normalize_session(["--close", "all"]) == ["--close=all"]
|
|
assert _normalize_session(["--close", "--debug"]) == ["--close=__all__", "--debug"]
|
|
|
|
|
|
def test_renderer_flush_paragraphs():
|
|
from shellbound.render import StreamRenderer
|
|
|
|
renderer = StreamRenderer(plain=False)
|
|
renderer.add_content("First paragraph.\n")
|
|
assert not renderer.flushed # in-flight tail, not yet a block
|
|
assert renderer.tail.endswith("\n")
|
|
renderer.add_content("Second.\n\n")
|
|
assert len(renderer.flushed) == 1 and not renderer.tail
|
|
|
|
|
|
def test_renderer_flush_code_fence():
|
|
from shellbound.render import StreamRenderer
|
|
|
|
renderer = StreamRenderer(plain=False)
|
|
# blank lines inside an open fence must not flush; the closed fence
|
|
# becomes a single Markdown block on the trailing blank line.
|
|
renderer.add_content("```bash\necho a\n\n echo b \n```\n\n")
|
|
assert len(renderer.flushed) == 1 and not renderer.tail
|
|
# an unclosed fence cap is force-flushed as raw text
|
|
renderer.add_content("```python\ndef f():\n")
|
|
assert len(renderer.flushed) == 1 # still inside the fence, nothing flushed
|
|
long = "x" * 5000 + "\n"
|
|
renderer.add_content(long)
|
|
assert len(renderer.flushed) == 2 and not renderer.tail
|
|
|
|
|
|
def test_renderer_plain_content_streams():
|
|
from shellbound.render import StreamRenderer
|
|
|
|
renderer = StreamRenderer(plain=True)
|
|
renderer.add_content("just text")
|
|
assert renderer.content_parts == ["just text"]
|
|
assert renderer.finish() == "just text"
|
|
|
|
|
|
# ---- attachments / piped stdin --------------------------------------------
|
|
|
|
|
|
def _warnings():
|
|
seen: list[str] = []
|
|
|
|
def warn(message: str) -> None:
|
|
seen.append(message)
|
|
|
|
return seen, warn
|
|
|
|
|
|
def test_fence_structure():
|
|
assert attach.fence("file: a.txt", "one\ntwo") == (
|
|
"[file: a.txt]\n```text\none\ntwo\n```"
|
|
)
|
|
assert attach.fence("stdin", "x", note="[truncated]") == (
|
|
"[stdin]\n```text\nx\n[truncated]\n```"
|
|
)
|
|
assert attach.fence("clipboard", "\n") == "[clipboard]\n```text\n\n```"
|
|
|
|
|
|
def test_assemble_order_and_stripping():
|
|
blocks = [
|
|
attach.fence("stdin", "pipe"),
|
|
attach.fence("file: a.txt", "file body"),
|
|
attach.fence("clipboard", "clip"),
|
|
]
|
|
assert attach.assemble("hello", blocks) == (
|
|
"hello\n\n"
|
|
"[stdin]\n```text\npipe\n```\n\n"
|
|
"[file: a.txt]\n```text\nfile body\n```\n\n"
|
|
"[clipboard]\n```text\nclip\n```"
|
|
)
|
|
|
|
|
|
def test_assemble_no_prompt_or_blocks():
|
|
assert attach.assemble("", []) == ""
|
|
assert attach.assemble("just prompt", []) == "just prompt"
|
|
assert attach.assemble("", [attach.fence("stdin", "s")]) == "[stdin]\n```text\ns\n```"
|
|
|
|
|
|
def test_read_file_binary_skipped(tmp_path):
|
|
p = tmp_path / "bin.dat"
|
|
p.write_bytes(b"\x00\x01\x02")
|
|
seen, warn = _warnings()
|
|
assert attach.read_file(str(p), 1024, warn=warn) is None
|
|
assert any("binary" in m for m in seen)
|
|
|
|
|
|
def test_read_file_truncates(tmp_path):
|
|
p = tmp_path / "big.txt"
|
|
p.write_text("a" * 5000)
|
|
seen, warn = _warnings()
|
|
result = attach.read_file(str(p), 100, warn=warn)
|
|
assert result.truncated
|
|
assert result.content == "a" * 100
|
|
assert not result.binary
|
|
assert any("truncating" in m for m in seen)
|
|
|
|
|
|
def test_read_file_missing_raises(tmp_path):
|
|
try:
|
|
attach.read_file(str(tmp_path / "nope.txt"), 1024)
|
|
except FileNotFoundError:
|
|
pass
|
|
else:
|
|
raise AssertionError("expected FileNotFoundError")
|
|
|
|
|
|
def test_read_stdin_truncates_and_warns():
|
|
seen, warn = _warnings()
|
|
result = attach.read_stdin(io.BytesIO(b"b" * 500), 10, warn=warn)
|
|
assert result.truncated and result.content == "b" * 10
|
|
assert any("truncating" in m for m in seen)
|
|
|
|
|
|
def test_read_stdin_empty_and_binary():
|
|
result = attach.read_stdin(io.BytesIO(b""), 100)
|
|
assert not result.content and not result.truncated
|
|
seen, warn = _warnings()
|
|
result = attach.read_stdin(io.BytesIO(b"\x00data"), 100, warn=warn)
|
|
assert result.binary and result.content == "\x00data"
|
|
assert any("binary" in m for m in seen)
|
|
|
|
|
|
def test_truncation_note():
|
|
assert attach.truncation_note("stdin", 42) == "[truncated: kept the first 42 bytes of stdin]"
|
|
|
|
|
|
def test_config_attach_max_bytes():
|
|
from shellbound.config import Config
|
|
|
|
assert Config({}).attach_max_bytes == 102_400
|
|
assert Config({"attach_max_bytes": "oops"}).attach_max_bytes == 102_400
|
|
assert Config({"attach_max_bytes": 0}).attach_max_bytes == 1
|
|
assert Config({"attach_max_bytes": 5_000}).attach_max_bytes == 5_000
|
|
|
|
|
|
def test_config_attach_max_bytes_environ(tmp_path, monkeypatch):
|
|
from shellbound import config as c
|
|
|
|
monkeypatch.setattr(c, "CONFIG_FILE", tmp_path / "missing.json")
|
|
monkeypatch.setenv("SHELLBOUND_ATTACH_MAX_BYTES", "4048")
|
|
assert c.Config.load().attach_max_bytes == 4048 |