292 lines
10 KiB
Python
292 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
tests/test_b1_tier3_identity.py — B-1 tier-3 identity cache regression suite (V-1..V-8).
|
|
"""
|
|
|
|
import os
|
|
import shutil
|
|
import sqlite3
|
|
import json
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
import pytest
|
|
|
|
REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
|
|
LIB_SH = os.path.join(REPO_ROOT, ".agents", "skills", "lib.sh")
|
|
|
|
@pytest.fixture
|
|
def split_sandbox(tmp_path):
|
|
"""
|
|
Creates a sandbox where workspace directory != state file directory,
|
|
breaking the coincidence in mam_sandbox where <ws>/.mam/ matches AGENT_SESSIONS_YAML.
|
|
"""
|
|
shutil.copytree(os.path.join(REPO_ROOT, ".agents", "skills"), tmp_path / ".agents" / "skills")
|
|
ws = tmp_path / "ws"
|
|
(ws / ".mam").mkdir(parents=True)
|
|
state_dir = tmp_path / "state"
|
|
state_dir.mkdir()
|
|
home = tmp_path / "home"
|
|
(home / ".claude" / "projects").mkdir(parents=True)
|
|
(home / ".gemini" / "antigravity-cli" / "conversations").mkdir(parents=True)
|
|
|
|
yaml_file = state_dir / "agent-sessions.yaml"
|
|
db_file = state_dir / "agent-sessions.db"
|
|
|
|
return {
|
|
"root": tmp_path,
|
|
"ws": ws,
|
|
"yaml": yaml_file,
|
|
"db": db_file,
|
|
"home": home,
|
|
"lib": tmp_path / ".agents" / "skills" / "lib.sh"
|
|
}
|
|
|
|
def run_find_workspace_uuid(sandbox, workspace, agent, env_extra=None, target=None):
|
|
env = dict(os.environ)
|
|
env["AGENT_SESSIONS_YAML"] = str(sandbox["yaml"])
|
|
env["YAML_PATH"] = str(sandbox["yaml"])
|
|
env["WORKSPACE_ROOT"] = str(sandbox["root"])
|
|
env["HOME_DIR"] = str(sandbox["home"])
|
|
env["CLAUDE_PROJECT_DIR"] = str(sandbox["home"] / ".claude" / "projects")
|
|
if env_extra:
|
|
env.update(env_extra)
|
|
|
|
target_arg = f" --target '{target}'" if target else ""
|
|
cmd = f"source {sandbox['lib']} && find_workspace_uuid '{workspace}' '{agent}'{target_arg}"
|
|
res = subprocess.run(["bash", "-c", cmd], capture_output=True, text=True, env=env)
|
|
return res
|
|
|
|
class TestB1Tier3Identity(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp_dir = tempfile.mkdtemp(prefix="mam_b1_test_")
|
|
self.tmp_path = tempfile.TemporaryDirectory()
|
|
self.sandbox = {
|
|
"root": self.tmp_dir,
|
|
"ws": os.path.join(self.tmp_dir, "ws"),
|
|
"yaml": os.path.join(self.tmp_dir, "state", "agent-sessions.yaml"),
|
|
"db": os.path.join(self.tmp_dir, "state", "agent-sessions.db"),
|
|
"home": os.path.join(self.tmp_dir, "home"),
|
|
"lib": LIB_SH
|
|
}
|
|
os.makedirs(self.sandbox["ws"], exist_ok=True)
|
|
os.makedirs(os.path.join(self.sandbox["tmp_dir"] if "tmp_dir" in self.sandbox else self.tmp_dir, "state"), exist_ok=True)
|
|
os.makedirs(os.path.join(self.sandbox["home"], ".claude", "projects"), exist_ok=True)
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.tmp_dir, ignore_errors=True)
|
|
|
|
def test_b1_tier3_honours_agent_sessions_yaml_path(split_sandbox):
|
|
"""V-1: D1 — verify tier-3 honours AGENT_SESSIONS_YAML path when ws != state dir."""
|
|
ws = str(split_sandbox["ws"])
|
|
uuid = "b1-test-uuid-0001"
|
|
|
|
# Create mock agy conversation DB
|
|
db_file = split_sandbox["home"] / ".gemini" / "antigravity-cli" / "conversations" / f"{uuid}.db"
|
|
conn = sqlite3.connect(db_file)
|
|
conn.execute("CREATE TABLE steps (id INT)")
|
|
conn.execute("INSERT INTO steps VALUES (1)")
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
# Write identity info into yaml at state_dir, NOT ws/.mam/
|
|
yaml_content = {
|
|
"herdr_sessions": [],
|
|
"agent_identities": {
|
|
"agy": {
|
|
"project_cwd": ws,
|
|
"conversation_id": uuid
|
|
}
|
|
}
|
|
}
|
|
import yaml
|
|
with open(split_sandbox["yaml"], "w") as f:
|
|
yaml.dump(yaml_content, f)
|
|
|
|
res = run_find_workspace_uuid(split_sandbox, ws, "agy")
|
|
assert res.returncode == 0
|
|
assert res.stdout.strip() == uuid
|
|
|
|
def test_b1_tier3_db_branch_survives_missing_pyyaml(split_sandbox):
|
|
"""V-2: D3 — verify tier-3 DB branch works even when PyYAML module is absent."""
|
|
ws = str(split_sandbox["ws"])
|
|
uuid = "b1-test-uuid-0002"
|
|
|
|
# Create mock agy conversation DB
|
|
db_file = split_sandbox["home"] / ".gemini" / "antigravity-cli" / "conversations" / f"{uuid}.db"
|
|
conn_conv = sqlite3.connect(db_file)
|
|
conn_conv.execute("CREATE TABLE steps (id INT)")
|
|
conn_conv.execute("INSERT INTO steps VALUES (1)")
|
|
conn_conv.commit()
|
|
conn_conv.close()
|
|
|
|
# Create SQLite DB with state data
|
|
conn = sqlite3.connect(split_sandbox["db"])
|
|
conn.execute("CREATE TABLE state (id INTEGER PRIMARY KEY, data TEXT)")
|
|
state_data = json.dumps({
|
|
"herdr_sessions": [],
|
|
"agent_identities": {
|
|
"agy": {
|
|
"project_cwd": ws,
|
|
"conversation_id": uuid
|
|
}
|
|
}
|
|
})
|
|
conn.execute("INSERT INTO state (id, data) VALUES (1, ?)", (state_data,))
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
# Stub out PyYAML via PYTHONPATH override with broken yaml package
|
|
stub_dir = split_sandbox["root"] / "noyaml" / "yaml"
|
|
stub_dir.mkdir(parents=True)
|
|
with open(stub_dir / "__init__.py", "w") as f:
|
|
f.write("raise ImportError('No module named yaml')\n")
|
|
|
|
env_extra = {"PYTHONPATH": str(split_sandbox["root"] / "noyaml")}
|
|
res = run_find_workspace_uuid(split_sandbox, ws, "agy", env_extra=env_extra)
|
|
assert res.returncode == 0
|
|
assert res.stdout.strip() == uuid
|
|
|
|
def test_b1_tier3_corrupt_identities_still_exits_zero(split_sandbox):
|
|
"""V-3: D4 — verify non-dict agent_identities handles gracefully and exits 0."""
|
|
ws = str(split_sandbox["ws"])
|
|
yaml_content = {
|
|
"herdr_sessions": [],
|
|
"agent_identities": "corrupted_string_not_dict"
|
|
}
|
|
import yaml
|
|
with open(split_sandbox["yaml"], "w") as f:
|
|
yaml.dump(yaml_content, f)
|
|
# Also write to ws/.mam/ to trigger HEAD's path-guessing code path
|
|
with open(split_sandbox["ws"] / ".mam" / "agent-sessions.yaml", "w") as f:
|
|
yaml.dump(yaml_content, f)
|
|
|
|
res = run_find_workspace_uuid(split_sandbox, ws, "agy")
|
|
assert res.returncode == 0
|
|
assert res.stdout.strip() == ""
|
|
|
|
def test_b1_tier3_hermes_conversation_id_fallback(split_sandbox):
|
|
"""V-4: D5 — verify hermes tier-3 fallback reads conversation_id from ai_agent."""
|
|
ws = str(split_sandbox["ws"])
|
|
uuid = "b1-test-uuid-hermes-0004"
|
|
|
|
# Create mock hermes state DB
|
|
hermes_dir = split_sandbox["home"] / ".hermes"
|
|
hermes_dir.mkdir(parents=True, exist_ok=True)
|
|
conn_h = sqlite3.connect(hermes_dir / "state.db")
|
|
conn_h.execute("CREATE TABLE sessions (id TEXT, cwd TEXT, started_at TEXT)")
|
|
conn_h.execute("INSERT INTO sessions VALUES (?, ?, ?)", (uuid, ws, "2026-08-05"))
|
|
conn_h.commit()
|
|
conn_h.close()
|
|
|
|
yaml_content = {
|
|
"herdr_sessions": [],
|
|
"agent_identities": {
|
|
"hermes": {
|
|
"project_cwd": ws,
|
|
"conversation_id": uuid
|
|
}
|
|
}
|
|
}
|
|
import yaml
|
|
with open(split_sandbox["yaml"], "w") as f:
|
|
yaml.dump(yaml_content, f)
|
|
with open(split_sandbox["ws"] / ".mam" / "agent-sessions.yaml", "w") as f:
|
|
yaml.dump(yaml_content, f)
|
|
|
|
res = run_find_workspace_uuid(split_sandbox, ws, "hermes")
|
|
assert res.returncode == 0
|
|
assert res.stdout.strip() == uuid
|
|
|
|
def test_b1_tier3_refuses_foreign_workspace_identity(split_sandbox):
|
|
"""V-5: Verify tier-3 identity is ignored if project_cwd does not match workspace."""
|
|
ws = str(split_sandbox["ws"])
|
|
foreign_ws = str(split_sandbox["root"] / "other_ws")
|
|
uuid = "b1-test-uuid-0005"
|
|
yaml_content = {
|
|
"herdr_sessions": [],
|
|
"agent_identities": {
|
|
"agy": {
|
|
"project_cwd": foreign_ws,
|
|
"conversation_id": uuid
|
|
}
|
|
}
|
|
}
|
|
import yaml
|
|
with open(split_sandbox["yaml"], "w") as f:
|
|
yaml.dump(yaml_content, f)
|
|
|
|
res = run_find_workspace_uuid(split_sandbox, ws, "agy")
|
|
assert res.returncode == 0
|
|
assert res.stdout.strip() == ""
|
|
|
|
def test_b1_tier3_absent_identities_is_silent(split_sandbox):
|
|
"""V-6: Verify tier-3 gracefully returns empty string when agent_identities is absent."""
|
|
ws = str(split_sandbox["ws"])
|
|
yaml_content = {"herdr_sessions": []}
|
|
import yaml
|
|
with open(split_sandbox["yaml"], "w") as f:
|
|
yaml.dump(yaml_content, f)
|
|
|
|
res = run_find_workspace_uuid(split_sandbox, ws, "agy")
|
|
assert res.returncode == 0
|
|
assert res.stdout.strip() == ""
|
|
|
|
def test_b1_load_state_json_surfaces_agent_identities(split_sandbox):
|
|
"""V-7: Verify load_state_json preserves agent_identities in state blob."""
|
|
ws = str(split_sandbox["ws"])
|
|
uuid = "b1-test-uuid-0007"
|
|
yaml_content = {
|
|
"herdr_sessions": [],
|
|
"agent_identities": {
|
|
"agy": {
|
|
"project_cwd": ws,
|
|
"conversation_id": uuid
|
|
}
|
|
}
|
|
}
|
|
import yaml
|
|
with open(split_sandbox["yaml"], "w") as f:
|
|
yaml.dump(yaml_content, f)
|
|
|
|
env = dict(os.environ)
|
|
env["AGENT_SESSIONS_YAML"] = str(split_sandbox["yaml"])
|
|
env["YAML_PATH"] = str(split_sandbox["yaml"])
|
|
cmd = f"source {split_sandbox['lib']} && load_state_json"
|
|
res = subprocess.run(["bash", "-c", cmd], capture_output=True, text=True, env=env)
|
|
assert res.returncode == 0
|
|
state = json.loads(res.stdout.strip())
|
|
assert state.get("agent_identities", {}).get("agy", {}).get("conversation_id") == uuid
|
|
|
|
def test_b1_tier3_does_not_read_yaml_mirror_behind_the_db(split_sandbox):
|
|
"""V-8: Verify tier-3 does not prioritize YAML mirror when DB exists without identity."""
|
|
ws = str(split_sandbox["ws"])
|
|
uuid = "b1-test-uuid-0008"
|
|
|
|
# DB has no agent_identities
|
|
conn = sqlite3.connect(split_sandbox["db"])
|
|
conn.execute("CREATE TABLE state (id INTEGER PRIMARY KEY, data TEXT)")
|
|
state_data = json.dumps({"herdr_sessions": []})
|
|
conn.execute("INSERT INTO state (id, data) VALUES (1, ?)", (state_data,))
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
# YAML has agent_identities
|
|
yaml_content = {
|
|
"herdr_sessions": [],
|
|
"agent_identities": {
|
|
"agy": {
|
|
"project_cwd": ws,
|
|
"conversation_id": uuid
|
|
}
|
|
}
|
|
}
|
|
import yaml
|
|
with open(split_sandbox["yaml"], "w") as f:
|
|
yaml.dump(yaml_content, f)
|
|
|
|
res = run_find_workspace_uuid(split_sandbox, ws, "agy")
|
|
assert res.returncode == 0
|
|
# Must miss because DB is authority and state row lacks identity
|
|
assert res.stdout.strip() == ""
|