import asyncio
import base64
import json
from pathlib import Path

import httpx
import imagefind.api as api_module
from imagefind import __version__
from imagefind.config import Settings
from imagefind.database import utcnow
from imagefind.main import create_app
from imagefind.speech import SPEECH_INDEX_REVISION
from imagefind.text import search_tokens


def test_setup_login_csrf_and_token(tmp_path: Path):
    app = create_app(Settings(data_dir=tmp_path, embedding_backend="hash", scan_interval_seconds=86400))

    async def scenario():
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            assert (await client.get("/api/v1/status")).json()["configured"] is False
            response = await client.post("/api/v1/setup", json={"password": "a secure test password"})
            assert response.status_code == 200
            csrf = response.json()["csrf_token"]
            assert (await client.get("/api/v1/auth/me")).status_code == 200
            assert (await client.post("/api/v1/tokens", json={"name": "test"})).status_code == 403
            token = await client.post(
                "/api/v1/tokens",
                json={"name": "test"},
                headers={"X-CSRF-Token": csrf},
            )
            assert token.status_code == 201
            api_token = token.json()["token"]
            assert (
                await client.get("/api/v1/sources", headers={"Authorization": f"Bearer {api_token}"})
            ).status_code == 200

    asyncio.run(scenario())


def test_status_frontend_fallback_and_openapi(tmp_path: Path):
    settings = Settings(
        data_dir=tmp_path,
        frontend_dir=tmp_path / "frontend-not-built",
        embedding_backend="hash",
        scan_interval_seconds=86400,
    )
    app = create_app(settings)

    async def scenario():
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            status = await client.get("/api/v1/status")
            assert status.status_code == 200
            assert status.json() == {"configured": False, "version": __version__, "access_mode": "direct"}
            root = await client.get("/")
            assert root.status_code == 200
            assert "ImageFind API 正在运行" in root.text
            openapi = await client.get("/api/openapi.json")
            assert openapi.status_code == 200
            assert "/api/v1/search" in openapi.json()["paths"]

    asyncio.run(scenario())


def test_security_headers_token_scope_payload_and_diagnostics(tmp_path: Path):
    app = create_app(Settings(data_dir=tmp_path, embedding_backend="hash", scan_interval_seconds=86400))

    async def scenario():
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            status = await client.get("/api/v1/status")
            assert status.headers["x-content-type-options"] == "nosniff"
            assert status.headers["referrer-policy"] == "same-origin"
            assert "camera=()" in status.headers["permissions-policy"]
            assert "default-src 'self'" in status.headers["content-security-policy"]

            setup = await client.post("/api/v1/setup", json={"password": "a secure test password"})
            csrf = setup.json()["csrf_token"]
            created = await client.post(
                "/api/v1/tokens",
                headers={"X-CSRF-Token": csrf},
                json={"name": "DAV automation", "scopes": ["webdav"]},
            )
            assert created.status_code == 201
            assert created.json()["scopes"] == ["webdav"]
            listed = await client.get("/api/v1/tokens")
            assert listed.json()[0]["scopes"] == ["webdav"]
            assert "token" not in listed.json()[0]

            diagnostics = await client.get("/api/v1/system/diagnostics")
            assert diagnostics.status_code == 200
            payload = diagnostics.json()
            assert payload["process_rss_bytes"] >= 0
            assert payload["database"]["engine"] == "postgresql"
            assert payload["database"]["pool_max"] >= 1
            assert isinstance(payload["database"]["activity"]["states"], dict)
            assert "running" in payload["inference"]
            assert payload["events"]["subscribers"] == 0

    asyncio.run(scenario())


