"""Integration coverage for persisted, administrator-owned performance controls."""

from types import SimpleNamespace
from unittest.mock import AsyncMock

import httpx
import pytest
import pytest_asyncio
from fastapi import FastAPI, HTTPException
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from test_api_read_cache import FakeRedis, _config

from app.gateway import api_read_cache as cache
from app.gateway.deps import get_config
from app.gateway.routers import browser_context, system_settings
from deerflow.config.browser_context_config import BrowserContextConfig
from deerflow.persistence.api_read_cache_settings.model import ApiReadCacheSettingRow


@pytest_asyncio.fixture
async def controls(monkeypatch, tmp_path):
    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'controls.db'}")
    async with engine.begin() as connection:
        await connection.run_sync(ApiReadCacheSettingRow.__table__.create)
    factory = async_sessionmaker(engine, expire_on_commit=False)
    monkeypatch.setattr("deerflow.persistence.engine.get_session_factory", lambda: factory)
    monkeypatch.setattr(cache, "_api_read_cache_settings_cache", None)
    redis = FakeRedis()
    monkeypatch.setattr(cache, "_client", lambda *_args, **_kwargs: (redis, "test:controls"))
    config = _config()
    config.browser_context = BrowserContextConfig(enabled=True)
    auth = AsyncMock(return_value=SimpleNamespace(id="admin", system_role="admin"))
    monkeypatch.setattr(system_settings, "get_current_user_from_request", auth)
    app = FastAPI()
    app.state.config = config
    app.dependency_overrides[get_config] = lambda: config
    app.include_router(system_settings.router)
    app.include_router(browser_context.router)
    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
        yield client, config, redis, auth
    await engine.dispose()


@pytest.mark.asyncio
async def test_partial_updates_persist_without_resetting_other_switches(controls, monkeypatch):
    client, config, _, _ = controls
    endpoint = "/api/system-settings/api-read-cache"
    response = await client.put(endpoint, json={"enabled": True, "ttl_seconds": 777, "collection_overrides": {"agents": False}})
    assert response.status_code == 200
    assert {row["id"]: row["enabled"] for row in response.json()["collections"]}["agents"] is False
    response = await client.put(endpoint, json={"browser_status_enabled": False, "collection_overrides": {"skills": False}})
    assert response.status_code == 200
    monkeypatch.setattr(cache, "_api_read_cache_settings_cache", None)
    result = (await client.get(endpoint)).json()
    assert result["enabled"] is True and result["ttl_seconds"] == 777
    assert result["browser_status_enabled"] is False
    rows = {row["id"]: row for row in result["collections"]}
    assert not rows["agents"]["enabled"] and not rows["skills"]["enabled"]
    assert rows["light_apps"]["enabled"]
    assert all(row["paths"] and row["scope"] and row["ttl_seconds"] == 777 for row in rows.values())
    assert await cache.is_browser_status_enabled(config) is False


@pytest.mark.asyncio
async def test_per_collection_disable_bypasses_reads_and_writes(controls):
    client, config, redis, _ = controls
    await cache.set_cached_json(config, "agents", "u:1", {"old": True})
    await client.put("/api/system-settings/api-read-cache", json={"collection_overrides": {"agents": False}})
    snapshot = dict(redis.values)
    assert await cache.get_cached_json(config, "agents", "u:1", version="0") is None
    assert await cache.get_cache_version(config, "agents") is None
    assert await cache.wait_for_cached_json_or_claim_fill(config, "agents", "u:1", version="0") == (None, None)
    await cache.set_cached_json(config, "agents", "u:1", {"new": True}, expected_version="0")
    assert redis.values == snapshot
    await cache.set_cached_json(config, "skills", "u:1", {"enabled": True})
    assert await cache.get_cached_json(config, "skills", "u:1") == {"enabled": True}
    await client.put("/api/system-settings/api-read-cache", json={"collection_overrides": {"agents": True}})
    assert await cache.get_cached_json(config, "agents", "u:1") is None


@pytest.mark.asyncio
async def test_browser_access_gate_precedes_warm_cache_and_recovers(controls):
    client, _, redis, _ = controls
    assert (await client.get("/api/browser-context/status")).status_code == 200
    assert any(":browser_status:v" in key for key in redis.values)
    response = await client.put("/api/system-settings/api-read-cache", json={"browser_status_enabled": False})
    assert response.status_code == 200
    response = await client.get("/api/browser-context/status")
    assert response.status_code == 403
    assert response.json()["detail"]["code"] == "browser_status_disabled"
    assert response.headers["Retry-After"] == "300"
    await client.put("/api/system-settings/api-read-cache", json={"browser_status_enabled": True})
    assert (await client.get("/api/browser-context/status")).json()["enabled"] is True


@pytest.mark.asyncio
@pytest.mark.parametrize("payload", [{"ttl_seconds": 0}, {"ttl_seconds": 3601}, {"collection_overrides": {"typo": True}}])
async def test_invalid_settings_are_rejected_without_persisting(controls, payload):
    client, _, _, _ = controls
    assert (await client.put("/api/system-settings/api-read-cache", json=payload)).status_code == 422
    assert (await client.get("/api/system-settings/api-read-cache")).json()["ttl_seconds"] == 300


@pytest.mark.asyncio
@pytest.mark.parametrize("role,status", [("user", 403), ("anonymous", 401)])
async def test_only_admin_can_read_or_update_controls(controls, role, status):
    client, _, _, auth = controls
    if role == "anonymous":
        auth.side_effect = HTTPException(status_code=401, detail="Login required")
    else:
        auth.return_value = SimpleNamespace(system_role=role)
    assert (await client.get("/api/system-settings/api-read-cache")).status_code == status
    assert (await client.put("/api/system-settings/api-read-cache", json={"browser_status_enabled": False})).status_code == status


@pytest.mark.asyncio
async def test_redis_failure_falls_back_and_global_off_preserves_individual_choices(controls, monkeypatch):
    client, config, redis, _ = controls
    await client.put("/api/system-settings/api-read-cache", json={"enabled": False, "collection_overrides": {"agents": False}})
    assert await cache.get_cache_version(config, "skills") is None
    response = await client.put("/api/system-settings/api-read-cache", json={"enabled": True})
    assert next(row for row in response.json()["collections"] if row["id"] == "agents")["enabled"] is False
    monkeypatch.setattr(redis, "get", AsyncMock(side_effect=ConnectionError("offline")))
    assert await cache.get_cached_json(config, "skills", "u:1") is None
    await cache.set_cached_json(config, "skills", "u:1", {"ok": True})
    assert (await client.get("/api/browser-context/status")).status_code == 200