384 lines
17 KiB
Python
384 lines
17 KiB
Python
# test_a4_adapter_contract.py — Contract tests for BaseAgentAdapter and resolve_home
|
||
|
||
import os, pytest
|
||
from lib_py.paths import resolve_home
|
||
from lib_py.agents.registry import get_adapter, own_key, agent_of_row
|
||
|
||
def test_resolve_home_contract():
|
||
# 1. When explicit home_dir is passed
|
||
assert resolve_home('/custom/home') == '/custom/home'
|
||
|
||
# 2. When HOME_DIR env var is set
|
||
orig = os.environ.get('HOME_DIR')
|
||
try:
|
||
os.environ['HOME_DIR'] = '/env/home'
|
||
assert resolve_home() == '/env/home'
|
||
finally:
|
||
if orig is None:
|
||
os.environ.pop('HOME_DIR', None)
|
||
else:
|
||
os.environ['HOME_DIR'] = orig
|
||
|
||
# 3. When HOME_DIR env var is NOT set, falls back to HOME / expanduser (~), NOT '' or '/'
|
||
env_copy = os.environ.copy()
|
||
env_copy.pop('HOME_DIR', None)
|
||
home_val = env_copy.get('HOME') or os.path.expanduser('~')
|
||
assert resolve_home() == home_val
|
||
|
||
def test_agent_adapter_registry():
|
||
EXPECTED_OWN_KEYS = {
|
||
'claude': 'claude_session_id_own',
|
||
'agy': 'agy_conversation_id_own',
|
||
'hermes': 'hermes_conversation_id_own',
|
||
'cline': 'cline_conversation_id_own',
|
||
'grok': 'grok_session_id_own',
|
||
}
|
||
for agent, expected in EXPECTED_OWN_KEYS.items():
|
||
adapter = get_adapter(agent)
|
||
assert adapter is not None
|
||
assert adapter.name == agent
|
||
assert adapter.own_key == expected
|
||
assert own_key(agent) == adapter.own_key
|
||
|
||
def test_agent_of_row_priority():
|
||
# Priority 1: Explicit 'agent' field
|
||
row1 = {'agent': 'agy', 'name': 'some-name-creator-claude'}
|
||
assert agent_of_row(row1) == 'agy'
|
||
|
||
# Priority 2: Session name suffix
|
||
row2 = {'name': 'my-workspace-planner-hermes'}
|
||
assert agent_of_row(row2) == 'hermes'
|
||
|
||
# Priority 3: pane.cmd exact match
|
||
row3 = {'pane': {'cmd': 'cline'}}
|
||
assert agent_of_row(row3) == 'cline'
|
||
|
||
def test_agent_of_row_pane_cmd_binary_path_and_failure():
|
||
# pane.cmd 가 절대 경로 형태여도 해석된다
|
||
assert agent_of_row({'pane': {'cmd': '/usr/local/bin/agy'}}) == 'agy'
|
||
# 세 경로 모두 실패하면 None — 호출자가 오류를 소유한다
|
||
assert agent_of_row({}, session_name='bad-session-name') is None
|
||
# 입양 조회용 match_cmd=False 에서는 pane.cmd 를 보지 않는다
|
||
assert agent_of_row({'name': 'agy-creator-01', 'pane': {'cmd': 'agy'}},
|
||
match_cmd=False) is None
|
||
|
||
def test_adapter_required_properties():
|
||
from lib_py.agents.base import BaseAgentAdapter
|
||
base = BaseAgentAdapter()
|
||
for prop in ('name', 'own_key', 'ready_tokens', 'exit_key', 'delegate_agent_key', 'identity_cache_fields'):
|
||
with pytest.raises(NotImplementedError):
|
||
getattr(base, prop)
|
||
|
||
expected = {
|
||
'claude': ('Anthropic|Assistant|Chat|Welcome|Claude Code|Opus|Sonnet|Haiku', '/exit', 'claude-code', ('session_id', 'session_jsonl', 'session_size_bytes', 'session_lines')),
|
||
'agy': ('Antigravity', 'Exit', 'antigravity-cli', ('conversation_id', 'conversation_db', 'conversation_brain_dir')),
|
||
'hermes': ('Hermes', '/exit', 'hermes-agent', ('session_id',)),
|
||
'cline': ('Cline|history|Chat|What can I do|slash commands', '/exit', 'cline-agent', ('session_id',)),
|
||
'grok': ('Grok|xAI|Assistant|❯|>>>', '/exit', 'grok-build', ('session_id', 'session_jsonl')),
|
||
}
|
||
for agent, (toks, exitk, delk, cache_f) in expected.items():
|
||
adapter = get_adapter(agent)
|
||
assert adapter is not None
|
||
assert adapter.ready_tokens == toks
|
||
assert adapter.exit_key == exitk
|
||
assert adapter.delegate_agent_key == delk
|
||
assert adapter.identity_cache_fields == cache_f
|
||
|
||
def test_facts_bridge_eval_contract():
|
||
import subprocess, sys
|
||
from pathlib import Path
|
||
env = os.environ.copy()
|
||
skills_dir = str(Path(__file__).resolve().parent.parent / ".agents" / "skills")
|
||
env["PYTHONPATH"] = f"{skills_dir}:{env.get('PYTHONPATH', '')}"
|
||
for agent in ('claude', 'agy', 'hermes', 'cline', 'grok'):
|
||
res = subprocess.run([sys.executable, "-m", "lib_py.agents", "facts", agent], capture_output=True, text=True, env=env)
|
||
assert res.returncode == 0
|
||
facts_output = res.stdout
|
||
|
||
# Verify eval in bash with set -euo pipefail
|
||
bash_cmd = f"""
|
||
set -euo pipefail
|
||
eval {shlex_quote(facts_output)}
|
||
echo "AGENT=$MAM_AGENT_NAME|OWN=$MAM_OWN_KEY|TOK=$MAM_READY_TOKENS|EXIT=$MAM_EXIT_KEY|DEL=$MAM_DELEGATE_AGENT_KEY|PH=$MAM_INPUT_PLACEHOLDER"
|
||
"""
|
||
res_bash = subprocess.run(["bash", "-c", bash_cmd], capture_output=True, text=True)
|
||
assert res_bash.returncode == 0, f"Bash eval failed for {agent}:\nStdout: {res_bash.stdout}\nStderr: {res_bash.stderr}"
|
||
if agent == 'cline':
|
||
assert "PH=Ask anything..." in res_bash.stdout
|
||
|
||
def shlex_quote(s):
|
||
import shlex
|
||
return shlex.quote(s)
|
||
|
||
def test_purge_artifacts_composite(tmp_path):
|
||
import sqlite3
|
||
from lib_py.agents.base import DiscoveryContext
|
||
ws = str(tmp_path / "ws")
|
||
home = str(tmp_path / "home")
|
||
os.makedirs(ws, exist_ok=True)
|
||
os.makedirs(home, exist_ok=True)
|
||
|
||
# 1. Claude
|
||
claude_adapter = get_adapter('claude')
|
||
claude_ctx = DiscoveryContext(workspace=ws, agent_name='claude', home_dir=home)
|
||
c_path = claude_adapter.artifact_path('uuid-c', claude_ctx)
|
||
os.makedirs(os.path.dirname(c_path), exist_ok=True)
|
||
with open(c_path, 'w') as f:
|
||
f.write('{"sessionId": "uuid-c"}')
|
||
assert os.path.exists(c_path)
|
||
purged_c = claude_adapter.purge_artifacts('uuid-c', claude_ctx)
|
||
assert len(purged_c) == 1
|
||
assert not os.path.exists(c_path)
|
||
|
||
# 2. Agy (both DB file and brain dir)
|
||
agy_adapter = get_adapter('agy')
|
||
agy_ctx = DiscoveryContext(workspace=ws, agent_name='agy', home_dir=home)
|
||
agy_db = agy_adapter.artifact_path('uuid-a', agy_ctx)
|
||
os.makedirs(os.path.dirname(agy_db), exist_ok=True)
|
||
with open(agy_db, 'w') as f:
|
||
f.write('mock db')
|
||
agy_brain = f"{home}/.gemini/antigravity-cli/brain/uuid-a"
|
||
os.makedirs(agy_brain, exist_ok=True)
|
||
with open(f"{agy_brain}/note.txt", 'w') as f:
|
||
f.write('brain note')
|
||
purged_a = agy_adapter.purge_artifacts('uuid-a', agy_ctx)
|
||
assert len(purged_a) == 2
|
||
assert not os.path.exists(agy_db)
|
||
assert not os.path.exists(agy_brain)
|
||
|
||
# 3. Hermes (JSON file and SQLite rows)
|
||
hermes_adapter = get_adapter('hermes')
|
||
hermes_ctx = DiscoveryContext(workspace=ws, agent_name='hermes', home_dir=home)
|
||
h_json = hermes_adapter.artifact_path('uuid-h', hermes_ctx)
|
||
os.makedirs(os.path.dirname(h_json), exist_ok=True)
|
||
with open(h_json, 'w') as f:
|
||
f.write('{}')
|
||
h_db = f"{home}/.hermes/state.db"
|
||
os.makedirs(os.path.dirname(h_db), exist_ok=True)
|
||
conn = sqlite3.connect(h_db)
|
||
conn.execute("CREATE TABLE IF NOT EXISTS sessions (id TEXT, cwd TEXT)")
|
||
conn.execute("CREATE TABLE IF NOT EXISTS messages (session_id TEXT, msg TEXT)")
|
||
conn.execute("INSERT INTO sessions VALUES (?, ?)", ('uuid-h', ws))
|
||
conn.execute("INSERT INTO messages VALUES (?, ?)", ('uuid-h', 'hello'))
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
purged_h = hermes_adapter.purge_artifacts('uuid-h', hermes_ctx)
|
||
assert len(purged_h) == 2
|
||
assert not os.path.exists(h_json)
|
||
conn = sqlite3.connect(h_db)
|
||
assert conn.execute("SELECT count(*) FROM sessions WHERE id='uuid-h'").fetchone()[0] == 0
|
||
assert conn.execute("SELECT count(*) FROM messages WHERE session_id='uuid-h'").fetchone()[0] == 0
|
||
conn.close()
|
||
|
||
# 4. Cline (sessions dir)
|
||
cline_adapter = get_adapter('cline')
|
||
cline_ctx = DiscoveryContext(workspace=ws, agent_name='cline', home_dir=home)
|
||
cline_dir = f"{home}/.cline/data/sessions/uuid-cl"
|
||
os.makedirs(cline_dir, exist_ok=True)
|
||
with open(f"{cline_dir}/uuid-cl.json", 'w') as f:
|
||
f.write('{"session_id": "uuid-cl"}')
|
||
purged_cl = cline_adapter.purge_artifacts('uuid-cl', cline_ctx)
|
||
assert len(purged_cl) == 1
|
||
assert not os.path.exists(cline_dir)
|
||
|
||
# 5. Grok (session directory containing chat_history.jsonl)
|
||
grok_adapter = get_adapter('grok')
|
||
grok_ctx = DiscoveryContext(workspace=ws, agent_name='grok', home_dir=home)
|
||
g_path = grok_adapter.artifact_path('uuid-g', grok_ctx)
|
||
os.makedirs(os.path.dirname(g_path), exist_ok=True)
|
||
with open(g_path, 'w') as f:
|
||
f.write('{"session_id": "uuid-g"}')
|
||
assert os.path.exists(g_path)
|
||
purged_g = grok_adapter.purge_artifacts('uuid-g', grok_ctx)
|
||
assert len(purged_g) == 1
|
||
assert not os.path.exists(g_path)
|
||
assert not os.path.exists(os.path.dirname(g_path))
|
||
|
||
def test_adapter_spawn_and_resume_specs():
|
||
claude = get_adapter('claude')
|
||
assert claude.spawn_spec('claude', 'u1') == 'claude --dangerously-skip-permissions --session-id u1'
|
||
assert claude.spawn_spec('claude', '', use_wrapper=True) == 'claude --dangerously-skip-permissions'
|
||
assert claude.resume_spec('claude', 'u1', materialized=True) == 'claude --dangerously-skip-permissions -r u1'
|
||
assert claude.resume_spec('claude', 'u1', materialized=False) == 'claude --dangerously-skip-permissions --session-id u1'
|
||
|
||
agy = get_adapter('agy')
|
||
assert agy.spawn_spec('agy', 'u1') == 'agy --dangerously-skip-permissions'
|
||
assert agy.resume_spec('agy', 'u1', materialized=True) == 'agy --dangerously-skip-permissions --conversation u1'
|
||
|
||
hermes = get_adapter('hermes')
|
||
assert hermes.spawn_spec('hermes', 'u1') == 'hermes'
|
||
assert hermes.resume_spec('hermes', 'u1', materialized=True) == 'hermes --resume u1'
|
||
|
||
cline = get_adapter('cline')
|
||
assert cline.spawn_spec('cline', 'u1') == 'cline -i'
|
||
assert cline.resume_spec('cline', 'u1', materialized=True) == 'cline -i --id u1'
|
||
assert cline.resume_spec('cline', 'u1', materialized=False) == 'cline -i'
|
||
|
||
grok = get_adapter('grok')
|
||
assert grok.spawn_spec('grok', 'u1') == 'grok --session-id u1 --permission-mode bypassPermissions'
|
||
assert grok.spawn_spec('grok', '') == 'grok --permission-mode bypassPermissions'
|
||
assert grok.resume_spec('grok', 'u1', materialized=True) == 'grok --resume u1 --permission-mode bypassPermissions'
|
||
assert grok.resume_spec('grok', 'u1', materialized=False) == 'grok --session-id u1 --permission-mode bypassPermissions'
|
||
|
||
def test_adapter_auth_ok(tmp_path, monkeypatch):
|
||
monkeypatch.setenv("HOME_DIR", str(tmp_path))
|
||
# Claude auth runner
|
||
claude = get_adapter('claude')
|
||
assert claude.auth_ok(run_cmd=lambda cmd: (0, '{"loggedIn": true}', '')) is True
|
||
assert claude.auth_ok(run_cmd=lambda cmd: (1, '{"loggedIn": false}', 'error')) is False
|
||
|
||
# Agy auth file check
|
||
agy = get_adapter('agy')
|
||
assert agy.auth_ok() is False
|
||
oauth_file = tmp_path / ".gemini" / "oauth_creds.json"
|
||
oauth_file.parent.mkdir(parents=True, exist_ok=True)
|
||
oauth_file.write_text("{}")
|
||
assert agy.auth_ok() is True
|
||
|
||
# Grok auth check (env var or auth.json file)
|
||
grok = get_adapter('grok')
|
||
assert grok.auth_ok() is False
|
||
monkeypatch.setenv("XAI_API_KEY", "test-key")
|
||
assert grok.auth_ok() is True
|
||
monkeypatch.delenv("XAI_API_KEY", raising=False)
|
||
grok_auth = tmp_path / ".grok" / "auth.json"
|
||
grok_auth.parent.mkdir(parents=True, exist_ok=True)
|
||
grok_auth.write_text("{}")
|
||
assert grok.auth_ok() is True
|
||
|
||
# Hermes & Cline always True
|
||
assert get_adapter('hermes').auth_ok() is True
|
||
assert get_adapter('cline').auth_ok() is True
|
||
|
||
def test_adapter_discover(tmp_path):
|
||
import sqlite3
|
||
from lib_py.agents.base import DiscoveryContext
|
||
ws = str(tmp_path / "ws")
|
||
home = str(tmp_path / "home")
|
||
os.makedirs(ws, exist_ok=True)
|
||
os.makedirs(home, exist_ok=True)
|
||
|
||
# 1. Claude
|
||
claude = get_adapter('claude')
|
||
ctx_c = DiscoveryContext(workspace=ws, agent_name='claude', home_dir=home)
|
||
c_proj = f"{ctx_c.claude_dir}/{ctx_c.ws_key}"
|
||
os.makedirs(c_proj, exist_ok=True)
|
||
with open(f"{c_proj}/u-c1.jsonl", 'w') as f:
|
||
f.write('{"sessionId": "u-c1"}\n')
|
||
assert claude.discover(ctx_c) == ['u-c1']
|
||
|
||
# 2. Agy
|
||
agy = get_adapter('agy')
|
||
ctx_a = DiscoveryContext(workspace=ws, agent_name='agy', home_dir=home)
|
||
a_db = f"{home}/.gemini/antigravity-cli/conversations/u-a1.db"
|
||
os.makedirs(os.path.dirname(a_db), exist_ok=True)
|
||
conn = sqlite3.connect(a_db)
|
||
conn.execute("CREATE TABLE steps (id INT)")
|
||
conn.execute("INSERT INTO steps VALUES (1)")
|
||
conn.commit()
|
||
conn.close()
|
||
lc = f"{home}/.gemini/antigravity-cli/cache/last_conversations.json"
|
||
os.makedirs(os.path.dirname(lc), exist_ok=True)
|
||
with open(lc, 'w') as f:
|
||
import json
|
||
json.dump({ws: 'u-a1'}, f)
|
||
assert agy.discover(ctx_a) == ['u-a1']
|
||
|
||
# 3. Hermes
|
||
hermes = get_adapter('hermes')
|
||
ctx_h = DiscoveryContext(workspace=ws, agent_name='hermes', home_dir=home)
|
||
h_db = f"{home}/.hermes/state.db"
|
||
os.makedirs(os.path.dirname(h_db), exist_ok=True)
|
||
conn = sqlite3.connect(h_db)
|
||
conn.execute("CREATE TABLE sessions (id TEXT, cwd TEXT, started_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)")
|
||
conn.execute("INSERT INTO sessions (id, cwd) VALUES ('u-h1', ?)", (ws,))
|
||
conn.commit()
|
||
conn.close()
|
||
assert hermes.discover(ctx_h) == ['u-h1']
|
||
|
||
# 4. Cline
|
||
cline = get_adapter('cline')
|
||
ctx_cl = DiscoveryContext(workspace=ws, agent_name='cline', home_dir=home)
|
||
cl_sess = f"{home}/.cline/data/sessions/u-cl1"
|
||
os.makedirs(cl_sess, exist_ok=True)
|
||
with open(f"{cl_sess}/u-cl1.json", 'w') as f:
|
||
f.write('{"session_id": "u-cl1", "cwd": "' + ws + '"}')
|
||
assert cline.discover(ctx_cl) == ['u-cl1']
|
||
|
||
# 5. Grok
|
||
grok = get_adapter('grok')
|
||
ctx_g = DiscoveryContext(workspace=ws, agent_name='grok', home_dir=home)
|
||
g_sess = f"{home}/.grok/sessions/{grok._ws_dir(ctx_g)}/u-g1"
|
||
os.makedirs(g_sess, exist_ok=True)
|
||
with open(f"{g_sess}/chat_history.jsonl", 'w') as f:
|
||
f.write('{"session_id": "u-g1", "cwd": "' + ws + '"}\n')
|
||
assert grok.discover(ctx_g) == ['u-g1']
|
||
|
||
def test_cli_bridge_subcommands_and_quote_safety():
|
||
import subprocess, sys
|
||
from pathlib import Path
|
||
env = os.environ.copy()
|
||
skills_dir = str(Path(__file__).resolve().parent.parent / ".agents" / "skills")
|
||
env["PYTHONPATH"] = f"{skills_dir}:{env.get('PYTHONPATH', '')}"
|
||
|
||
# 1. spawn-spec
|
||
res = subprocess.run([sys.executable, "-m", "lib_py.agents", "spawn-spec", "claude", "/path with spaces/claude", "uuid-test", "0"], capture_output=True, text=True, env=env)
|
||
assert res.returncode == 0
|
||
assert res.stdout.strip() == "/path with spaces/claude --dangerously-skip-permissions --session-id uuid-test"
|
||
|
||
# 2. resume-spec with single quotes in workspace path
|
||
res = subprocess.run([sys.executable, "-m", "lib_py.agents", "resume-spec", "claude", "/bin/claude", "uuid-test", "/tmp/bob's ws"], capture_output=True, text=True, env=env)
|
||
assert res.returncode == 0
|
||
assert res.stdout.strip() == "/bin/claude --dangerously-skip-permissions --session-id uuid-test"
|
||
|
||
# 3. exit-key
|
||
for agent, expected_key in [('claude', '/exit'), ('agy', 'Exit'), ('hermes', '/exit'), ('cline', '/exit'), ('grok', '/exit')]:
|
||
res = subprocess.run([sys.executable, "-m", "lib_py.agents", "exit-key", agent], capture_output=True, text=True, env=env)
|
||
assert res.returncode == 0
|
||
assert res.stdout.strip() == expected_key
|
||
|
||
def test_delegate_agent_resolution_and_fallback():
|
||
import subprocess
|
||
expected_map = {
|
||
'claude': 'claude-code',
|
||
'agy': 'antigravity-cli',
|
||
'hermes': 'hermes-agent',
|
||
'cline': 'cline-agent',
|
||
'grok': 'grok-build',
|
||
}
|
||
# 1. Adapter property
|
||
for agent, expected_key in expected_map.items():
|
||
adapter = get_adapter(agent)
|
||
assert adapter.delegate_agent_key == expected_key
|
||
|
||
# 2. Shell fallback resolution when MAM_DELEGATE_AGENT_KEY is unset (R1 fallback)
|
||
for agent, expected_key in expected_map.items():
|
||
sh_snippet = f'''
|
||
AGENT="{agent}"
|
||
MAM_DELEGATE_AGENT_KEY=""
|
||
delegate_agent="${{MAM_DELEGATE_AGENT_KEY:-}}"
|
||
if [ -z "$delegate_agent" ]; then
|
||
case "$AGENT" in
|
||
claude) delegate_agent="claude-code" ;;
|
||
hermes) delegate_agent="hermes-agent" ;;
|
||
cline) delegate_agent="cline-agent" ;;
|
||
agy) delegate_agent="antigravity-cli" ;;
|
||
grok) delegate_agent="grok-build" ;;
|
||
*) echo "ERROR: cannot resolve delegate agent key for '$AGENT'" >&2; exit 2 ;;
|
||
esac
|
||
fi
|
||
echo "$delegate_agent"
|
||
'''
|
||
res = subprocess.run(["bash", "-c", sh_snippet], capture_output=True, text=True)
|
||
assert res.returncode == 0
|
||
assert res.stdout.strip() == expected_key
|
||
|
||
def test_wait_for_tui_ready_missing_tokens_diagnostic(mam_sandbox):
|
||
import subprocess
|
||
lib_sh = mam_sandbox / "skills" / "lib.sh"
|
||
cmd = f'source "{lib_sh}" && wait_for_tui_ready "dummy-sess" "bogus-agent"'
|
||
res = subprocess.run(["bash", "-c", cmd], capture_output=True, text=True)
|
||
assert res.returncode != 0
|
||
assert "no ready tokens for agent 'bogus-agent'" in res.stderr
|