def test_fnos_gateway_auth_is_bound_to_unix_socket_and_prefix(tmp_path: Path):
    socket_path = tmp_path / "imagefind.sock"
    settings = Settings(
        data_dir=tmp_path / "data",
        gateway_socket=socket_path,
        gateway_prefix="/app/imagefind",
        embedding_backend="hash",
        scan_interval_seconds=86400,
    )
    app = create_app(settings)
    app.state.services.auth.setup("gateway fallback password")

    class UnixSocketScope:
        def __init__(self, target):
            self.target = target

        async def __call__(self, scope, receive, send):
            mounted = dict(scope)
            mounted["server"] = (str(socket_path), None)
            await self.target(mounted, receive, send)

    async def scenario():
        # Header spoofing over TCP must never enable gateway SSO.
        direct_transport = httpx.ASGITransport(app=app)
        spoofed = {
            "X-Trim-Isadmin": "true",
            "X-Trim-Userid": "nas-admin",
            "X-Trim-Username": "Administrator",
        }
        async with httpx.AsyncClient(transport=direct_transport, base_url="http://test") as direct:
            status = await direct.get("/api/v1/status", headers=spoofed)
            assert status.json()["access_mode"] == "direct"
            assert (await direct.post("/api/v1/auth/gateway", headers=spoofed)).status_code == 404
            assert (await direct.get("/api/v1/sources", headers=spoofed)).status_code == 401

        gateway_transport = httpx.ASGITransport(app=UnixSocketScope(app))
        async with httpx.AsyncClient(
            transport=gateway_transport,
            base_url="http://test",
            headers=spoofed,
            follow_redirects=False,
        ) as gateway:
            redirect = await gateway.get("/app/imagefind")
            assert redirect.status_code == 307
            assert redirect.headers["location"] == "/app/imagefind/"
            assert (await gateway.get("/outside-prefix")).status_code == 404
            status = await gateway.get("/app/imagefind/api/v1/status")
            assert status.json()["access_mode"] == "gateway"
            login = await gateway.post("/app/imagefind/api/v1/auth/gateway")
            assert login.status_code == 200
            csrf = login.json()["csrf_token"]
            gateway_token = login.json()["gateway_session_token"]
            assert gateway_token
            assert login.json()["nas_username"] == "Administrator"
            cookie = login.headers["set-cookie"]
            assert "imagefind_gateway_session=" in cookie
            assert "Path=/app/imagefind/" in cookie
            me = await gateway.get("/app/imagefind/api/v1/auth/me")
            assert me.status_code == 200
            assert me.json()["kind"] == "gateway"
            assert (await gateway.post("/app/imagefind/api/v1/tokens", json={"name": "blocked"})).status_code == 403
            created = await gateway.post(
                "/app/imagefind/api/v1/tokens",
                json={"name": "gateway"},
                headers={**spoofed, "X-CSRF-Token": csrf},
            )
            assert created.status_code == 201
            changed_user = await gateway.get(
                "/app/imagefind/api/v1/auth/me",
                headers={**spoofed, "X-Trim-Userid": "another-admin"},
            )
            assert changed_user.status_code == 401

        # The fnOS WebView may omit the scoped cookie. The short-lived session
        # header keeps SSO working, but only inside the trusted Unix gateway.
        async with httpx.AsyncClient(
            transport=gateway_transport,
            base_url="http://test",
            headers={**spoofed, "X-ImageFind-Gateway-Session": gateway_token},
        ) as header_only:
            me = await header_only.get("/app/imagefind/api/v1/auth/me")
            assert me.status_code == 200
            assert me.json()["nas_user_id"] == "nas-admin"

        async with httpx.AsyncClient(
            transport=gateway_transport,
            base_url="http://test",
            headers={
                **spoofed,
                "X-Trim-Userid": "another-admin",
                "X-ImageFind-Gateway-Session": gateway_token,
            },
        ) as wrong_identity:
            assert (await wrong_identity.get("/app/imagefind/api/v1/auth/me")).status_code == 401

        other_headers = {
            "X-Trim-Isadmin": "true",
            "X-Trim-Userid": "other-admin",
            "X-Trim-Username": "Other administrator",
        }
        async with httpx.AsyncClient(
            transport=gateway_transport,
            base_url="http://test",
            headers=other_headers,
        ) as other_gateway:
            other_login = await other_gateway.post("/app/imagefind/api/v1/auth/gateway")
            other_token = other_login.json()["gateway_session_token"]

        async with httpx.AsyncClient(
            transport=gateway_transport,
            base_url="http://test",
            headers={
                **spoofed,
                "Cookie": f"imagefind_gateway_session={other_token}",
                "X-ImageFind-Gateway-Session": gateway_token,
            },
        ) as stale_cookie:
            me = await stale_cookie.get("/app/imagefind/api/v1/auth/me")
            assert me.status_code == 200
            assert me.json()["nas_user_id"] == "nas-admin"

        async with httpx.AsyncClient(transport=direct_transport, base_url="http://test") as direct:
            response = await direct.get(
                "/api/v1/auth/me",
                headers={**spoofed, "X-ImageFind-Gateway-Session": gateway_token},
            )
            assert response.status_code == 401

        async with httpx.AsyncClient(
            transport=gateway_transport,
            base_url="http://test",
            headers={"X-Trim-Isadmin": "false", "X-Trim-Userid": "ordinary-user"},
        ) as ordinary:
            assert (await ordinary.post("/app/imagefind/api/v1/auth/gateway")).status_code == 403
        async with httpx.AsyncClient(transport=gateway_transport, base_url="http://test") as anonymous:
            assert (await anonymous.post("/app/imagefind/api/v1/auth/gateway")).status_code == 401

    asyncio.run(scenario())


