from __future__ import annotations

import asyncio
import contextlib
import json
import sys
import types
from pathlib import Path

import httpx
from imagefind import api as api_module
from imagefind.config import Settings
from imagefind.database import DatabaseTransientError, utcnow
from imagefind.main import create_app


def make_app(tmp_path: Path):
    settings = Settings(data_dir=tmp_path / "data", embedding_backend="hash", upload_reserve_gb=0)
    settings.prepare()
    app = create_app(settings)
    app.state.services.auth.setup("original administrator password")
    _, token = app.state.services.auth.create_api_token("test")
    return app, {"Authorization": f"Bearer {token}"}


def seed_videos(app) -> None:
    now = utcnow()
    with app.state.services.db.transaction() as conn:
        conn.execute(
            "INSERT INTO sources(id,kind,name,config_json,created_at,updated_at) VALUES(?,?,?,?,?,?)",
            ("source", "local", "Local", json.dumps({"path": "/media"}), now, now),
        )
        for index in range(3):
            conn.execute(
                "INSERT INTO videos(id,source_id,source_key,display_name,location,size_bytes,fingerprint,"
                "duration_ms,status,available,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,'indexed',1,?,?)",
                (
                    f"video-{index}",
                    "source",
                    f"video-{index}.mp4",
                    f"Video {index}",
                    f"/media/video-{index}.mp4",
                    100 + index,
                    f"fingerprint-{index}",
                    (index + 1) * 10_000,
                    now,
                    now,
                ),
            )
        conn.execute(
            "INSERT INTO video_metadata(video_id,series,updated_at) VALUES('video-0','Series A',?),"
            "('video-1','series a',?)",
            (now, now),
        )
        conn.execute(
            "INSERT INTO video_state(video_id,liked,favorited,progress_ms,completed,last_played_at,updated_at) "
            "VALUES('video-0',1,0,3000,0,?,?),('video-1',0,1,10000,1,?,?)",
            (now, now, now, now),
        )


def test_video_list_batches_tag_items_in_one_read_connection(tmp_path: Path):
    app, headers = make_app(tmp_path)
    seed_videos(app)
    service = app.state.services
    now = utcnow()
    with service.db.transaction() as conn:
        conn.execute(
            "INSERT INTO tag_groups(id,name,selection_mode,sort_order,created_at,updated_at) "
            "VALUES('batch-group','类型','multi',1,?,?)",
            (now, now),
        )
        conn.execute(
            "INSERT INTO tags(id,group_id,name,created_at,updated_at) "
            "VALUES('batch-tag','batch-group','测试标签',?,?)",
            (now, now),
        )
        conn.executemany(
            "INSERT INTO video_tags(video_id,tag_id) VALUES(?,'batch-tag')",
            [(f"video-{index}",) for index in range(3)],
        )

    original_read = service.db.read
    read_calls = []

    @contextlib.contextmanager
    def counted_read():
        read_calls.append(1)
        with original_read() as connection:
            yield connection

    service.db.read = counted_read

    async def scenario() -> None:
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            response = await client.get("/api/v1/videos?limit=500", headers=headers)
            assert response.status_code == 200
            assert len(response.json()) == 3
            assert all(item["tag_items"][0]["name"] == "测试标签" for item in response.json())

    asyncio.run(scenario())
    # The freshly issued API token is served by the bounded authentication
    # cache, while one WAL read batches the complete video list/tag payload.
    assert len(read_calls) == 1


