dev-puppeteer/my-deepagent/tests/unit/test_session.py

"""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)