def test_frontend_runtime_base_is_injected_for_gateway_and_direct_access(tmp_path: Path):
    frontend = tmp_path / "frontend"
    frontend.mkdir()
    (frontend / "index.html").write_text("<html><head></head><body>ImageFind</body></html>")
    socket_path = tmp_path / "imagefind.sock"
    settings = Settings(
        data_dir=tmp_path / "data",
        frontend_dir=frontend,
        gateway_socket=socket_path,
        embedding_backend="hash",
        scan_interval_seconds=86400,
    )
    app = create_app(settings)

    class UnixSocketScope:
        async def __call__(self, scope, receive, send):
            mounted = dict(scope)
            mounted["server"] = (str(socket_path), None)
            await app(mounted, receive, send)

    async def scenario():
        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as direct:
            index = await direct.get("/")
            assert 'window.__IMAGEFIND_BASE__="/"' in index.text
            assert 'dataset.imagefindAccess="direct"' in index.text
            assert index.headers["cache-control"] == "no-store"
        async with httpx.AsyncClient(
            transport=httpx.ASGITransport(app=UnixSocketScope()), base_url="http://test"
        ) as gateway:
            index = await gateway.get("/app/imagefind/")
            assert 'window.__IMAGEFIND_BASE__="/app/imagefind/"' in index.text
            assert 'dataset.imagefindAccess="gateway"' in index.text

    asyncio.run(scenario())


def _seed_local_search(app, tmp_path: Path) -> tuple[str, str, bytes]:
    service = app.state.services
    source_id = "local-source"
    video_id = "local-video"
    frame_id = "local-frame"
    video_bytes = b"0123456789abcdef"
    video_path = tmp_path / "sample.mp4"
    video_path.write_bytes(video_bytes)
    now = utcnow()
    with service.db.transaction() as conn:
        conn.execute(
            "INSERT INTO sources(id,kind,name,config_json,created_at,updated_at) VALUES(?,?,?,?,?,?)",
            (source_id, "local", "本地资料库", json.dumps({"path": str(tmp_path)}), now, now),
        )
        conn.execute(
            "INSERT INTO videos(id,source_id,source_key,display_name,location,fingerprint,duration_ms,status,"
            "available,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?,?)",
            (
                video_id,
                source_id,
                "sample.mp4",
                "海边假期.mp4",
                str(video_path),
                "fingerprint",
                120_000,
                "ready",
                1,
                now,
                now,
            ),
        )
        conn.execute(
            "INSERT INTO frames(id,video_id,timestamp_ms,segment_start_ms,segment_end_ms,thumbnail_path,created_at) "
            "VALUES(?,?,?,?,?,?,?)",
            (frame_id, video_id, 42_000, 40_000, 48_000, str(tmp_path / "frame.webp"), now),
        )
        tokens = " ".join(search_tokens("海边日落"))
        for kind in ("subtitle", "filename"):
            entry_id = f"text-{kind}"
            conn.execute(
                "INSERT INTO text_entries(id,video_id,frame_id,kind,start_ms,end_ms,raw_text,tokens,created_at) "
                "VALUES(?,?,?,?,?,?,?,?,?)",
                (entry_id, video_id, frame_id, kind, 40_000, 48_000, "海边日落", tokens, now),
            )
            conn.execute("INSERT INTO text_fts(entry_id,tokens) VALUES(?,?)", (entry_id, tokens))
        audio_tokens = " ".join(search_tokens("海浪声音"))
        conn.execute(
            "INSERT INTO text_entries(id,video_id,frame_id,kind,start_ms,end_ms,raw_text,tokens,created_at) "
            "VALUES('text-audio',?,?, 'audio',?,?,?,?,?)",
            (video_id, frame_id, 65_000, 68_000, "远处传来海浪声音", audio_tokens, now),
        )
        conn.execute("INSERT INTO text_fts(entry_id,tokens) VALUES('text-audio',?)", (audio_tokens,))
    _, token = service.auth.create_api_token("test")
    return token, video_id, video_bytes