def test_home_feed_groups_recent_collections_categories_and_unorganized(tmp_path: Path):
    app, headers = make_app(tmp_path)
    seed_videos(app)
    now = utcnow()
    with app.state.services.db.transaction() as conn:
        conn.execute(
            "INSERT INTO tag_groups(id,name,selection_mode,sort_order,created_at,updated_at) "
            "VALUES('home-group','类型','multi',1,?,?)",
            (now, now),
        )
        conn.execute(
            "INSERT INTO tags(id,group_id,name,created_at,updated_at) VALUES"
            "('home-tag-a','home-group','剧情',?,?),('home-tag-b','home-group','纪录',?,?),"
            "('home-tag-small','home-group','单片',?,?)",
            (now, now, now, now, now, now),
        )
        conn.executemany(
            "INSERT INTO video_tags(video_id,tag_id) VALUES(?,?)",
            [
                ("video-0", "home-tag-a"),
                ("video-1", "home-tag-a"),
                ("video-0", "home-tag-b"),
                ("video-1", "home-tag-b"),
                ("video-0", "home-tag-small"),
            ],
        )

    async def scenario() -> None:
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            assert (await client.get("/api/v1/home")).status_code == 401
            collection = await client.post(
                "/api/v1/collections", headers=headers, json={"name": "首页合集", "video_ids": ["video-0"]}
            )
            assert collection.status_code == 201
            response = await client.get("/api/v1/home?item_limit=4&tag_limit=1", headers=headers)
            assert response.status_code == 200
            feed = response.json()
            assert feed["video_count"] == 3
            assert [item["id"] for item in feed["recent"]["items"]] == ["video-0", "video-1", "video-2"]
            assert feed["collections"]["items"][0]["name"] == "首页合集"
            assert len(feed["categories"]) == 1
            assert feed["categories"][0]["title"] == "剧情"
            assert feed["categories"][0]["total"] == 2
            assert feed["unorganized"]["total"] == 1
            assert [item["id"] for item in feed["unorganized"]["items"]] == ["video-2"]

            with app.state.services.db.transaction() as conn:
                conn.execute("DELETE FROM video_tags")
                conn.execute("DELETE FROM video_metadata")
                conn.execute("DELETE FROM collection_items")
                conn.execute("DELETE FROM collection_videos")
                conn.execute("DELETE FROM collections")
            deduplicated = (await client.get("/api/v1/home", headers=headers)).json()
            assert deduplicated["categories"] == []
            assert deduplicated["collections"]["items"] == []
            assert deduplicated["unorganized"] is None

    asyncio.run(scenario())


def test_profile_validation_stats_filters_and_password_revocation(tmp_path: Path):
    app, headers = make_app(tmp_path)
    seed_videos(app)

    async def scenario() -> None:
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            profile = await client.get("/api/v1/profile", headers=headers)
            assert profile.status_code == 200
            assert profile.json()["nickname"] == "管理员"
            assert profile.json()["avatar"]["text"] == "管"
            assert profile.json()["avatar"]["accent"].startswith("#")
            assert profile.json()["counts"] == {
                "favorites": 1,
                "likes": 1,
                "history": 2,
                "continue_watching": 1,
            }
            invalid = await client.patch("/api/v1/profile", headers=headers, json={"nickname": "   "})
            assert invalid.status_code == 400
            changed = await client.patch("/api/v1/profile", headers=headers, json={"nickname": "NAS 管理员"})
            assert changed.json()["avatar"]["text"] == "NA"

            assert len((await client.get("/api/v1/videos?liked=true", headers=headers)).json()) == 1
            assert len((await client.get("/api/v1/videos?favorite=true", headers=headers)).json()) == 1
            assert len((await client.get("/api/v1/videos?continue_only=true", headers=headers)).json()) == 1
            history = (await client.get("/api/v1/videos?played_only=true&sort=last_played", headers=headers)).json()
            assert len(history) == 2

            first = await client.post(
                "/api/v1/auth/login", json={"password": "original administrator password"}
            )
            assert first.status_code == 200
            old_cookie = first.cookies.get("imagefind_session")
            wrong = await client.patch(
                "/api/v1/profile/password",
                headers=headers,
                json={"current_password": "wrong", "new_password": "replacement administrator password"},
            )
            assert wrong.status_code == 400
            changed_password = await client.patch(
                "/api/v1/profile/password",
                headers=headers,
                json={
                    "current_password": "original administrator password",
                    "new_password": "replacement administrator password",
                },
            )
            assert changed_password.status_code == 200
            assert changed_password.json()["reauthenticate"] is True
            client.cookies.set("imagefind_session", old_cookie)
            revoked = await client.get("/api/v1/auth/me")
            assert revoked.status_code == 401
            assert (
                await client.post("/api/v1/auth/login", json={"password": "replacement administrator password"})
            ).status_code == 200

    asyncio.run(scenario())


