dev-puppeteer/my-deepagent/tests/integration/test_openrouter_smoke.py

"""Real OpenRouter API smoke test. Costs ~$0.001-$0.003 per full run.

Skipped automatically when no API key is configured.
Uses deepseek/deepseek-chat (cheapest available) with max_tokens=50.
"""

from __future__ import annotations

import os
from pathlib import Path

import pytest

from my_deepagent.config import load_config
from my_deepagent.persona import Persona
from my_deepagent.session import resolve_model_instance

_HAS_KEY = (
    bool(os.environ.get("MYDEEPAGENT_OPENROUTER_API_KEY") or os.environ.get("OPENROUTER_API_KEY"))
    or Path(".env").is_file()
)

pytestmark = [
    pytest.mark.integration,
    pytest.mark.skipif(not _HAS_KEY, reason="no OpenRouter API key configured"),
]


def _smoke_persona() -> Persona:
    return Persona.model_validate(
        {
            "name": "smoke-test",
            "version": 1,
            "backend": "openrouter",
            "model": "openrouter:deepseek/deepseek-chat",
            "provider_origin": "China/DeepSeek",
            "capabilities": ["evidence_check"],
            "max_risk_level": "low",
            "system_prompt": (
                "You are a smoke-test echo bot. Reply only with the literal token 'OK'."
            ),
            "model_params": {"max_tokens": 50, "temperature": 0.0},
            # deepagents 0.6.x: local_shell backend + permissions 동시 사용 시
            # NotImplementedError 발생. state 백엔드는 permissions 제약 없음.
            "deepagents_backend": "state",
        }
    )


def _smoke_persona_local_shell() -> Persona:
    return Persona.model_validate(
        {
            "name": "smoke-test-local-shell",
            "version": 1,
            "backend": "openrouter",
            "model": "openrouter:deepseek/deepseek-chat",
            "provider_origin": "China/DeepSeek",
            "capabilities": ["evidence_check"],
            "max_risk_level": "low",
            "system_prompt": (
                "You are a smoke-test echo bot. Reply only with the literal token 'OK'."
            ),
            "model_params": {"max_tokens": 50, "temperature": 0.0},
            # local_shell backend: SafetyShellMiddleware enforces path + destructive-command
            # policy; permissions kwarg is skipped to avoid deepagents 0.6.1 NotImplementedError.
            "deepagents_backend": "local_shell",
        }
    )


def test_openrouter_chat_completion_returns_response() -> None:
    """ChatOpenAI 인스턴스로 1회 호출하여 OpenRouter base_url + auth + 응답 흐름 검증."""
    config = load_config()
    persona = _smoke_persona()
    chat = resolve_model_instance(persona, config)
    response = chat.invoke(
        [
            ("system", persona.system_prompt),
            ("user", "Reply with the exact string 'OK' and nothing else."),
        ]
    )
    assert response is not None
    content = response.content
    # langchain BaseMessage.content는 str | list[content_block_dict]
    if isinstance(content, str):
        assert len(content) > 0
    else:
        assert len(content) > 0


def test_openrouter_usage_metadata_present() -> None:
    """response.usage_metadata가 input_tokens/output_tokens를 채워야 cost 계측 가능."""
    config = load_config()
    persona = _smoke_persona()
    chat = resolve_model_instance(persona, config)
    response = chat.invoke(
        [
            ("system", persona.system_prompt),
            ("user", "Reply with 'OK'."),
        ]
    )
    usage = getattr(response, "usage_metadata", None)
    assert usage is not None, "OpenRouter response must include usage_metadata"
    assert usage.get("input_tokens", 0) > 0
    assert usage.get("output_tokens", 0) > 0


def test_openrouter_deepagents_create_smoke() -> None:
    """deepagents create_deep_agent + 실 OpenRouter 호출 1회. 가장 비싼 검증."""
    config = load_config()
    persona = _smoke_persona()
    from my_deepagent.session import build_agent

    agent = build_agent(persona, config, root_dir=Path.cwd())
    result = agent.invoke({"messages": [{"role": "user", "content": "Reply with 'OK' only."}]})
    messages = result.get("messages", [])
    assert len(messages) > 0
    last = messages[-1]
    content = getattr(last, "content", "")
    if isinstance(content, list):
        content = " ".join(str(c) for c in content)
    assert len(str(content)) > 0


def test_openrouter_deepagents_local_shell_smoke(tmp_path: Path) -> None:
    """Real OpenRouter call via deepagents + LocalShellBackend + SafetyShellMiddleware.

    Verifies deepagents 0.6.1 workaround: local_shell backend with permissions kwarg
    skipped, SafetyShellMiddleware automatically injected by build_agent.
    """
    config = load_config()
    persona = _smoke_persona_local_shell()
    from my_deepagent.session import build_agent

    agent = build_agent(persona, config, root_dir=tmp_path)
    result = agent.invoke({"messages": [{"role": "user", "content": "Reply 'OK' only."}]})
    messages = result.get("messages", [])
    assert len(messages) > 0
    last = messages[-1]
    content = getattr(last, "content", "")
    if isinstance(content, list):
        content = " ".join(str(c) for c in content)
    assert len(str(content)) > 0