def test_gateway_media_token_reads_native_media_without_cookie_or_session_header(tmp_path: Path):
    socket_path = tmp_path / "imagefind.sock"
    settings = Settings(
        data_dir=tmp_path / "data",
        gateway_socket=socket_path,
        gateway_prefix="/app/imagefind",
        embedding_backend="hash",
        scan_interval_seconds=86400,
    )
    settings.prepare()
    app = create_app(settings)
    service = app.state.services
    service.auth.setup("gateway native media test password")
    _, video_id, video_bytes = _seed_local_search(app, tmp_path)
    thumbnail_path = tmp_path / "frame.webp"
    thumbnail_path.write_bytes(b"webp-thumbnail")
    preview = settings.preview_dir / "media123"
    preview.mkdir(parents=True)
    (preview / "index.m3u8").write_text("#EXTM3U\nsegment-00001.ts\n", encoding="utf-8")
    (preview / "segment-00001.ts").write_bytes(b"gateway-segment")

    class UnixSocketScope:
        async def __call__(self, scope, receive, send):
            mounted = dict(scope)
            mounted["server"] = (str(socket_path), None)
            await app(mounted, receive, send)

    identity = {
        "X-Trim-Isadmin": "true",
        "X-Trim-Userid": "nas-admin",
        "X-Trim-Username": "Administrator",
    }
    gateway_transport = httpx.ASGITransport(app=UnixSocketScope())

    async def scenario():
        async with httpx.AsyncClient(
            transport=gateway_transport,
            base_url="http://test",
            headers=identity,
        ) as login_client:
            login = await login_client.post("/app/imagefind/api/v1/auth/gateway")
            assert login.status_code == 200
            session_token = login.json()["gateway_session_token"]
            media_token = login.json()["gateway_media_token"]
            assert media_token and media_token != session_token
            refreshed = await login_client.post(
                "/app/imagefind/api/v1/auth/gateway/media-token",
                headers={"X-CSRF-Token": login.json()["csrf_token"]},
            )
            assert refreshed.status_code == 200
            refreshed_token = refreshed.json()["gateway_media_token"]

        query = f"media_token={media_token}"
        async with httpx.AsyncClient(
            transport=gateway_transport,
            base_url="http://test",
            headers=identity,
        ) as native:
            thumbnail = await native.get(f"/app/imagefind/api/v1/frames/local-frame/thumbnail?{query}")
            assert thumbnail.status_code == 200
            assert thumbnail.content == b"webp-thumbnail"
            stream = await native.get(
                f"/app/imagefind/api/v1/videos/{video_id}/stream?{query}",
                headers={"Range": "bytes=2-5"},
            )
            assert stream.status_code == 206
            assert stream.content == video_bytes[2:6]
            download = await native.get(f"/app/imagefind/api/v1/videos/{video_id}/download?{query}")
            assert download.status_code == 200
            assert download.content == video_bytes
            playlist = await native.get(f"/app/imagefind/api/v1/previews/media123/index.m3u8?{query}")
            assert playlist.status_code == 200
            assert f"segment-00001.ts?media_token={media_token}" in playlist.text
            segment = await native.get(
                f"/app/imagefind/api/v1/previews/media123/segment-00001.ts?{query}"
            )
            assert segment.status_code == 200
            assert segment.content == b"gateway-segment"
            assert (await native.get(f"/app/imagefind/api/v1/sources?{query}")).status_code == 401

        async with httpx.AsyncClient(
            transport=gateway_transport,
            base_url="http://test",
            headers={**identity, "X-Trim-Userid": "another-admin"},
        ) as wrong_user:
            assert (
                await wrong_user.get(f"/app/imagefind/api/v1/videos/{video_id}/stream?{query}")
            ).status_code == 401

        async with httpx.AsyncClient(
            transport=httpx.ASGITransport(app=app), base_url="http://test"
        ) as direct:
            assert (await direct.get(f"/api/v1/videos/{video_id}/stream?{query}")).status_code == 401

        with service.db.transaction() as conn:
            conn.execute(
                "UPDATE gateway_media_tokens SET expires_at='2000-01-01T00:00:00+00:00' WHERE token_hash=?",
                (service.auth._digest(refreshed_token),),
            )
        async with httpx.AsyncClient(
            transport=gateway_transport,
            base_url="http://test",
            headers=identity,
        ) as expired:
            assert (
                await expired.get(
                    f"/app/imagefind/api/v1/videos/{video_id}/stream?media_token={refreshed_token}"
                )
            ).status_code == 401

        service.auth.logout(session_token)
        async with httpx.AsyncClient(
            transport=gateway_transport,
            base_url="http://test",
            headers=identity,
        ) as revoked:
            assert (
                await revoked.get(f"/app/imagefind/api/v1/videos/{video_id}/stream?{query}")
            ).status_code == 401

    asyncio.run(scenario())