def test_bulk_video_state_actions_are_atomic_and_preserve_unrelated_state(tmp_path: Path):
    app, headers = make_app(tmp_path)
    seed_videos(app)

    async def scenario() -> None:
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            missing = await client.patch(
                "/api/v1/videos/state/bulk",
                headers=headers,
                json={"video_ids": ["video-0", "missing"], "action": "unlike"},
            )
            assert missing.status_code == 404
            assert len((await client.get("/api/v1/videos?liked=true", headers=headers)).json()) == 1

            assert (
                await client.patch(
                    "/api/v1/videos/state/bulk",
                    headers=headers,
                    json={"video_ids": ["video-0", "video-0"], "action": "unlike"},
                )
            ).json() == {"updated": 1, "action": "unlike"}
            await client.patch(
                "/api/v1/videos/state/bulk",
                headers=headers,
                json={"video_ids": ["video-1"], "action": "unfavorite"},
            )
            await client.patch(
                "/api/v1/videos/state/bulk",
                headers=headers,
                json={"video_ids": ["video-0", "video-1"], "action": "clear_history"},
            )
            assert (await client.get("/api/v1/videos?liked=true", headers=headers)).json() == []
            assert (await client.get("/api/v1/videos?favorite=true", headers=headers)).json() == []
            assert (await client.get("/api/v1/videos?played_only=true", headers=headers)).json() == []

    asyncio.run(scenario())


def test_jobs_support_stable_numbered_pagination_and_legacy_limit(tmp_path: Path):
    app, headers = make_app(tmp_path)
    diagnostic_job = ""
    for index in range(23):
        diagnostic_job = app.state.services.jobs.enqueue("scan_source", {"index": index})
    app.state.services.jobs.set_diagnostics(
        diagnostic_job,
        {
            "requested_device": "GPU.0",
            "actual_device": "CPU",
            "fallback_scope": "job",
            "fallback_reason": "low_quality_result",
            "private_value": "must not be exposed",
        },
    )

    async def scenario() -> None:
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            first = (await client.get("/api/v1/jobs?page=1&page_size=10", headers=headers)).json()
            last = (await client.get("/api/v1/jobs?page=3&page_size=10", headers=headers)).json()
            assert (first["page"], first["pages"], first["total"], len(first["items"])) == (1, 3, 23, 10)
            assert (last["page"], len(last["items"])) == (3, 3)
            assert not ({item["id"] for item in first["items"]} & {item["id"] for item in last["items"]})
            diagnosed = next(item for item in first["items"] if item["id"] == diagnostic_job)
            assert diagnosed["inference_diagnostics"] == {
                "requested_device": "GPU.0",
                "actual_device": "CPU",
                "fallback_scope": "job",
                "fallback_reason": "low_quality_result",
            }
            assert "diagnostics_json" not in diagnosed
            legacy = (await client.get("/api/v1/jobs?limit=4", headers=headers)).json()
            assert isinstance(legacy, list) and len(legacy) == 4

    asyncio.run(scenario())


def test_job_diagnostics_are_best_effort_on_transient_database_conflict(
    tmp_path: Path,
    monkeypatch,
    caplog,
):
    app, _headers = make_app(tmp_path)
    job_id = app.state.services.jobs.enqueue("scan_source", {"index": 1})

    def conflict(*_args, **_kwargs):
        raise DatabaseTransientError("temporary conflict")

    monkeypatch.setattr(app.state.services.db, "write_with_retry", conflict)
    app.state.services.jobs.set_diagnostics(job_id, {"actual_device": "GPU.0"})

    assert "skipped diagnostics update" in caplog.text


def test_model_status_exposes_audio_circuit_breaker_state(tmp_path: Path):
    app, headers = make_app(tmp_path)
    accelerator = app.state.services.accelerator
    accelerator.record_transient_failure("audio", "stall one", stage="inference_stall")
    accelerator.record_transient_failure("audio", "stall two", stage="inference_stall")
    accelerator.record_transient_failure("audio", "stall three", stage="inference_stall")

    async def scenario() -> None:
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            response = await client.get("/api/v1/models", headers=headers)
            assert response.status_code == 200
            audio = response.json()["accelerator"]["components"]["audio"]
            assert audio["circuit_state"] == "open"
            assert audio["fallback_scope"] == "component"
            assert audio["failure_count"] == 3
            assert audio["retry_at"]

    asyncio.run(scenario())


