"""Tests for per-model concurrency limit settings and enforcement."""

from __future__ import annotations

import json
from pathlib import Path
from types import SimpleNamespace

import pytest

from app.gateway.routers import system_settings as ss_router
from app.gateway.services import (
    RUN_SCOPE_EXTERNAL_ROOT,
    RUN_SCOPE_ORCHESTRATION_CHILD,
    _get_model_concurrency_limit,
    _get_user_model_concurrency_limit,
    _resolve_run_scope,
    _resolve_effective_run_model_name,
    internal_run_scope_headers,
)
from deerflow.config import system_settings as ss
from deerflow.config.concurrency_config import RedisConcurrencyConfig
from deerflow.runtime.runs.concurrency_gate import RedisConcurrencyGate
from deerflow.runtime.runs.manager import ModelConcurrencyLimitError, RunManager


@pytest.fixture
def tmp_runtime_home(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> Path:
    def fake_home() -> Path:
        return tmp_path

    monkeypatch.setattr(ss, "runtime_home", fake_home)
    return tmp_path


def _patch_admin(monkeypatch: pytest.MonkeyPatch, user: object | None = None) -> None:
    admin = user or SimpleNamespace(
        id="admin1",
        system_role="admin",
        email="admin@example.com",
    )

    async def _optional(_request):
        return admin

    monkeypatch.setattr(ss_router, "get_optional_user_from_request", _optional)


@pytest.mark.asyncio
async def test_model_concurrency_settings_round_trip(
    tmp_runtime_home: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    _patch_admin(monkeypatch)
    req = SimpleNamespace()

    saved = await ss_router.update_model_concurrency(
        ss_router.ModelConcurrencyUpdateRequest(
            enabled=True,
            default_limit=8,
            model_limits={"glm5": 10, "deepseek": 4},
            user_default_limit=2,
            user_model_limits={"glm5": 1},
        ),
        req,
    )
    assert saved.enabled is True
    assert saved.default_limit == 8
    assert saved.model_limits == {"glm5": 10, "deepseek": 4}
    assert saved.user_default_limit == 2
    assert saved.user_model_limits == {"glm5": 1}

    file_path = tmp_runtime_home / "system_settings.json"
    payload = json.loads(file_path.read_text(encoding="utf-8"))
    assert payload["model_concurrency"]["enabled"] is True
    assert payload["model_concurrency"]["default_limit"] == 8
    assert payload["model_concurrency"]["model_limits"]["glm5"] == 10
    assert payload["model_concurrency"]["user_default_limit"] == 2
    assert payload["model_concurrency"]["user_model_limits"]["glm5"] == 1

    loaded = await ss_router.get_model_concurrency(req)
    assert loaded.enabled is True
    assert loaded.default_limit == 8
    assert loaded.model_limits["deepseek"] == 4
    assert loaded.user_default_limit == 2
    assert loaded.user_model_limits["glm5"] == 1


def test_resolve_effective_run_model_name_prefers_requested_model(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    app_config = SimpleNamespace(
        models=[SimpleNamespace(name="default-model")],
        get_model_config=lambda name: {"name": name} if name in {"default-model", "glm5", "agent-model"} else None,
    )
    monkeypatch.setattr("app.gateway.services.get_app_config", lambda: app_config)
    monkeypatch.setattr(
        "app.gateway.services.load_agent_config",
        lambda _agent_id: SimpleNamespace(model="agent-model"),
    )

    body = SimpleNamespace(
        assistant_id="custom-agent",
        context={"model_name": "glm5"},
        config=None,
    )
    assert _resolve_effective_run_model_name(body) == "glm5"


def test_resolve_effective_run_model_name_uses_agent_default_when_request_missing(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    app_config = SimpleNamespace(
        models=[SimpleNamespace(name="default-model")],
        get_model_config=lambda name: {"name": name} if name in {"default-model", "agent-model"} else None,
    )
    monkeypatch.setattr("app.gateway.services.get_app_config", lambda: app_config)
    monkeypatch.setattr(
        "app.gateway.services.load_agent_config",
        lambda _agent_id: SimpleNamespace(model="agent-model"),
    )

    body = SimpleNamespace(
        assistant_id="custom-agent",
        context={},
        config=None,
    )
    assert _resolve_effective_run_model_name(body) == "agent-model"


def test_resolve_effective_run_model_name_uses_context_agent_id(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    app_config = SimpleNamespace(
        models=[SimpleNamespace(name="default-model")],
        get_model_config=lambda name: {"name": name} if name in {"default-model", "agent-model"} else None,
    )
    monkeypatch.setattr("app.gateway.services.get_app_config", lambda: app_config)
    monkeypatch.setattr(
        "app.gateway.services.load_agent_config",
        lambda agent_id: SimpleNamespace(model="agent-model") if agent_id == "custom-agent" else SimpleNamespace(model="default-model"),
    )

    body = SimpleNamespace(
        assistant_id="lead_agent",
        context={"agent_id": "custom-agent"},
        config=None,
    )
    assert _resolve_effective_run_model_name(body) == "agent-model"


def test_resolve_effective_run_model_name_context_agent_overrides_assistant_id(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    app_config = SimpleNamespace(
        models=[SimpleNamespace(name="default-model")],
        get_model_config=lambda name: {"name": name} if name in {"default-model", "context-model", "assistant-model"} else None,
    )
    monkeypatch.setattr("app.gateway.services.get_app_config", lambda: app_config)
    monkeypatch.setattr(
        "app.gateway.services.load_agent_config",
        lambda agent_id: SimpleNamespace(model="context-model" if agent_id == "context-agent" else "assistant-model"),
    )

    body = SimpleNamespace(
        assistant_id="assistant-agent",
        context={"agent_id": "context-agent"},
        config=None,
    )
    assert _resolve_effective_run_model_name(body) == "context-model"


@pytest.mark.asyncio
async def test_run_manager_rejects_model_limit_atomically(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(
        "app.gateway.services.load_system_settings",
        lambda: SimpleNamespace(
            model_concurrency=SimpleNamespace(
                enabled=True,
                default_limit=5,
                model_limits={"glm5": 2},
            )
        ),
    )
    assert _get_model_concurrency_limit("glm5") == 2

    run_mgr = RunManager()
    await run_mgr.create_or_reject("t1", metadata={"model_name": "glm5"})
    await run_mgr.create_or_reject("t2", metadata={"model_name": "glm5"})

    with pytest.raises(ModelConcurrencyLimitError):
        await run_mgr.create_or_reject(
            "t3",
            metadata={"model_name": "glm5"},
            model_concurrency_limit=("glm5", 2),
        )


@pytest.mark.asyncio
async def test_run_manager_rejects_user_model_limit_atomically(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(
        "app.gateway.services.load_system_settings",
        lambda: SimpleNamespace(
            model_concurrency=SimpleNamespace(
                enabled=True,
                default_limit=10,
                model_limits={},
                user_default_limit=3,
                user_model_limits={"glm5": 1},
            )
        ),
    )
    assert _get_user_model_concurrency_limit("glm5") == 1
    assert _get_user_model_concurrency_limit("other") == 3

    run_mgr = RunManager()
    await run_mgr.create_or_reject(
        "t-child",
        metadata={
            "model_name": "glm5",
            "user_id": "u1",
            "run_scope": RUN_SCOPE_ORCHESTRATION_CHILD,
        },
        user_model_concurrency_limit=("glm5", "u1", 1),
    )
    await run_mgr.create_or_reject("t1", metadata={"model_name": "glm5", "user_id": "u1"})
    await run_mgr.create_or_reject(
        "t2",
        metadata={"model_name": "glm5", "user_id": "u2"},
        user_model_concurrency_limit=("glm5", "u2", 1),
    )

    with pytest.raises(ModelConcurrencyLimitError) as exc_info:
        await run_mgr.create_or_reject(
            "t3",
            metadata={"model_name": "glm5", "user_id": "u1"},
            user_model_concurrency_limit=("glm5", "u1", 1),
        )

    assert exc_info.value.scope == "user_model"
    assert exc_info.value.user_id == "u1"


def test_get_model_concurrency_limit_skips_when_disabled(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(
        "app.gateway.services.load_system_settings",
        lambda: SimpleNamespace(
            model_concurrency=SimpleNamespace(
                enabled=False,
                default_limit=1,
                model_limits={"glm5": 1},
                user_default_limit=1,
                user_model_limits={"glm5": 1},
            )
        ),
    )
    assert _get_model_concurrency_limit("glm5") is None
    assert _get_user_model_concurrency_limit("glm5") is None


def test_internal_run_scope_is_only_trusted_for_loopback_or_server_argument() -> None:
    body = SimpleNamespace(metadata={"run_scope": RUN_SCOPE_ORCHESTRATION_CHILD})

    trusted_loopback_request = SimpleNamespace(
        client=SimpleNamespace(host="127.0.0.1"),
        headers=internal_run_scope_headers(),
    )
    untrusted_loopback_request = SimpleNamespace(
        client=SimpleNamespace(host="127.0.0.1"),
        headers={},
    )
    public_request = SimpleNamespace(
        client=SimpleNamespace(host="10.0.0.8"),
        headers=internal_run_scope_headers(),
    )

    assert (
        _resolve_run_scope(body, trusted_loopback_request, None)
        == RUN_SCOPE_ORCHESTRATION_CHILD
    )
    assert (
        _resolve_run_scope(body, untrusted_loopback_request, None)
        == RUN_SCOPE_EXTERNAL_ROOT
    )
    assert _resolve_run_scope(body, public_request, None) == RUN_SCOPE_EXTERNAL_ROOT
    assert (
        _resolve_run_scope(
            SimpleNamespace(metadata={}),
            public_request,
            RUN_SCOPE_ORCHESTRATION_CHILD,
        )
        == RUN_SCOPE_ORCHESTRATION_CHILD
    )


def test_redis_concurrency_namespace_has_system_prefix() -> None:
    assert RedisConcurrencyConfig().namespace == "cmzs_deerflow:concurrency"
    gate = RedisConcurrencyGate(RedisConcurrencyConfig(namespace=""))
    assert gate._ns() == "cmzs_deerflow:concurrency"