def test_search_match_sources_local_range_and_hls_paths(tmp_path: Path):
    settings = Settings(data_dir=tmp_path, embedding_backend="hash", scan_interval_seconds=86400)
    settings.prepare()
    app = create_app(settings)
    token, video_id, video_bytes = _seed_local_search(app, tmp_path)
    headers = {"Authorization": f"Bearer {token}"}

    preview = settings.preview_dir / "abc123"
    preview.mkdir(parents=True)
    (preview / "index.m3u8").write_bytes(b"#EXTM3U\n")
    (preview / "segment-00001.ts").write_bytes(b"segment")

    async def scenario():
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            search = await client.post("/api/v1/search", headers=headers, json={"text": "海边日落"})
            assert search.status_code == 200
            item = search.json()["items"][0]
            assert item["video_id"] == video_id
            assert item["timestamp_ms"] == 40_000
            assert item["match_sources"] == ["filename", "subtitle"]
            assert {detail["type"] for detail in item["match_details"]} == {"metadata", "subtitle"}

            audio = await client.post(
                "/api/v1/search",
                headers=headers,
                json={"text": "海浪声音", "recognition_types": ["audio"]},
            )
            audio_item = audio.json()["items"][0]
            assert audio_item["match_sources"] == ["audio"]
            assert audio_item["segment_start_ms"] == 65_000
            assert audio_item["match_details"][0]["text"] == "远处传来海浪声音"
            excluded = await client.post(
                "/api/v1/search",
                headers=headers,
                json={"text": "海浪声音", "recognition_types": ["ocr"]},
            )
            assert excluded.json()["items"] == []

            partial = await client.get(
                f"/api/v1/videos/{video_id}/stream",
                headers={**headers, "Range": "bytes=2-5"},
            )
            assert partial.status_code == 206
            assert partial.content == video_bytes[2:6]
            assert partial.headers["content-range"] == f"bytes 2-5/{len(video_bytes)}"
            assert partial.headers["accept-ranges"] == "bytes"

            suffix = await client.get(
                f"/api/v1/videos/{video_id}/stream",
                headers={**headers, "Range": "bytes=-3"},
            )
            assert suffix.status_code == 206
            assert suffix.content == video_bytes[-3:]
            invalid = await client.get(
                f"/api/v1/videos/{video_id}/stream",
                headers={**headers, "Range": "bytes=99-100"},
            )
            assert invalid.status_code == 416

            playlist = await client.get("/api/v1/previews/abc123/index.m3u8", headers=headers)
            assert playlist.status_code == 200
            assert playlist.content == b"#EXTM3U\n"
            segment = await client.get("/api/v1/previews/abc123/segment-00001.ts", headers=headers)
            assert segment.status_code == 200
            assert segment.content == b"segment"
            assert (await client.get("/api/v1/previews/not-safe/index.m3u8", headers=headers)).status_code == 404
            assert (await client.get("/api/v1/previews/abc123/metadata.json", headers=headers)).status_code == 404

    asyncio.run(scenario())