def test_upload_list_returns_last_snapshot_when_refresh_exceeds_budget(tmp_path: Path, monkeypatch):
    app, headers = make_app(tmp_path)
    cached = {"id": "cached-upload", "status": "receiving", "stage": "receiving"}
    app.state.services.uploads._list_cache = [cached]
    monkeypatch.setattr(api_module, "API_READ_DEADLINE_SECONDS", 0.01)

    async def delayed(function, /, *args, **kwargs):
        await asyncio.sleep(0.05)
        return function(*args, **kwargs)

    monkeypatch.setattr(api_module, "_background_api", delayed)

    async def scenario() -> None:
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            response = await client.get("/api/v1/uploads", headers=headers)
            assert response.status_code == 200
            assert response.json() == [cached]
            # Let the shielded refresh finish so the test loop closes cleanly.
            await asyncio.sleep(0.06)

    asyncio.run(scenario())


def test_upload_page_returns_matching_snapshot_when_refresh_exceeds_budget(tmp_path: Path, monkeypatch):
    app, headers = make_app(tmp_path)
    cached = {
        "items": [{"id": "cached-page-upload", "status": "completed", "stage": "done"}],
        "status_items": [{"id": "cached-live-upload", "status": "receiving", "stage": "receiving"}],
        "active_count": 1,
        "failed_count": 0,
        "page": 2,
        "page_size": 10,
        "total": 11,
        "pages": 2,
    }
    app.state.services.uploads._page_cache[(2, 10)] = cached
    monkeypatch.setattr(api_module, "API_READ_DEADLINE_SECONDS", 0.01)

    async def delayed(function, /, *args, **kwargs):
        await asyncio.sleep(0.05)
        return function(*args, **kwargs)

    monkeypatch.setattr(api_module, "_background_api", delayed)

    async def scenario() -> None:
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            response = await client.get("/api/v1/uploads?page=2&page_size=10", headers=headers)
            assert response.status_code == 200
            assert response.json() == cached
            await asyncio.sleep(0.06)

    asyncio.run(scenario())


def test_failed_jobs_can_be_retried_without_overwriting_history(tmp_path: Path):
    app, headers = make_app(tmp_path)
    failed_id = app.state.services.jobs.enqueue(
        "scan_source",
        {"source_id": "source-retry"},
        dedupe_key="scan:source-retry",
    )
    with app.state.services.db.transaction() as connection:
        connection.execute(
            "UPDATE jobs SET status='failed',progress=.35,error='temporary failure',finished_at=? WHERE id=?",
            (utcnow(), failed_id),
        )

    async def scenario() -> None:
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            response = await client.post(f"/api/v1/jobs/{failed_id}/retry", headers=headers, json={})
            assert response.status_code == 202
            retried_id = response.json()["job_id"]
            assert response.json()["retried_from"] == failed_id
            duplicate = await client.post(f"/api/v1/jobs/{failed_id}/retry", headers=headers, json={})
            assert duplicate.json()["job_id"] == retried_id
            active = await client.post(f"/api/v1/jobs/{retried_id}/retry", headers=headers, json={})
            assert active.status_code == 409
            missing = await client.post("/api/v1/jobs/missing/retry", headers=headers, json={})
            assert missing.status_code == 404

        with app.state.services.db.read() as connection:
            failed = connection.execute("SELECT status,error FROM jobs WHERE id=?", (failed_id,)).fetchone()
            retried = connection.execute(
                "SELECT kind,payload_json,dedupe_key,status,attempts FROM jobs WHERE id=?",
                (retried_id,),
            ).fetchone()
        assert dict(failed) == {"status": "failed", "error": "temporary failure"}
        assert retried["kind"] == "scan_source"
        assert json.loads(retried["payload_json"]) == {"source_id": "source-retry"}
        assert retried["dedupe_key"] == "scan:source-retry"
        assert (retried["status"], retried["attempts"]) == ("queued", 0)

    asyncio.run(scenario())


