"""Unit tests for src/my_deepagent/session.py. Tests verify the dataclass-based deepagents API (FilesystemPermission attributes, build_backend backend type dispatch, _map_operations deduplication, etc.). No real API calls are made. """ from __future__ import annotations from pathlib import Path from typing import Any import pytest from deepagents import FilesystemPermission from deepagents.backends import ( CompositeBackend, FilesystemBackend, LocalShellBackend, ) from langchain_openai import ChatOpenAI from langgraph.graph.state import CompiledStateGraph from my_deepagent.config import load_config from my_deepagent.errors import MyDeepAgentError from my_deepagent.persona import FilesystemPermissionSpec, Persona, PersonaSubagent from my_deepagent.session import ( _map_operations, _resolve_openrouter_api_key, _spec_to_permission, _subagent_to_dict, build_agent, build_backend, default_safety_permissions, resolve_model_instance, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _minimal_persona(**overrides: Any) -> Persona: base: dict[str, Any] = { "name": "test-persona", "version": 1, "backend": "openrouter", "model": "openrouter:anthropic/claude-sonnet-4-6", "provider_origin": "US/Anthropic", "capabilities": ["spec_write"], "max_risk_level": "low", "system_prompt": "You are a test assistant for unit tests.", } base.update(overrides) return Persona.model_validate(base) def _minimal_permission_spec( operations: list[str] | None = None, paths: list[str] | None = None, mode: str = "allow", ) -> FilesystemPermissionSpec: return FilesystemPermissionSpec( operations=tuple(operations or ["read"]), paths=tuple(paths or ["/**"]), mode=mode, ) def _minimal_subagent(**overrides: Any) -> PersonaSubagent: base: dict[str, Any] = { "name": "test-sub", "description": "A test subagent description.", "system_prompt": "You are a subagent for unit tests.", } base.update(overrides) return PersonaSubagent.model_validate(base) # --------------------------------------------------------------------------- # default_safety_permissions — dataclass attribute access # --------------------------------------------------------------------------- def test_default_safety_permissions_returns_two_entries() -> None: perms = default_safety_permissions() assert len(perms) == 2 def test_default_safety_permissions_returns_filesystem_permission_instances() -> None: perms = default_safety_permissions() for p in perms: assert isinstance(p, FilesystemPermission) def test_default_safety_permissions_allow_is_first() -> None: perms = default_safety_permissions() assert perms[0].mode == "allow" assert "/**" in perms[0].paths def test_default_safety_permissions_allow_has_both_operations() -> None: perms = default_safety_permissions() assert "read" in perms[0].operations assert "write" in perms[0].operations def test_default_safety_permissions_deny_is_second() -> None: perms = default_safety_permissions() assert perms[1].mode == "deny" deny_paths = perms[1].paths assert any("env" in p for p in deny_paths) assert any("ssh" in p for p in deny_paths) def test_default_safety_permissions_deny_covers_secrets() -> None: perms = default_safety_permissions() deny_paths = perms[1].paths assert any("secret" in p for p in deny_paths) assert any("token" in p for p in deny_paths) assert any("pem" in p for p in deny_paths) # --------------------------------------------------------------------------- # _map_operations — 8 케이스 # --------------------------------------------------------------------------- def test_map_operations_read() -> None: assert _map_operations(("read",)) == ["read"] def test_map_operations_write() -> None: assert _map_operations(("write",)) == ["write"] def test_map_operations_edit_maps_to_write() -> None: assert _map_operations(("edit",)) == ["write"] def test_map_operations_ls_maps_to_read() -> None: assert _map_operations(("ls",)) == ["read"] def test_map_operations_deduplicates_all_four() -> None: result = _map_operations(("read", "write", "edit", "ls")) assert result == ["read", "write"] def test_map_operations_ls_and_edit() -> None: assert _map_operations(("ls", "edit")) == ["read", "write"] def test_map_operations_preserves_order_write_then_read() -> None: result = _map_operations(("write", "read")) assert result == ["write", "read"] def test_map_operations_empty_returns_empty() -> None: assert _map_operations(()) == [] # --------------------------------------------------------------------------- # _spec_to_permission — dataclass attribute + mapping # --------------------------------------------------------------------------- def test_spec_to_permission_returns_filesystem_permission() -> None: spec = _minimal_permission_spec(operations=["read"], paths=["/**"], mode="allow") result = _spec_to_permission(spec) assert isinstance(result, FilesystemPermission) def test_spec_to_permission_maps_read_write_correctly() -> None: spec = _minimal_permission_spec(operations=["read", "write"], paths=["/**"], mode="allow") result = _spec_to_permission(spec) assert result.operations == ["read", "write"] assert result.paths == ["/**"] assert result.mode == "allow" def test_spec_to_permission_maps_edit_to_write() -> None: spec = _minimal_permission_spec(operations=["edit"], paths=["/src/**"], mode="allow") result = _spec_to_permission(spec) assert result.operations == ["write"] def test_spec_to_permission_maps_ls_to_read() -> None: spec = _minimal_permission_spec(operations=["ls"], paths=["/data/**"], mode="allow") result = _spec_to_permission(spec) assert result.operations == ["read"] def test_spec_to_permission_deduplicates_read_edit_ls() -> None: spec = _minimal_permission_spec( operations=["read", "edit", "ls"], paths=["/workspace/**"], mode="allow" ) result = _spec_to_permission(spec) # read=read, edit=write, ls=read → ["read", "write"] assert result.operations == ["read", "write"] def test_spec_to_permission_deny_mode_passthrough() -> None: spec = _minimal_permission_spec(operations=["read"], paths=["/.env*"], mode="deny") result = _spec_to_permission(spec) assert result.mode == "deny" assert "/.env*" in result.paths # --------------------------------------------------------------------------- # _subagent_to_dict # --------------------------------------------------------------------------- def test_subagent_to_dict_required_fields() -> None: sub = _minimal_subagent() d = _subagent_to_dict(sub) assert d["name"] == "test-sub" assert d["description"] == "A test subagent description." assert d["system_prompt"] == "You are a subagent for unit tests." def test_subagent_to_dict_optional_tools_included_when_set() -> None: sub = _minimal_subagent(allowed_tools=["read_file", "write_file"]) d = _subagent_to_dict(sub) assert "tools" in d # _subagent_to_dict serializes allowed_tools as a list[str]; SubAgent TypedDict # widens the tools type to include BaseTool/Callable, hence the cast for mypy. tools_list: list[Any] = list(d["tools"]) assert tools_list == ["read_file", "write_file"] def test_subagent_to_dict_no_tools_key_when_empty() -> None: sub = _minimal_subagent() d = _subagent_to_dict(sub) assert "tools" not in d def test_subagent_to_dict_optional_model_included_when_set() -> None: sub = _minimal_subagent(model="openrouter:deepseek/deepseek-chat") d = _subagent_to_dict(sub) assert "model" in d assert d["model"] == "openrouter:deepseek/deepseek-chat" def test_subagent_to_dict_no_model_key_when_none() -> None: sub = _minimal_subagent() d = _subagent_to_dict(sub) assert "model" not in d def test_subagent_to_dict_permissions_included_when_set() -> None: sub = _minimal_subagent( permissions=[{"operations": ["read"], "paths": ["/**"], "mode": "allow"}] ) d = _subagent_to_dict(sub) assert "permissions" in d assert len(d["permissions"]) == 1 # permissions 안의 항목도 FilesystemPermission 인스턴스 assert isinstance(d["permissions"][0], FilesystemPermission) def test_subagent_to_dict_permissions_empty_not_included() -> None: sub = _minimal_subagent() d = _subagent_to_dict(sub) assert "permissions" not in d def test_subagent_to_dict_interrupt_on_included_when_set() -> None: sub = _minimal_subagent(interrupt_on={"write_file": {"allowed_decisions": ["approve"]}}) d = _subagent_to_dict(sub) assert "interrupt_on" in d def test_subagent_to_dict_no_interrupt_on_when_empty() -> None: sub = _minimal_subagent() d = _subagent_to_dict(sub) assert "interrupt_on" not in d # --------------------------------------------------------------------------- # _resolve_openrouter_api_key # --------------------------------------------------------------------------- def test_resolve_api_key_from_config() -> None: config = load_config(openrouter_api_key="sk-or-from-config") key = _resolve_openrouter_api_key(config) assert key == "sk-or-from-config" def test_resolve_api_key_from_mydeepagent_env(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.delenv("MYDEEPAGENT_OPENROUTER_API_KEY", raising=False) monkeypatch.delenv("OPENROUTER_API_KEY", raising=False) monkeypatch.setenv("MYDEEPAGENT_OPENROUTER_API_KEY", "sk-or-env-mydeepagent") config = load_config(openrouter_api_key=None) key = _resolve_openrouter_api_key(config) assert key == "sk-or-env-mydeepagent" def test_resolve_api_key_fallback_to_openrouter_env(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.delenv("MYDEEPAGENT_OPENROUTER_API_KEY", raising=False) monkeypatch.delenv("OPENROUTER_API_KEY", raising=False) monkeypatch.setenv("OPENROUTER_API_KEY", "sk-or-env-fallback") config = load_config(openrouter_api_key=None) key = _resolve_openrouter_api_key(config) assert key == "sk-or-env-fallback" def test_resolve_api_key_raises_when_missing(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.delenv("MYDEEPAGENT_OPENROUTER_API_KEY", raising=False) monkeypatch.delenv("OPENROUTER_API_KEY", raising=False) config = load_config(openrouter_api_key=None) with pytest.raises(MyDeepAgentError) as exc_info: _resolve_openrouter_api_key(config) assert exc_info.value.code == "backend_auth_failed" def test_resolve_api_key_config_takes_priority_over_env(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setenv("MYDEEPAGENT_OPENROUTER_API_KEY", "sk-or-env") config = load_config(openrouter_api_key="sk-or-config-wins") key = _resolve_openrouter_api_key(config) assert key == "sk-or-config-wins" # --------------------------------------------------------------------------- # resolve_model_instance # --------------------------------------------------------------------------- def test_resolve_model_openrouter_returns_chat_openai() -> None: config = load_config(openrouter_api_key="sk-or-test") persona = _minimal_persona(model="openrouter:anthropic/claude-sonnet-4-6") instance = resolve_model_instance(persona, config) assert isinstance(instance, ChatOpenAI) assert instance.openai_api_base == config.openrouter_base_url def test_resolve_model_openrouter_uses_model_params() -> None: config = load_config(openrouter_api_key="sk-or-test") persona = _minimal_persona( model="openrouter:anthropic/claude-sonnet-4-6", model_params={"max_tokens": 1024, "temperature": 0.5}, ) instance = resolve_model_instance(persona, config) assert isinstance(instance, ChatOpenAI) assert instance.max_tokens == 1024 def test_resolve_model_non_openrouter_returns_string() -> None: config = load_config() persona = _minimal_persona( backend="anthropic", model="anthropic:claude-3-5-sonnet-20241022", ) result = resolve_model_instance(persona, config) assert isinstance(result, str) assert result == "anthropic:claude-3-5-sonnet-20241022" def test_resolve_model_with_override_openrouter() -> None: config = load_config(openrouter_api_key="sk-or-test") persona = _minimal_persona(model="openrouter:anthropic/claude-sonnet-4-6") instance = resolve_model_instance( persona, config, model_override="openrouter:deepseek/deepseek-chat" ) assert isinstance(instance, ChatOpenAI) assert "deepseek-chat" in instance.model_name # --------------------------------------------------------------------------- # build_backend — 5 케이스 # --------------------------------------------------------------------------- def test_build_backend_local_shell(tmp_path: Path) -> None: persona = _minimal_persona(deepagents_backend="local_shell") result = build_backend(persona, tmp_path) assert isinstance(result, LocalShellBackend) def test_build_backend_filesystem(tmp_path: Path) -> None: persona = _minimal_persona(deepagents_backend="filesystem") result = build_backend(persona, tmp_path) assert isinstance(result, FilesystemBackend) def test_build_backend_state_returns_none(tmp_path: Path) -> None: persona = _minimal_persona(deepagents_backend="state") result = build_backend(persona, tmp_path) assert result is None def test_build_backend_composite(tmp_path: Path) -> None: persona = _minimal_persona(deepagents_backend="composite") result = build_backend(persona, tmp_path) assert isinstance(result, CompositeBackend) def test_build_backend_langsmith_raises_config_invalid(tmp_path: Path) -> None: persona = _minimal_persona(deepagents_backend="langsmith") with pytest.raises(MyDeepAgentError) as exc_info: build_backend(persona, tmp_path) assert exc_info.value.code == "config_invalid" # --------------------------------------------------------------------------- # build_agent # --------------------------------------------------------------------------- def test_build_agent_returns_compiled_state_graph(tmp_path: Path) -> None: """build_agent should construct a CompiledStateGraph without calling the LLM API.""" config = load_config(openrouter_api_key="sk-or-test") persona = _minimal_persona(deepagents_backend="state") graph = build_agent(persona, config, root_dir=tmp_path) assert isinstance(graph, CompiledStateGraph) assert hasattr(graph, "invoke") assert hasattr(graph, "ainvoke") def test_build_agent_with_middleware_list(tmp_path: Path) -> None: """Extra middleware is accepted without error. build_agent automatically prepends SafetyShellMiddleware. Callers should pass *other* middleware here; passing a second SafetyShellMiddleware would hit deepagents' duplicate-name guard. """ from my_deepagent.middleware.audit import AuditToolMiddleware config = load_config(openrouter_api_key="sk-or-test") persona = _minimal_persona(deepagents_backend="state") graph = build_agent( persona, config, root_dir=tmp_path, middleware=[AuditToolMiddleware()], ) assert isinstance(graph, CompiledStateGraph) def test_build_agent_filesystem_backend(tmp_path: Path) -> None: """build_agent works with filesystem backend.""" config = load_config(openrouter_api_key="sk-or-test") persona = _minimal_persona(deepagents_backend="filesystem") graph = build_agent(persona, config, root_dir=tmp_path) assert isinstance(graph, CompiledStateGraph) def test_build_agent_with_persona_permissions(tmp_path: Path) -> None: """build_agent merges persona permissions with default safety permissions.""" config = load_config(openrouter_api_key="sk-or-test") persona = _minimal_persona( deepagents_backend="state", permissions=[{"operations": ["read"], "paths": ["/workspace/**"], "mode": "allow"}], ) graph = build_agent(persona, config, root_dir=tmp_path) assert isinstance(graph, CompiledStateGraph)