def test_audio_search_has_an_independent_candidate_budget_and_simplified_traditional_variants(
    tmp_path: Path,
):
    settings = Settings(data_dir=tmp_path, embedding_backend="hash", scan_interval_seconds=86400)
    settings.prepare()
    app = create_app(settings)
    token, video_id, _ = _seed_local_search(app, tmp_path)
    service = app.state.services
    now = utcnow()
    simplified = "繁体关键词"
    traditional = "繁體關鍵詞"
    with service.db.transaction() as conn:
        for index in range(520):
            tokens = " ".join(search_tokens(traditional))
            entry_id = f"crowding-ocr-{index:03d}"
            conn.execute(
                "INSERT INTO text_entries(id,video_id,frame_id,kind,start_ms,end_ms,raw_text,tokens,created_at) "
                "VALUES(?,?,?,'ocr',0,1000,?,?,?)",
                (entry_id, video_id, "local-frame", traditional, tokens, now),
            )
        tokens = " ".join(search_tokens(simplified))
        conn.execute(
            "INSERT INTO text_entries(id,video_id,frame_id,kind,start_ms,end_ms,raw_text,tokens,created_at) "
            "VALUES('audio-simplified',?,?, 'audio',66000,69000,?,?,?)",
            (video_id, "local-frame", f"这里说的是{simplified}", tokens, now),
        )

    async def scenario():
        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
            response = await client.post(
                "/api/v1/search",
                headers={"Authorization": f"Bearer {token}"},
                json={"text": traditional},
            )
            assert response.status_code == 200
            item = response.json()["items"][0]
            assert "audio" in item["match_sources"]
            assert any(detail["type"] == "audio" for detail in item["match_details"])

    asyncio.run(scenario())