def test_failed_jobs_can_be_retried_in_bulk_by_lane_without_replaying_superseded_history(
    tmp_path: Path,
):
    app, headers = make_app(tmp_path)
    jobs = app.state.services.jobs
    scan_failed = jobs.enqueue(
        "scan_source", {"source_id": "source-bulk"}, dedupe_key="scan:source-bulk"
    )
    audio_failed = jobs.enqueue(
        "transcribe_audio", {"video_id": "video-bulk"}, dedupe_key="audio:video-bulk"
    )
    superseded_failed = jobs.enqueue(
        "index_video", {"video_id": "video-superseded"}, dedupe_key="index:video-superseded"
    )
    with app.state.services.db.transaction() as connection:
        connection.execute(
            "UPDATE jobs SET status='failed',error='temporary failure',finished_at=? "
            "WHERE id IN (?,?,?)",
            (utcnow(), scan_failed, audio_failed, superseded_failed),
        )
    superseding_job = jobs.enqueue(
        "index_video", {"video_id": "video-superseded"}, dedupe_key="index:video-superseded"
    )

    async def scenario() -> None:
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            scan_page = await client.get(
                "/api/v1/jobs?page=1&page_size=10&lane=scan", headers=headers
            )
            assert scan_page.status_code == 200
            assert scan_page.json()["retryable_failed_count"] == 1

            retried_scan = await client.post(
                "/api/v1/jobs/retry-failed?lane=scan", headers=headers, json={}
            )
            assert retried_scan.status_code == 202
            assert retried_scan.json()["retried"] == 1
            assert retried_scan.json()["lane"] == "scan"

            duplicate = await client.post(
                "/api/v1/jobs/retry-failed?lane=scan", headers=headers, json={}
            )
            assert duplicate.status_code == 202
            assert duplicate.json()["retried"] == 0

            all_page = await client.get("/api/v1/jobs?page=1&page_size=10", headers=headers)
            assert all_page.json()["retryable_failed_count"] == 1
            retried_audio = await client.post("/api/v1/jobs/retry-failed", headers=headers, json={})
            assert retried_audio.status_code == 202
            assert retried_audio.json()["retried"] == 1

        with app.state.services.db.read() as connection:
            old_statuses = {
                row["id"]: row["status"]
                for row in connection.execute(
                    "SELECT id,status FROM jobs WHERE id IN (?,?,?)",
                    (scan_failed, audio_failed, superseded_failed),
                ).fetchall()
            }
            active = connection.execute(
                "SELECT id,kind,dedupe_key FROM jobs WHERE status='queued' ORDER BY created_at"
            ).fetchall()
        assert old_statuses == {
            scan_failed: "failed",
            audio_failed: "failed",
            superseded_failed: "failed",
        }
        assert [row["dedupe_key"] for row in active].count("index:video-superseded") == 1
        assert {row["kind"] for row in active} >= {
            "scan_source",
            "transcribe_audio",
            "index_video",
        }
        assert superseding_job in {row["id"] for row in active}

    asyncio.run(scenario())


def test_series_aggregate_merge_bulk_and_filter(tmp_path: Path):
    app, headers = make_app(tmp_path)
    seed_videos(app)

    async def scenario() -> None:
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            rows = (await client.get("/api/v1/series", headers=headers)).json()
            assert rows[0]["video_count"] == 2
            assert rows[0]["duration_ms"] == 30_000
            renamed = await client.patch(
                "/api/v1/series", headers=headers, json={"name": "SERIES A", "new_name": "Merged"}
            )
            assert renamed.json()["updated"] == 2
            assigned = await client.post(
                "/api/v1/videos/series/bulk",
                headers=headers,
                json={"video_ids": ["video-2"], "series": "Merged"},
            )
            assert assigned.json()["updated"] == 1
            assert len((await client.get("/api/v1/videos?series=Merged", headers=headers)).json()) == 3
            removed = await client.post(
                "/api/v1/videos/series/bulk",
                headers=headers,
                json={"video_ids": ["video-2"], "series": None},
            )
            assert removed.json()["series"] is None

    asyncio.run(scenario())


