diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..5c3586d --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,539 @@ +import importlib +import itertools +import json +import re +import sys +import types +from collections import deque +from concurrent.futures import Future +from dataclasses import dataclass +from pathlib import Path + +import pytest + + +ROOT = Path(__file__).parent.parent +FRAMEWORK = ROOT / "framework" +MODULES = ( + "agent", + "critic", + "engineer", + "secretary", + "tools", + "transfer", + "framework.agent", + "framework.critic", + "framework.engineer", + "framework.secretary", + "framework.tools", + "framework.transfer", + "framework.lab_config", + "framework.watch", + "framework.slack_to_board", +) + + +@dataclass +class Turn: + messages: list + expected_query: object = None + query_error: Exception | None = None + stream_error: Exception | None = None + stream_error_after: int | None = None + + +class FakeClaude: + def __init__(self): + self.client_plans = deque() + self.clients = [] + self.options = [] + self.agent_definitions = [] + self.tools = [] + self.mcp_servers = [] + self.module = types.ModuleType("claude_agent_sdk") + self._build_module() + + def _build_module(self): + harness = self + + class AssistantMessage: + def __init__(self, content=None, parent_tool_use_id=None): + self.content = list(content or []) + self.parent_tool_use_id = parent_tool_use_id + + class ResultMessage: + def __init__(self, subtype="success", session_id=None): + self.subtype = subtype + self.session_id = session_id + + class ClaudeAgentOptions: + def __init__(self, **kwargs): + self.kwargs = dict(kwargs) + self.__dict__.update(kwargs) + harness.options.append(self) + + class AgentDefinition: + def __init__(self, **kwargs): + self.kwargs = dict(kwargs) + self.__dict__.update(kwargs) + harness.agent_definitions.append(self) + + class ClaudeSDKClient: + def __init__(self, *, options): + self.options = options + self.queries = [] + self.context_calls = 0 + self.receive_calls = 0 + self.enter_count = 0 + self.exit_calls = [] + self._active_turn = None + self._response_read = False + self._plan = ( + harness.client_plans.popleft() + if harness.client_plans + else { + "turns": deque(), + "contexts": deque(), + "enter_error": None, + "exit_error": None, + } + ) + harness.clients.append(self) + + async def __aenter__(self): + self.enter_count += 1 + if self._plan["enter_error"]: + raise self._plan["enter_error"] + return self + + async def __aexit__(self, exc_type, exc, traceback): + self.exit_calls.append((exc_type, exc, traceback)) + if self._plan["exit_error"]: + raise self._plan["exit_error"] + return False + + async def query(self, prompt): + self.queries.append(prompt) + if not self._plan["turns"]: + raise AssertionError(f"unexpected Claude query: {prompt!r}") + turn = self._plan["turns"].popleft() + expected = turn.expected_query + if isinstance(expected, str) and prompt != expected: + raise AssertionError(f"expected query {expected!r}, got {prompt!r}") + if hasattr(expected, "search") and not expected.search(prompt): + raise AssertionError( + f"query did not match {expected.pattern!r}: {prompt!r}" + ) + if callable(expected) and not expected(prompt): + raise AssertionError(f"query predicate rejected: {prompt!r}") + if turn.query_error: + raise turn.query_error + self._active_turn = turn + self._response_read = False + + async def receive_response(self): + self.receive_calls += 1 + if self._active_turn is None or self._response_read: + raise AssertionError("receive_response requires one unread query") + self._response_read = True + turn = self._active_turn + self._active_turn = None + for index, message in enumerate(turn.messages): + if turn.stream_error_after == index: + raise turn.stream_error + yield message + if turn.stream_error_after == len(turn.messages): + raise turn.stream_error + + async def get_context_usage(self): + self.context_calls += 1 + contexts = self._plan["contexts"] + value = contexts.popleft() if contexts else None + if callable(value): + value = value() + if hasattr(value, "__await__"): + value = await value + if isinstance(value, BaseException): + raise value + return value + + def tool(name, description, schema): + def decorate(fn): + fn._tool_name = name + fn._tool_description = description + fn._tool_schema = schema + harness.tools.append(fn) + return fn + + return decorate + + def create_sdk_mcp_server(**kwargs): + server = types.SimpleNamespace(**kwargs) + harness.mcp_servers.append(server) + return server + + self.module.AssistantMessage = AssistantMessage + self.module.ResultMessage = ResultMessage + self.module.ClaudeAgentOptions = ClaudeAgentOptions + self.module.AgentDefinition = AgentDefinition + self.module.ClaudeSDKClient = ClaudeSDKClient + self.module.tool = tool + self.module.create_sdk_mcp_server = create_sdk_mcp_server + + def plan_client(self, *, turns=(), contexts=(), enter_error=None, exit_error=None): + self.client_plans.append( + { + "turns": deque(turns), + "contexts": deque(contexts), + "enter_error": enter_error, + "exit_error": exit_error, + } + ) + + def turn( + self, + *messages, + expected_query=None, + query_error=None, + stream_error=None, + stream_error_after=None, + ): + return Turn( + list(messages), + expected_query, + query_error, + stream_error, + stream_error_after, + ) + + def text(self, text): + return types.SimpleNamespace(text=text) + + def tool_use(self, name, input=None, id=None): + return types.SimpleNamespace(name=name, input=input or {}, id=id) + + def assistant(self, *blocks, parent_tool_use_id=None): + return self.module.AssistantMessage(blocks, parent_tool_use_id) + + def result(self, subtype="success", session_id=None): + return self.module.ResultMessage(subtype, session_id) + + @staticmethod + def context(*, tokens, window, percentage, model): + return { + "totalTokens": tokens, + "rawMaxTokens": window, + "percentage": percentage, + "model": model, + } + + @property + def last_client(self): + return self.clients[-1] + + +class FakeGlobusFuture(Future): + def __init__(self, task_id=None): + super().__init__() + self.task_id = task_id + + @classmethod + def pending(cls, task_id=None): + return cls(task_id) + + @classmethod + def successful(cls, value, task_id=None): + future = cls(task_id) + future.set_result(value) + return future + + @classmethod + def failed(cls, error, task_id=None): + future = cls(task_id) + future.set_exception(error) + return future + + +@dataclass +class SubmitPlan: + kind: str + value: object = None + task_id: str | None = None + + +class FakeGlobus: + def __init__(self): + self.submit_plans = deque() + self.endpoint_status = {"status": "online"} + self.endpoint_metadata = {"name": "fake-endpoint"} + self.task_statuses = {} + self.executor_inits = [] + self.serializer_inits = [] + self.strategy_inits = [] + self.submits = [] + self.shutdowns = [] + self.client_inits = [] + self.endpoint_status_calls = [] + self.endpoint_metadata_calls = [] + self.task_calls = [] + self.futures = [] + self._task_ids = itertools.count(1) + self.module, self.serialize_module = self._build_modules() + + def _build_modules(self): + harness = self + root = types.ModuleType("globus_compute_sdk") + serialize = types.ModuleType("globus_compute_sdk.serialize") + + class AllCodeStrategies: + def __init__(self): + harness.strategy_inits.append(self) + + class ComputeSerializer: + def __init__(self, *, strategy_code): + self.strategy_code = strategy_code + harness.serializer_inits.append(self) + + class Executor: + def __init__(self, *, endpoint_id, user_endpoint_config): + self.endpoint_id = endpoint_id + self.user_endpoint_config = user_endpoint_config + self.serializer = None + self._stopped = False + harness.executor_inits.append(self) + + def submit(self, fn, args, target): + harness.submits.append((self, fn, args, target)) + plan = ( + harness.submit_plans.popleft() + if harness.submit_plans + else SubmitPlan("pending") + ) + if plan.kind == "raise": + raise plan.value + task_id = plan.task_id or f"fake-task-{next(harness._task_ids)}" + if plan.kind == "pending": + future = FakeGlobusFuture.pending(task_id) + elif plan.kind == "success": + future = FakeGlobusFuture.successful(plan.value, task_id) + elif plan.kind == "error": + future = FakeGlobusFuture.failed(plan.value, task_id) + else: + raise AssertionError(f"unknown submit plan {plan.kind}") + harness.futures.append(future) + return future + + def shutdown(self, *, wait): + self._stopped = True + harness.shutdowns.append((self, wait)) + + class Client: + def __init__(self): + harness.client_inits.append(self) + + def get_endpoint_status(self, endpoint_id): + harness.endpoint_status_calls.append(endpoint_id) + return harness._resolve(harness.endpoint_status, endpoint_id) + + def get_endpoint_metadata(self, endpoint_id): + harness.endpoint_metadata_calls.append(endpoint_id) + return harness._resolve(harness.endpoint_metadata, endpoint_id) + + def get_task(self, task_id): + harness.task_calls.append(task_id) + return harness._resolve( + harness.task_statuses.get(task_id, {"status": "running"}), task_id + ) + + root.Executor = Executor + root.Client = Client + root.serialize = serialize + serialize.ComputeSerializer = ComputeSerializer + serialize.AllCodeStrategies = AllCodeStrategies + return root, serialize + + @staticmethod + def _resolve(value, *args): + value = value(*args) if callable(value) else value + if isinstance(value, BaseException): + raise value + return value + + def plan_pending(self, task_id=None): + self.submit_plans.append(SubmitPlan("pending", task_id=task_id)) + + def plan_success(self, value, task_id=None): + self.submit_plans.append(SubmitPlan("success", value, task_id)) + + def plan_future_error(self, error, task_id=None): + self.submit_plans.append(SubmitPlan("error", error, task_id)) + + def plan_submit_error(self, error): + self.submit_plans.append(SubmitPlan("raise", error)) + + +class FakeTask: + def __init__(self, name, *, remote=True, local=False, bucket=False): + self.module = types.ModuleType(name) + self.job_key_calls = [] + self.bucket_calls = [] + self.remote_calls = [] + self.local_calls = [] + self.preflight_result = None + if remote: + self.module.JOB_DESC = "fake remote job" + self.module.JOB_SCHEMA = {"value": int} + self.module.job_key = self.job_key + self.module.remote_fn = self.remote_fn + if local: + self.module.LOCAL_DESC = "fake local job" + self.module.LOCAL_SCHEMA = {"value": int} + self.module.local_fn = self.local_fn + if bucket: + self.module.bucket_for = self.bucket_for + + def job_key(self, args): + self.job_key_calls.append(args) + return f"value={args.get('value')}" + + def remote_fn(self, args, target): + self.remote_calls.append((args, target)) + return {"value": args.get("value")} + + def local_fn(self, args): + self.local_calls.append(args) + return {"value": args.get("value")} + + def bucket_for(self, args): + self.bucket_calls.append(args) + return args.get("bucket", "default") + + +@dataclass +class ToolsLab: + root: Path + campaign_dir: Path + workspace: Path + tools: object + globus: FakeGlobus + task: FakeTask + + +@pytest.fixture +def fake_claude_sdk(monkeypatch): + fake = FakeClaude() + monkeypatch.setitem(sys.modules, "claude_agent_sdk", fake.module) + return fake + + +@pytest.fixture +def fake_globus_sdk(monkeypatch): + fake = FakeGlobus() + monkeypatch.setitem(sys.modules, "globus_compute_sdk", fake.module) + monkeypatch.setitem( + sys.modules, "globus_compute_sdk.serialize", fake.serialize_module + ) + return fake + + +@pytest.fixture +def import_module(monkeypatch, fake_claude_sdk): + def load(name, env=None): + for module_name in MODULES: + sys.modules.pop(module_name, None) + for key, value in (env or {}).items(): + monkeypatch.setenv(key, value) + monkeypatch.syspath_prepend(str(FRAMEWORK)) + return importlib.import_module(name) + + return load + + +@pytest.fixture +def transfer_module(import_module): + return import_module("transfer") + + +@pytest.fixture +def tools_lab(tmp_path, monkeypatch, fake_claude_sdk, fake_globus_sdk): + counter = itertools.count(1) + loaded = [] + + def load(*, remote=True, local=False, bucket=False, task_mutator=None, env=None): + index = next(counter) + root = tmp_path / f"lab-{index}" + campaign_dir = root / "campaigns" / "test-campaign" + workspace = root / "workspace" / "test-campaign" + system_dir = root / "systems" + campaign_dir.mkdir(parents=True) + workspace.mkdir(parents=True) + system_dir.mkdir() + + campaign = { + "system": "test-system", + "max_concurrent": 2, + "local_max_concurrent": 2, + "target": {"env": {"CAMPAIGN_ENV": "yes"}}, + } + system = { + "ppn": 4, + "target": {"env": {"SYSTEM_ENV": "yes"}}, + "bucket_defaults": { + "queue": "test", + "walltime": "00:10:00", + "num_nodes": 1, + }, + } + (campaign_dir / "campaign.json").write_text(json.dumps(campaign)) + (system_dir / "test-system.json").write_text(json.dumps(system)) + for name in ("prompt.md", "user_prompt.md", "method.md"): + (campaign_dir / name).write_text(name) + + if remote: + user_dir = root / "users" / "test-user" + user_dir.mkdir(parents=True) + (user_dir / "test-system.json").write_text( + json.dumps( + { + "endpoint": "test-endpoint", + "account": "test-account", + "work_dir": "/remote/work", + } + ) + ) + + task_name = f"_agentlab_test_task_{index}" + task = FakeTask(task_name, remote=remote, local=local, bucket=bucket) + if task_mutator: + task_mutator(task.module) + monkeypatch.setitem(sys.modules, task_name, task.module) + + for module_name in MODULES: + sys.modules.pop(module_name, None) + monkeypatch.syspath_prepend(str(FRAMEWORK)) + values = { + "LAB_DIR": str(root), + "CAMPAIGN": "test-campaign", + "CAMPAIGN_DIR": str(campaign_dir), + "WORKSPACE_DIR": str(workspace), + "USER_NAME": "test-user", + "TASK_DIR": str(campaign_dir), + "TASK_MODULE": task_name, + "MAX_SUBMITS": "10", + "MAX_CONCURRENT": "2", + "LOCAL_MAX_CONCURRENT": "2", + "RUN_ID": "test-run", + "NOTIFY_SCRIPT": "", + } + values.update(env or {}) + for key, value in values.items(): + monkeypatch.setenv(key, str(value)) + tools = importlib.import_module("tools") + lab = ToolsLab(root, campaign_dir, workspace, tools, fake_globus_sdk, task) + loaded.append(lab) + return lab + + yield load + + for lab in loaded: + lab.tools.shutdown_executor() diff --git a/tests/test_agent_fixtures.py b/tests/test_agent_fixtures.py new file mode 100644 index 0000000..c5de96b --- /dev/null +++ b/tests/test_agent_fixtures.py @@ -0,0 +1,45 @@ +import asyncio +import importlib +import sys + + +def test_agent_drain_turn_tracks_delegate_phase_and_session( + tools_lab, fake_claude_sdk, monkeypatch, capsys +): + lab = tools_lab(remote=False, local=True) + sys.modules.pop("agent", None) + agent = importlib.import_module("agent") + agent.RUN_DIR = str(lab.workspace / "runs" / "test") + (lab.workspace / "runs" / "test").mkdir(parents=True) + phases = [] + monkeypatch.setattr(agent, "_set_phase", phases.append) + fake_claude_sdk.plan_client( + turns=[ + fake_claude_sdk.turn( + fake_claude_sdk.assistant( + fake_claude_sdk.tool_use( + "Agent", {"subagent_type": "reader"}, "call-1" + ), + ), + fake_claude_sdk.assistant( + fake_claude_sdk.tool_use("mcp__cas__submit_local"), + parent_tool_use_id="call-1", + ), + fake_claude_sdk.result("success", "agent-session"), + ) + ], + contexts=[None], + ) + client = fake_claude_sdk.module.ClaudeSDKClient(options=object()) + + async def run(): + await client.query("turn") + await agent.drain_turn(client, 2) + await agent._context_task + + asyncio.run(run()) + + assert any("waiting on subagent reader" in phase for phase in phases) + assert any("subagent reader (submit_local)" in phase for phase in phases) + assert agent._session_id == "agent-session" + assert "[round 2 turn end] success" in capsys.readouterr().out diff --git a/tests/test_claude_harness.py b/tests/test_claude_harness.py new file mode 100644 index 0000000..55f220d --- /dev/null +++ b/tests/test_claude_harness.py @@ -0,0 +1,58 @@ +import asyncio +import importlib + + +def test_engineer_answer_records_query_and_session( + fake_claude_sdk, monkeypatch, tmp_path, capsys +): + monkeypatch.setenv("LAB_DIR", str(tmp_path)) + monkeypatch.setenv("SLACK_INBOX", str(tmp_path / "run" / "inbox.md")) + sys_module = importlib.import_module("sys") + sys_module.modules.pop("engineer", None) + monkeypatch.syspath_prepend(str(tmp_path.parent / "AgentLab" / "framework")) + engineer = importlib.import_module("framework.engineer") + engineer.SESSION_FILE = str(tmp_path / "run" / "session") + fake_claude_sdk.plan_client( + turns=[ + fake_claude_sdk.turn( + fake_claude_sdk.assistant(fake_claude_sdk.text("answer")), + fake_claude_sdk.result("success", "session-1"), + expected_query="question\n\nAnswer it, then stop.", + ) + ] + ) + client = fake_claude_sdk.module.ClaudeSDKClient(options=object()) + + asyncio.run(engineer.answer(client, "question")) + + assert fake_claude_sdk.last_client.queries == ["question\n\nAnswer it, then stop."] + assert (tmp_path / "run" / "session").read_text() == f"session-1\n{tmp_path}\n" + assert "answer" in capsys.readouterr().out + + +def test_secretary_answer_includes_live_agent_status( + fake_claude_sdk, monkeypatch, tmp_path +): + monkeypatch.setenv("WORKSPACE_ROOT", str(tmp_path)) + sys_module = importlib.import_module("sys") + sys_module.modules.pop("framework.secretary", None) + secretary = importlib.import_module("framework.secretary") + (tmp_path / "run").mkdir() + fake_claude_sdk.plan_client( + turns=[ + fake_claude_sdk.turn( + fake_claude_sdk.result("success", "session-2"), + expected_query=lambda prompt: ( + "Research agents running:\nagent1" in prompt + ), + ) + ] + ) + client = fake_claude_sdk.module.ClaudeSDKClient(options=object()) + + asyncio.run(secretary.answer(client, "status?", ["agent1 -- campaign demo"])) + + assert "New from Slack:\n\nstatus?" in fake_claude_sdk.last_client.queries[0] + assert ( + tmp_path / "run" / "secretary_session" + ).read_text() == f"session-2\n{tmp_path}\n" diff --git a/tests/test_critic.py b/tests/test_critic.py new file mode 100644 index 0000000..0d9b91e --- /dev/null +++ b/tests/test_critic.py @@ -0,0 +1,53 @@ +import types + + +def test_resolve_explicit_and_auto(import_module, monkeypatch): + critic = import_module("critic", {"CRITIC_MODEL": "auto"}) + monkeypatch.setattr( + critic, "_served_models", lambda: {"gpt": "openai/gpt", "claude": "claude-3"} + ) + assert critic.resolve("claude-sonnet")[0] == "gpt" + + critic.MODEL_SETTING = "missing" + monkeypatch.setattr(critic, "_served_models", lambda: {"gpt": "openai/gpt"}) + try: + critic.resolve() + except critic.CriticUnavailable as exc: + assert "not served" in str(exc) + else: + raise AssertionError("expected unavailable critic") + + +def test_blocking_only_returns_structured_blocking_findings(import_module): + critic = import_module("critic") + reply = """CLAIM: x is lower +VERDICT: unsupported +SEVERITY: blocking + +CLAIM: y improved +VERDICT: supported +SEVERITY: minor""" + assert critic.blocking(reply) == [("x is lower", "unsupported")] + assert critic.blocking("plain prose") == [] + + +def test_review_builds_request_and_joins_content(import_module, monkeypatch): + critic = import_module("critic", {"CRITIC_API_KEY": "secret"}) + request = types.SimpleNamespace() + request.data = None + + class Response: + def __enter__(self): + return self + + def __exit__(self, *_): + return None + + response = Response() + monkeypatch.setattr(critic.urllib.request, "urlopen", lambda req, timeout: response) + monkeypatch.setattr( + critic.json, "load", lambda _: {"content": [{"text": "a"}, {"text": " b "}]} + ) + result = critic.review("model", "write-up", "evidence") + assert result == "a b" + assert request.data is None diff --git a/tests/test_lab_config.py b/tests/test_lab_config.py new file mode 100644 index 0000000..7ed8788 --- /dev/null +++ b/tests/test_lab_config.py @@ -0,0 +1,55 @@ +import importlib +from pathlib import Path + + +def test_load_strips_comments_and_resolves_path_values(tmp_path, monkeypatch): + monkeypatch.setenv("LAB_DIR", str(tmp_path)) + module = importlib.import_module("framework.lab_config") + config = tmp_path / "lab.yaml" + config.write_text(""" + # comment + litellm-bin: ~/bin/litellm # inline + litellm-url: http://localhost:4000 + malformed + key: value:with:colons + """) + + loaded = module.load(config) + assert loaded["litellm-bin"] == str(Path.home() / "bin" / "litellm") + assert loaded["key"] == "value:with:colons" + + +def test_start_command_and_service_selection(tmp_path, monkeypatch): + monkeypatch.setenv("LAB_DIR", str(tmp_path)) + module = importlib.import_module("framework.lab_config") + config = { + "litellm-bin": "/bin/litellm", + "litellm-config": "/tmp/config.yaml", + "litellm-url": "http://localhost:4000", + "bridge": "yes", + "secretary": "off", + } + assert ( + module.start_command(config) + == "/bin/litellm --config /tmp/config.yaml --port 4000" + ) + assert module.on("bridge", config) + assert not module.on("secretary", config) + + +def test_main_get_services_and_export(tmp_path, monkeypatch, capsys): + monkeypatch.setenv("LAB_DIR", str(tmp_path)) + (tmp_path / "lab.yaml").write_text("slack-channel: C123\nbridge: on\n") + module = importlib.import_module("framework.lab_config") + module.load = lambda path=str(tmp_path / "lab.yaml"): { + "slack-channel": "C123", + "bridge": "on", + } + + monkeypatch.setattr(module.sys, "argv", ["lab_config.py", "--get", "slack-channel"]) + module.main() + assert capsys.readouterr().out.strip() == "C123" + + monkeypatch.setattr(module.sys, "argv", ["lab_config.py", "--services"]) + module.main() + assert capsys.readouterr().out.strip() == "bridge" diff --git a/tests/test_slack_to_board.py b/tests/test_slack_to_board.py new file mode 100644 index 0000000..92281ca --- /dev/null +++ b/tests/test_slack_to_board.py @@ -0,0 +1,47 @@ +import importlib + + +def test_forward_orders_messages_and_routes_to_campaigns(tmp_path, monkeypatch): + monkeypatch.setenv("WORKSPACE_ROOT", str(tmp_path)) + module = importlib.import_module("framework.slack_to_board") + (tmp_path / "alpha").mkdir() + (tmp_path / "beta").mkdir() + monkeypatch.setattr(module, "DEDICATED", False) + monkeypatch.setattr(module, "READ_ALL", False) + delivered = module.forward( + [ + {"user": "u2", "text": "<@BOT> beta second"}, + {"user": "u1", "text": "<@BOT> alpha first"}, + ], + "BOT", + ) + assert delivered == 2 + assert "alpha first" in (tmp_path / "alpha" / "ANNOUNCEMENTS.md").read_text() + assert "beta second" in (tmp_path / "beta" / "ANNOUNCEMENTS.md").read_text() + + +def test_forward_skips_bots_and_unaddressed_messages(tmp_path, monkeypatch): + monkeypatch.setenv("WORKSPACE_ROOT", str(tmp_path)) + module = importlib.import_module("framework.slack_to_board") + (tmp_path / "alpha").mkdir() + delivered = module.forward( + [ + {"bot_id": "B", "text": "<@BOT> bot"}, + {"user": "u", "text": "ordinary"}, + ], + "BOT", + ) + assert delivered == 0 + assert not (tmp_path / "alpha" / "ANNOUNCEMENTS.md").exists() + + +def test_check_first_run_does_not_fetch_history(tmp_path, monkeypatch): + monkeypatch.setenv("WORKSPACE_ROOT", str(tmp_path)) + module = importlib.import_module("framework.slack_to_board") + called = [] + monkeypatch.setattr( + module, "slack_get", lambda *args, **kwargs: called.append(args) + ) + module.check("token", "BOT") + assert called == [] + assert module.read_state() diff --git a/tests/test_tools.py b/tests/test_tools.py new file mode 100644 index 0000000..37a4c42 --- /dev/null +++ b/tests/test_tools.py @@ -0,0 +1,67 @@ +import asyncio +import json + + +def payload(result): + return json.loads(result["content"][0]["text"]) + + +def test_remote_submit_and_collect_success(tools_lab): + lab = tools_lab(remote=True) + lab.globus.plan_success({"metric": 7}, task_id="task-7") + + submitted = asyncio.run(lab.tools.submit_job({"value": 7})) + assert payload(submitted)["key"] == "value=7" + assert lab.globus.submits[0][3]["nranks"] == 4 + assert len(lab.globus.executor_inits) == 1 + + completed = payload(asyncio.run(lab.tools.get_completed_jobs({}))) + assert completed["completed"] == [{"metric": 7, "job_id": 1, "key": "value=7"}] + assert completed["pending"] == [] + assert lab.tools.jobs_in_flight() == 0 + assert '"event": "submit"' in (lab.workspace / "jobs.jsonl").read_text() + + +def test_submit_failure_releases_claim(tools_lab): + lab = tools_lab(remote=True) + lab.globus.plan_submit_error(RuntimeError("endpoint down")) + + result = asyncio.run(lab.tools.submit_job({"value": 1})) + + assert result["is_error"] + assert lab.tools.submit_count() == 0 + claims = (lab.workspace / "claims.jsonl").read_text() + assert '"state": "claimed"' in claims + assert '"state": "done"' in claims + + +def test_local_only_needs_no_user_file_and_exposes_local_tools(tools_lab): + lab = tools_lab(remote=False, local=True) + + assert lab.tools.HAS_LOCAL + assert not lab.tools.HAS_REMOTE + assert lab.tools.ENDPOINT_ID == "" + assert lab.tools.tool_names() == [ + "mcp__cas__submit_local", + "mcp__cas__get_local_completed", + "mcp__cas__notify", + "mcp__cas__cycle_done", + "mcp__cas__goal_met", + ] + + +def test_backend_trouble_requires_all_pending_tasks_to_fail(tools_lab): + lab = tools_lab(remote=True) + lab.globus.plan_pending("failed-task") + asyncio.run(lab.tools.submit_job({"value": 1})) + lab.globus.task_statuses["failed-task"] = {"status": "failed"} + assert "all 1 in-flight task(s) failed/lost" in lab.tools.backend_trouble() + + +def test_goal_met_stops_new_work(tools_lab): + lab = tools_lab(remote=True) + asyncio.run(lab.tools.goal_met({"reason": "enough evidence"})) + refused = asyncio.run(lab.tools.submit_job({"value": 3})) + assert lab.tools.goal_is_met() == "enough evidence" + assert refused["is_error"] + assert "winding down" in refused["content"][0]["text"] diff --git a/tests/test_transfer.py b/tests/test_transfer.py new file mode 100644 index 0000000..a2724e7 --- /dev/null +++ b/tests/test_transfer.py @@ -0,0 +1,61 @@ +import asyncio + + +def test_configure_and_path_confinement(transfer_module, tmp_path): + cfg = transfer_module.configure( + { + "work_dir": "/remote/work", + "globus": {"remote_collection": "r", "local_collection": "l"}, + }, + str(tmp_path), + str(tmp_path / "campaign"), + {"globus_collection_root": "/projects"}, + ) + assert cfg["remote_write_root"] == "/remote/work" + assert cfg["collection_root"] == ["/projects"] + assert transfer_module._local_dest("file.txt", [str(tmp_path)]) == str( + tmp_path / "file.txt" + ) + assert transfer_module._local_dest("../secret", [str(tmp_path)]) is None + + +def test_transfer_requires_configuration_and_authentication( + transfer_module, monkeypatch +): + transfer_module.CFG = None + result = asyncio.run(transfer_module.transfer({"op": "ls", "path": "/"})) + assert result["is_error"] + assert "not configured" in result["content"][0]["text"] + + transfer_module.CFG = { + "remote_collection": "r", + "local_collection": "l", + "remote_write_root": "/work", + "remote_read_root": "", + "collection_root": [], + "local_roots": ["/tmp"], + } + monkeypatch.setattr( + transfer_module, "_authenticated", lambda: (False, "login required") + ) + result = asyncio.run(transfer_module.transfer({"op": "ls", "path": "/"})) + assert "login required" in result["content"][0]["text"] + + +def test_ls_and_unknown_operation(transfer_module, monkeypatch): + transfer_module.CFG = { + "remote_collection": "r", + "local_collection": "l", + "remote_write_root": "/work", + "remote_read_root": "", + "collection_root": [], + "local_roots": ["/tmp"], + } + monkeypatch.setattr(transfer_module, "_authenticated", lambda: (True, "user")) + monkeypatch.setattr( + transfer_module, "_globus", lambda *args, **kwargs: (0, "a\nb", "") + ) + result = asyncio.run(transfer_module.transfer({"op": "ls", "path": "/data"})) + assert result["content"][0]["text"] == "a\nb" + result = asyncio.run(transfer_module.transfer({"op": "wat", "path": "x"})) + assert result["is_error"] diff --git a/tests/test_watch.py b/tests/test_watch.py new file mode 100644 index 0000000..0fcd82d --- /dev/null +++ b/tests/test_watch.py @@ -0,0 +1,42 @@ +import importlib +import json + + +def test_newest_run_prefers_live_run(tmp_path, monkeypatch): + monkeypatch.setenv("LAB_DIR", str(tmp_path)) + module = importlib.import_module("framework.watch") + live = tmp_path / "workspace" / "camp" / "runs" / "live" + finished = tmp_path / "workspace" / "camp" / "runs" / "finished" + live.mkdir(parents=True) + finished.mkdir(parents=True) + (live / "meta.json").write_text("{}") + (finished / "meta.json").write_text("{}") + (live / "heartbeat").write_text("1") + assert module.newest_run("camp") == str(live) + + +def test_status_counts_jobs_and_results(tmp_path, monkeypatch): + monkeypatch.setenv("LAB_DIR", str(tmp_path)) + module = importlib.import_module("framework.watch") + module.LAB_DIR = str(tmp_path) + run = tmp_path / "workspace" / "camp" / "runs" / "r" + run.mkdir(parents=True) + (run / "meta.json").write_text(json.dumps({"run_id": "r", "status": "stopped"})) + ws = tmp_path / "workspace" / "camp" + (ws / "jobs.jsonl").write_text( + '{"event":"submit","run":"r"}\n{"event":"completed","run":"r"}\n' + ) + (ws / "results.jsonl").write_text("one\ntwo\n") + assert module.status("camp")["jobs_run"] == 1 + assert module.status("camp")["jobs_done"] == 1 + assert module.status("camp")["results"] == 2 + + +def test_render_and_file_safety(tmp_path, monkeypatch): + monkeypatch.setenv("LAB_DIR", str(tmp_path)) + module = importlib.import_module("framework.watch") + rendered = module._render("# Heading\n\n![plot](figure.png)") + assert "figure.png" in rendered + handler = object.__new__(module.Handler) + handler.campaign = "camp" + assert handler._file("secret.txt") == "not a file this watcher serves"