def test_transcript_coverage_pagination_and_manual_reindex(tmp_path: Path, monkeypatch):
    settings = Settings(data_dir=tmp_path, embedding_backend="hash", scan_interval_seconds=86400)
    settings.prepare()
    app = create_app(settings)
    token, video_id, _ = _seed_local_search(app, tmp_path)
    service = app.state.services
    monkeypatch.setattr(service.models, "runnable_component_versions", lambda: {"audio": "audio-v1"})
    now = utcnow()
    with service.db.transaction() as conn:
        conn.execute(
            "UPDATE videos SET audio_model_version='audio-v1',audio_index_revision=?,"
            "audio_detected_language='zh',audio_quality_score=.94,audio_rejected_segments=1,"
            "audio_quality_flags_json='not-json' WHERE id=?",
            (SPEECH_INDEX_REVISION, video_id),
        )
        conn.execute("DELETE FROM text_entries WHERE video_id=? AND kind='audio'", (video_id,))
        for index in range(35):
            text = f"第{index + 1}个中文音频片段"
            conn.execute(
                "INSERT INTO text_entries(id,video_id,frame_id,kind,start_ms,end_ms,raw_text,tokens,created_at) "
                "VALUES(?,?,NULL,'audio',?,?,?,?,?)",
                (
                    f"transcript-{index:02d}",
                    video_id,
                    index * 1000,
                    index * 1000 + 900,
                    text,
                    " ".join(search_tokens(text)),
                    now,
                ),
            )

        # A trusted legacy transcript remains searchable while the current model reindex is queued.
        conn.execute("UPDATE videos SET audio_model_version='audio-legacy' WHERE id=?", (video_id,))

    async def scenario():
        headers = {"Authorization": f"Bearer {token}"}
        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
            coverage = await client.get("/api/v1/search/coverage", headers=headers)
            assert coverage.status_code == 200
            coverage_payload = coverage.json()
            assert coverage_payload["indexed"] == 0
            assert coverage_payload["searchable"] == 1
            assert coverage_payload["empty"] == 0
            assert coverage_payload["percent"] == 100.0
            assert coverage_payload["current_model_percent"] == 0.0

            with service.db.transaction() as conn:
                conn.execute(
                    "UPDATE videos SET audio_model_version='audio-v1' WHERE id=?", (video_id,)
                )
            current_coverage = await client.get("/api/v1/search/coverage", headers=headers)
            assert current_coverage.status_code == 200
            assert current_coverage.json()["indexed"] == 1
            assert current_coverage.json()["searchable"] == 1
            assert current_coverage.json()["current_model_percent"] == 100.0

            transcript = await client.get(
                f"/api/v1/videos/{video_id}/transcript?page=2&page_size=10", headers=headers
            )
            assert transcript.status_code == 200
            payload = transcript.json()
            assert payload["status"] == "ready"
            assert payload["page"] == 2 and payload["pages"] == 4 and payload["total"] == 35
            assert payload["items"][0]["start_ms"] == 10_000
            assert payload["detected_language"] == "zh"
            assert payload["quality_state"] == "ready"
            assert payload["aggregate_risk"] is False
            assert payload["repeated_phrases"] == []
            assert payload["quality_score_semantics"] == "rule_check"
            assert payload["quality_flags"] == []

            speech = await client.get("/api/v1/speech/config", headers=headers)
            assert speech.status_code == 200
            assert speech.json()["language_policy"] == "zh_priority"
            configured = await client.patch(
                "/api/v1/speech/config",
                headers=headers,
                json={"language_policy": "auto", "quality_profile": "balanced"},
            )
            assert configured.status_code == 200
            assert configured.json()["quality_profile"] == "balanced"

            queued = await client.post(
                f"/api/v1/videos/{video_id}/transcript/reindex", headers=headers, json={"language": "zh"}
            )
            assert queued.status_code == 202
            with service.db.read() as conn:
                job = conn.execute(
                    "SELECT kind,priority,payload_json FROM jobs WHERE id=?", (queued.json()["job_id"],)
                ).fetchone()
            assert {"kind": job["kind"], "priority": job["priority"]} == {
                "kind": "transcribe_audio",
                "priority": 0,
            }
            assert json.loads(job["payload_json"]) == {"video_id": video_id, "language": "zh"}

    asyncio.run(scenario())