def test_collections_support_single_membership_order_cover_and_dissolve(tmp_path: Path):
    app, headers = make_app(tmp_path)
    seed_videos(app)

    async def scenario() -> None:
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            first = await client.post(
                "/api/v1/collections",
                headers=headers,
                json={"name": "第一季", "description": "按顺序播放", "video_ids": ["video-0", "video-1"]},
            )
            assert first.status_code == 201
            first_id = first.json()["id"]
            second = await client.post(
                "/api/v1/collections",
                headers=headers,
                json={"name": "临时合集", "video_ids": ["video-2"]},
            )
            second_id = second.json()["id"]
            moved = await client.post(
                f"/api/v1/collections/{first_id}/videos",
                headers=headers,
                json={"video_ids": ["video-2"]},
            )
            assert moved.json()["updated"] == 1
            assert (await client.get(f"/api/v1/collections/{second_id}", headers=headers)).json()[
                "video_count"
            ] == 0

            detail = (await client.get(f"/api/v1/collections/{first_id}", headers=headers)).json()
            assert [video["id"] for video in detail["videos"]] == ["video-0", "video-1", "video-2"]
            reordered = await client.patch(
                f"/api/v1/collections/{first_id}/videos/order",
                headers=headers,
                json={"video_ids": ["video-2", "video-0", "video-1"]},
            )
            assert reordered.json()["updated"] == 3
            invalid = await client.patch(
                f"/api/v1/collections/{first_id}/videos/order",
                headers=headers,
                json={"video_ids": ["video-0"]},
            )
            assert invalid.status_code == 400
            changed = await client.patch(
                f"/api/v1/collections/{first_id}",
                headers=headers,
                json={"description": "更新后的简介", "cover_video_id": "video-1"},
            )
            assert changed.json()["cover_video_id"] == "video-1"
            filtered = (
                await client.get(
                    f"/api/v1/videos?collection_id={first_id}&sort=collection",
                    headers=headers,
                )
            ).json()
            assert [video["id"] for video in filtered] == ["video-2", "video-0", "video-1"]
            assert all(video["collection_id"] == first_id for video in filtered)

            dissolved = await client.delete(f"/api/v1/collections/{first_id}", headers=headers)
            assert dissolved.json()["detached_videos"] == 3
            assert len((await client.get("/api/v1/videos?limit=20", headers=headers)).json()) == 3
            with app.state.services.db.read() as conn:
                assert conn.execute("SELECT count(*) FROM collection_videos").fetchone()[0] == 0

    asyncio.run(scenario())


def test_proxy_is_encrypted_redacted_and_excluded_from_model_status(tmp_path: Path):
    app, headers = make_app(tmp_path)

    async def scenario() -> None:
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            response = await client.patch(
                "/api/v1/models/config",
                headers=headers,
                json={
                    "hf_endpoint": "https://hf-mirror.example",
                    "proxy_url": "http://proxy.example:8080",
                    "proxy_username": "proxy-user",
                    "proxy_password": "proxy-secret",
                },
            )
            assert response.status_code == 200
            assert response.json()["proxy"] == {
                "enabled": True,
                "url": "http://proxy.example:8080",
                "username": "proxy-user",
                "has_password": True,
            }
            status = (await client.get("/api/v1/models", headers=headers)).json()
            assert "proxy-secret" not in json.dumps(status)
            assert "password" not in status["proxy"]
            assert status["accelerator"]["policy"] == "gpu_preferred"
            assert status["accelerator"]["cpu_threads"] == 1
            assert set(status["accelerator"]["components"]) == {"visual", "ocr", "faces", "audio"}
            assert status["pip"]["index_url"] == "https://pypi.tuna.tsinghua.edu.cn/simple"

            disabled = await client.patch(
                "/api/v1/models/config",
                headers=headers,
                json={"proxy_enabled": False},
            )
            assert disabled.json()["proxy"]["enabled"] is False
            assert disabled.json()["proxy"]["has_password"] is True
            assert app.state.services.models._httpx_proxy() is None
            assert app.state.services.models._hf_proxies() is None

            sources = await client.patch(
                "/api/v1/models/config",
                headers=headers,
                json={
                    "pip_index_url": "https://pypi.example/simple/",
                    "pytorch_index_url": "https://torch.example/cpu/",
                },
            )
            assert sources.json()["pip"] == {
                "index_url": "https://pypi.example/simple",
                "pytorch_index_url": "https://torch.example/cpu",
            }
            rejected = await client.patch(
                "/api/v1/models/config",
                headers=headers,
                json={"pip_index_url": "https://user:secret@pypi.example/simple"},
            )
            assert rejected.status_code == 400

    asyncio.run(scenario())
    stored = app.state.services.db.setting("model_proxy_config")
    assert "proxy-secret" not in json.dumps(stored)
    assert app.state.services.models.proxy_config(include_password=True)["password"] == "proxy-secret"


