beaver_project/app-instance/backend/tests/unit/test_config_loader.py

import json

from beaver.engine import AgentLoop, EngineLoader
from beaver.engine.providers import make_provider_bundle
from beaver.engine.providers.litellm import LiteLLMProvider
from beaver.foundation.config import load_config


def test_load_config_reads_current_instance_shape(tmp_path) -> None:
    config_path = tmp_path / "config.json"
    config_path.write_text(
        json.dumps(
            {
                "agents": {
                    "defaults": {
                        "workspace": str(tmp_path / "workspace"),
                        "model": "qwen-plus",
                    }
                },
                "providers": {
                    "openai": {
                        "apiKey": "sk-test",
                        "apiBase": "https://oai.example.com/v1",
                        "extraHeaders": {"X-Test": "1"},
                    }
                },
                "embeddingModel": "text-embedding-v4",
            }
        ),
        encoding="utf-8",
    )

    config = load_config(config_path=config_path)
    target = config.resolve_provider_target()

    assert config.default_model == "qwen-plus"
    assert config.default_embedding_model == "text-embedding-v4"
    assert target["provider_name"] == "openai"
    assert target["model"] == "qwen-plus"
    assert target["api_key"] == "sk-test"
    assert target["api_base"] == "https://oai.example.com/v1"
    assert target["extra_headers"] == {"X-Test": "1"}


def test_provider_resolution_ignores_custom_and_disabled_overrides(tmp_path) -> None:
    config_path = tmp_path / "config.json"
    config_path.write_text(
        json.dumps(
            {
                "agents": {
                    "defaults": {
                        "workspace": str(tmp_path / "workspace"),
                        "model": "qwen-plus",
                        "provider": "custom",
                    }
                },
                "providers": {
                    "custom": {},
                    "openai": {
                        "apiKey": "sk-test",
                        "apiBase": "https://oai.example.com/v1",
                    },
                },
            }
        ),
        encoding="utf-8",
    )

    config = load_config(config_path=config_path)

    assert config.resolve_provider_target()["provider_name"] == "openai"
    assert config.resolve_provider_target(provider_name="custom")["provider_name"] == "openai"
    assert config.resolve_provider_target(provider_name="deepseek")["provider_name"] == "openai"


def test_engine_loader_uses_config_workspace(tmp_path) -> None:
    workspace = tmp_path / "workspace"
    config_path = tmp_path / "config.json"
    config_path.write_text(
        json.dumps(
            {
                "agents": {
                    "defaults": {
                        "workspace": str(workspace),
                        "model": "qwen-plus",
                    }
                },
                "providers": {"openai": {"apiKey": "sk-test", "apiBase": "https://oai.example.com/v1"}},
            }
        ),
        encoding="utf-8",
    )

    loader = EngineLoader(config_path=config_path)
    assert loader.workspace == workspace


def test_agent_loop_config_drives_provider_bundle(tmp_path) -> None:
    workspace = tmp_path / "workspace"
    config_path = tmp_path / "config.json"
    config_path.write_text(
        json.dumps(
            {
                "agents": {
                    "defaults": {
                        "workspace": str(workspace),
                        "model": "qwen-plus",
                    }
                },
                "providers": {"openai": {"apiKey": "sk-test", "apiBase": "https://oai.example.com/v1"}},
            }
        ),
        encoding="utf-8",
    )

    loop = AgentLoop(loader=EngineLoader(config_path=config_path))
    loaded = loop.boot()
    target = loaded.config.resolve_provider_target()

    assert target["provider_name"] == "openai"
    assert target["model"] == "qwen-plus"
    assert target["api_key"] == "sk-test"
    assert target["api_base"] == "https://oai.example.com/v1"
    loop.close()


def test_openai_compatible_qwen_config_keeps_openai_provider() -> None:
    bundle = make_provider_bundle(
        model="qwen-plus",
        provider_name="openai",
        api_key="sk-test",
        api_base="https://oai.example.com/v1",
    )

    assert bundle.main_runtime.provider_name == "openai"
    assert bundle.main_runtime.api_base == "https://oai.example.com/v1"
    assert isinstance(bundle.main_provider, LiteLLMProvider)
    assert bundle.main_provider._resolve_model("qwen-plus") == "openai/qwen-plus"


def test_load_config_reads_mcp_authz_identity(tmp_path) -> None:
    config_path = tmp_path / "beaver-home" / "config.json"
    config_path.parent.mkdir()
    config_path.write_text(
        json.dumps(
            {
                "tools": {
                    "mcpServers": {
                        "outlook_mcp": {
                            "url": "http://10.6.80.29:8000/mcp",
                            "authMode": "oauth_backend_token",
                            "authAudience": "mcp:outlook_mcp",
                            "authScopes": ["list_tools", "tool:mail_list_messages"],
                            "toolTimeout": 60,
                            "sensitive": True,
                        }
                    }
                },
                "authz": {
                    "enabled": True,
                    "baseUrl": "http://nano-authz-service:19090",
                },
                "backend_identity": {
                    "backend_id": "stevenli",
                    "client_id": "stevenli",
                },
            }
        ),
        encoding="utf-8",
    )
    config = load_config(config_path=config_path)

    server = config.tools.mcp_servers["outlook_mcp"]
    assert server.transport == "http"
    assert server.url == "http://10.6.80.29:8000/mcp"
    assert server.auth_mode == "oauth_backend_token"
    assert server.auth_audience == "mcp:outlook_mcp"
    assert "tool:mail_list_messages" in server.auth_scopes
    assert server.tool_timeout == 60
    assert server.sensitive is True

    assert config.authz.enabled is True
    assert config.authz.base_url == "http://nano-authz-service:19090"
    assert config.backend_identity.backend_id == "stevenli"
    assert config.backend_identity.client_id == "stevenli"


def test_load_config_adds_managed_local_mcp_servers(tmp_path) -> None:
    config_path = tmp_path / "config.json"
    config_path.write_text(
        json.dumps({"tools": {"mcpServers": {}}}),
        encoding="utf-8",
    )

    config = load_config(config_path=config_path)

    local = config.tools.mcp_servers["local_filesystem_mcp"]
    assert local.transport == "stdio"
    assert local.kind == "local"
    assert local.category == "filesystem"
    assert local.managed is True
    assert "beaver.interfaces.mcp.tools_server" in local.args