def test_webdav_range_proxy_forwards_headers_and_hides_credentials(tmp_path: Path, monkeypatch):
    socket_path = tmp_path / "imagefind.sock"
    settings = Settings(
        data_dir=tmp_path,
        gateway_socket=socket_path,
        gateway_prefix="/app/imagefind",
        embedding_backend="hash",
        scan_interval_seconds=86400,
    )
    settings.prepare()
    app = create_app(settings)
    service = app.state.services
    service.auth.setup("gateway WebDAV media test password")
    now = utcnow()
    secret_blob = service.secrets.encrypt_json({"password": "remote-password"})
    with service.db.transaction() as conn:
        conn.execute(
            "INSERT INTO sources(id,kind,name,config_json,secret_blob,created_at,updated_at) "
            "VALUES(?,?,?,?,?,?,?)",
            (
                "dav-source",
                "webdav",
                "远程资料库",
                json.dumps(
                    {
                        "base_url": "https://dav.example/videos/",
                        "username": "remote-user",
                        "verify_tls": True,
                    }
                ),
                secret_blob,
                now,
                now,
            ),
        )
        conn.execute(
            "INSERT INTO videos(id,source_id,source_key,display_name,location,fingerprint,status,available,"
            "created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?)",
            (
                "dav-video",
                "dav-source",
                "movie.mp4",
                "movie.mp4",
                "https://dav.example/videos/movie.mp4",
                "remote-fingerprint",
                "ready",
                1,
                now,
                now,
            ),
        )
    _, token = service.auth.create_api_token("test")
    service.db.set_setting(
        "webdav_server", {"enabled": True, "source_id": "dav-source", "relative_path": ""}
    )
    webdav_auth = {
        "Authorization": "Basic " + base64.b64encode(f"imagefind:{token}".encode()).decode()
    }
    observed: dict[str, object] = {"requests": []}

    class UnixSocketScope:
        async def __call__(self, scope, receive, send):
            mounted = dict(scope)
            mounted["server"] = (str(socket_path), None)
            await app(mounted, receive, send)

    identity = {
        "X-Trim-Isadmin": "true",
        "X-Trim-Userid": "nas-admin",
        "X-Trim-Username": "Administrator",
    }

    class FakeAsyncClient:
        def __init__(self, **kwargs):
            observed["client_kwargs"] = kwargs

        def build_request(self, method, url, headers):
            request = httpx.Request(method, url, headers=headers)
            observed["requests"].append(request)
            return request

        async def send(self, request, stream=False):
            assert stream is True
            if request.method == "HEAD":
                return httpx.Response(405, request=request)
            if request.headers.get("range") == "bytes=0-0":
                return httpx.Response(
                    206,
                    request=request,
                    content=b"0",
                    headers={
                        "Content-Length": "1",
                        "Content-Range": "bytes 0-0/10",
                        "Accept-Ranges": "bytes",
                        "Content-Type": "application/octet-stream",
                    },
                )
            return httpx.Response(
                206,
                request=request,
                content=b"2345",
                headers={
                    "Content-Length": "4",
                    "Content-Range": "bytes 2-5/10",
                    "Accept-Ranges": "bytes",
                    "Content-Type": "application/octet-stream",
                },
            )

        async def aclose(self):
            observed["closed"] = True

    async def scenario():
        transport = httpx.ASGITransport(app=app)
        gateway_transport = httpx.ASGITransport(app=UnixSocketScope())
        async with (
            httpx.AsyncClient(transport=transport, base_url="http://test") as client,
            httpx.AsyncClient(
                transport=gateway_transport,
                base_url="http://test",
                headers=identity,
            ) as gateway,
        ):
            login = await gateway.post("/app/imagefind/api/v1/auth/gateway")
            assert login.status_code == 200
            media_token = login.json()["gateway_media_token"]
            monkeypatch.setattr(api_module.httpx, "AsyncClient", FakeAsyncClient)
            response = await client.get(
                "/api/v1/videos/dav-video/stream",
                headers={"Authorization": f"Bearer {token}", "Range": "bytes=2-5", "If-Range": '"etag"'},
            )
            assert response.status_code == 206
            assert response.content == b"2345"
            assert response.headers["content-range"] == "bytes 2-5/10"
            assert response.headers["content-type"] == "video/mp4"
            gateway_response = await gateway.get(
                f"/app/imagefind/api/v1/videos/dav-video/stream?media_token={media_token}",
                headers={"Range": "bytes=2-5", "If-Range": '"etag"'},
            )
            assert gateway_response.status_code == 206
            assert gateway_response.content == b"2345"
            assert gateway_response.headers["content-type"] == "video/mp4"
            head = await client.request(
                "HEAD",
                "/webdav/movie.mp4",
                headers=webdav_auth,
            )
            assert head.status_code == 200
            assert head.headers["content-length"] == "10"
            assert "content-range" not in head.headers
            assert head.headers["accept-ranges"] == "bytes"
            assert head.headers["content-type"] == "video/mp4"
            ranged_head = await client.request(
                "HEAD",
                "/webdav/movie.mp4",
                headers={**webdav_auth, "Range": "bytes=2-5"},
            )
            assert ranged_head.status_code == 206
            assert ranged_head.headers["content-length"] == "4"
            assert ranged_head.headers["content-range"] == "bytes 2-5/10"

    asyncio.run(scenario())
    requests = observed["requests"]
    assert isinstance(requests, list)
    assert [request.method for request in requests] == ["GET", "GET", "HEAD", "GET", "HEAD", "GET"]
    upstream = requests[0]
    assert isinstance(upstream, httpx.Request)
    assert upstream.url == "https://dav.example/videos/movie.mp4"
    assert upstream.headers["range"] == "bytes=2-5"
    assert upstream.headers["if-range"] == '"etag"'
    assert "remote-password" not in str(upstream.url)
    assert observed["closed"] is True