def test_huggingface_downloads_ignore_inherited_proxy_when_switch_is_off_for_non_mirror_endpoint(
    tmp_path: Path, monkeypatch
):
    app, _ = make_app(tmp_path)
    manager = app.state.services.models
    factories = []

    class Session:
        def __init__(self):
            self.trust_env = True
            self.proxies = {}

    monkeypatch.setitem(sys.modules, "requests", types.SimpleNamespace(Session=Session))
    monkeypatch.setitem(
        sys.modules,
        "huggingface_hub",
        types.SimpleNamespace(
            configure_http_backend=lambda *, backend_factory: factories.append(backend_factory)
        ),
    )
    manager.settings.model_hf_endpoint = "https://huggingface.co"
    manager.set_proxy_config("http://proxy.example:8080", enabled=False)
    manager._configure_hf_http_backend()
    direct = factories[-1]()
    assert direct.trust_env is False
    assert direct.proxies == {}

    manager.set_proxy_config("http://proxy.example:8080", enabled=True)
    manager._configure_hf_http_backend()
    proxied = factories[-1]()
    assert proxied.trust_env is False
    assert proxied.proxies == {
        "http": "http://proxy.example:8080",
        "https": "http://proxy.example:8080",
    }


def test_hf_mirror_downloads_stay_direct_even_when_model_proxy_is_enabled(tmp_path: Path, monkeypatch):
    app, _ = make_app(tmp_path)
    manager = app.state.services.models
    factories = []

    class Session:
        def __init__(self):
            self.trust_env = True
            self.proxies = {}

    monkeypatch.setitem(sys.modules, "requests", types.SimpleNamespace(Session=Session))
    monkeypatch.setitem(
        sys.modules,
        "huggingface_hub",
        types.SimpleNamespace(configure_http_backend=lambda *, backend_factory: factories.append(backend_factory)),
    )
    manager.set_proxy_config("http://proxy.example:8080", enabled=True)
    manager.settings.model_hf_endpoint = "https://hf-mirror.com"

    assert manager._hf_proxies() is None
    manager._configure_hf_http_backend()
    direct = factories[-1]()
    assert direct.trust_env is False
    assert direct.proxies == {}


def test_model_uninstall_preserves_indexes_and_reports_usage(tmp_path: Path):
    app, headers = make_app(tmp_path)
    seed_videos(app)
    root = app.state.services.settings.models_dir
    (root / "ocr").mkdir(parents=True)
    for name in ("det.onnx", "rec.onnx", "cls.onnx"):
        (root / "ocr" / name).write_bytes(b"model-contents")
    (root / "manifest.json").write_text(
        json.dumps({"version": "v1", "components": {"ocr": {"version": "ocr-v1"}}}), encoding="utf-8"
    )
    with app.state.services.db.transaction() as conn:
        conn.execute(
            "INSERT INTO frames(id,video_id,timestamp_ms,segment_start_ms,segment_end_ms,thumbnail_path,"
            "vector_blob,created_at) VALUES('frame','video-0',0,0,1,'thumb',?,?)",
            (b"existing-vector", utcnow()),
        )
    app.state.services.accelerator.mark_ready("ocr", "CPU")
    assert app.state.services.accelerator.status()["components"]["ocr"]["state"] == "ready"

    async def scenario() -> None:
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            deleted = await client.delete("/api/v1/models/ocr", headers=headers)
            assert deleted.status_code == 200
            assert deleted.json()["freed_bytes"] > 0
            usage = (await client.get("/api/v1/storage/usage?refresh=true", headers=headers)).json()
            assert set(usage["categories"]) == {
                "database",
                "ai_index",
                "models",
                "thumbnails",
                "preview_cache",
                "remote_cache",
                "upload_staging",
                "other",
            }
            assert usage["categories"]["ai_index"]["mode"] == "pgvector"

    asyncio.run(scenario())
    assert not (root / "ocr").exists()
    assert app.state.services.accelerator.status()["components"]["ocr"]["state"] == "not_loaded"
    with app.state.services.db.read() as conn:
        assert conn.execute("SELECT vector_blob FROM frames WHERE id='frame'").fetchone()[0] == b"existing-vector"


def test_model_uninstall_conflicts_with_install_job(tmp_path: Path):
    app, headers = make_app(tmp_path)
    app.state.services.jobs.enqueue("install_models", {}, dedupe_key="install-models")

    async def scenario() -> None:
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            response = await client.delete("/api/v1/models/all", headers=headers)
            assert response.status_code == 409

    asyncio.run(scenario())