from __future__ import annotations

import asyncio
import base64
import hashlib
import sys
import time
from pathlib import Path

import httpx
import pytest
from imagefind import webdav as webdav_module
from imagefind.collections import assign_videos, set_collection_tags
from imagefind.config import Settings
from imagefind.database import utcnow
from imagefind.main import create_app
from imagefind.webdav import _path_lock


@pytest.fixture(autouse=True)
def _inline_webdav_background_io(monkeypatch):
    """Avoid Python 3.13's sandbox-only to_thread selector deadlock.

    fnOS runs Python 3.12 and production keeps these calls in worker threads;
    the test suite still exercises the same write, hash and state logic inline.
    """

    async def inline(_app, function, /, *args, **kwargs):
        return function(*args, **kwargs)

    monkeypatch.setattr(webdav_module, "_background_io", inline)


def _app(tmp_path: Path, **overrides):
    settings = Settings(
        data_dir=tmp_path / "data",
        embedding_backend="hash",
        upload_chunk_mb=1,
        upload_staging_gb=1,
        upload_reserve_gb=0,
        **overrides,
    )
    settings.prepare()
    app = create_app(settings)
    media = tmp_path / "media"
    media.mkdir()
    source_id = app.state.services.sources.add_local("媒体库", str(media))
    app.state.services.storage.set_writable(source_id, True)
    _, token = app.state.services.auth.create_api_token("test")
    return app, media, source_id, token


def _seed_taxonomy(app, source_id: str) -> tuple[str, str, str, str]:
    now = utcnow()
    with app.state.services.db.transaction() as conn:
        group_id = "kind"
        conn.execute(
            "INSERT INTO tag_groups(id,name,selection_mode,sort_order,created_at,updated_at) "
            "VALUES(?,?,'single',0,?,?)",
            (group_id, "类型", now, now),
        )
        conn.executemany(
            "INSERT INTO tags(id,group_id,name,created_at,updated_at) VALUES(?,?,?,?,?)",
            (("movie", group_id, "电影", now, now), ("course", group_id, "课程", now, now)),
        )
        collection_id = "collection"
        conn.execute(
            "INSERT INTO collections(id,name,description,created_at,updated_at) VALUES(?,?, '',?,?)",
            (collection_id, "旅行", now, now),
        )
        video_id = "existing-video"
        conn.execute(
            "INSERT INTO videos(id,source_id,source_key,display_name,location,size_bytes,fingerprint,"
            "duration_ms,status,created_at,updated_at) VALUES(?,?,?,?,?,1,?,60000,'ready',?,?)",
            (video_id, source_id, "existing.mp4", "existing.mp4", "", "fingerprint", now, now),
        )
    return collection_id, video_id, "movie", "course"


def test_collection_default_tags_apply_now_and_on_future_membership(tmp_path: Path):
    app, _, source_id, _ = _app(tmp_path)
    collection_id, existing_id, movie_tag, course_tag = _seed_taxonomy(app, source_id)
    now = utcnow()
    with app.state.services.db.transaction() as conn:
        assign_videos(conn, collection_id, [existing_id])
        conn.execute("INSERT INTO video_tags(video_id,tag_id) VALUES(?,?)", (existing_id, course_tag))
        applied = set_collection_tags(conn, collection_id, [movie_tag])
        assert applied == 1
        future_id = "future-video"
        conn.execute(
            "INSERT INTO videos(id,source_id,source_key,display_name,location,size_bytes,fingerprint,"
            "duration_ms,status,created_at,updated_at) VALUES(?,?,?,?,?,1,?,60000,'ready',?,?)",
            (future_id, source_id, "future.mp4", "future.mp4", "", "future", now, now),
        )
        assign_videos(conn, collection_id, [future_id])
        assert conn.execute(
            "SELECT tag_id FROM video_tags WHERE video_id=?", (existing_id,)
        ).fetchone()[0] == movie_tag
        assert conn.execute(
            "SELECT tag_id FROM video_tags WHERE video_id=?", (future_id,)
        ).fetchone()[0] == movie_tag
        set_collection_tags(conn, collection_id, [])
        assign_videos(conn, None, [future_id])
        assert conn.execute(
            "SELECT tag_id FROM video_tags WHERE video_id=?", (future_id,)
        ).fetchone()[0] == movie_tag


def test_upload_metadata_is_applied_when_catalogued(tmp_path: Path):
    app, media, source_id, _ = _app(tmp_path)
    collection_id, _, movie_tag, _ = _seed_taxonomy(app, source_id)
    payload = b"video" * 100
    upload = app.state.services.uploads.create(
        source_id,
        "imports",
        "trip.mp4",
        len(payload),
        collection_id=collection_id,
        tag_ids=[movie_tag],
    )
    app.state.services.uploads.receive_chunk(
        upload["id"], 0, payload, hashlib.sha256(payload).hexdigest()
    )
    app.state.services.uploads.complete(upload["id"])
    queued = app.state.services.uploads._get(upload["id"])
    app.state.services.uploads.transfer(queued["job_id"], upload["id"])
    assert (media / "imports" / "trip.mp4").read_bytes() == payload
    app.state.services.scanner.refresh_path("refresh", source_id, "imports/trip.mp4", upload["id"])
    with app.state.services.db.read() as conn:
        video = conn.execute(
            "SELECT id FROM videos WHERE source_id=? AND source_key='imports/trip.mp4'", (source_id,)
        ).fetchone()
        assert conn.execute(
            "SELECT collection_id FROM collection_videos WHERE video_id=?", (video["id"],)
        ).fetchone()[0] == collection_id
        assert conn.execute(
            "SELECT tag_id FROM video_tags WHERE video_id=?", (video["id"],)
        ).fetchone()[0] == movie_tag


def test_video_marker_crud_and_duration_validation(tmp_path: Path):
    app, _, source_id, token = _app(tmp_path)
    _, video_id, _, _ = _seed_taxonomy(app, source_id)
    headers = {"Authorization": f"Bearer {token}"}

    async def scenario():
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            created = await client.post(
                f"/api/v1/videos/{video_id}/markers",
                headers=headers,
                json={"position_ms": 12345},
            )
            assert created.status_code == 201
            marker = created.json()
            updated = await client.patch(
                f"/api/v1/videos/{video_id}/markers/{marker['id']}",
                headers=headers,
                json={"title": "精彩片段"},
            )
            assert updated.json()["title"] == "精彩片段"
            assert (await client.get(f"/api/v1/videos/{video_id}/markers", headers=headers)).json()[0][
                "position_ms"
            ] == 12345
            invalid = await client.post(
                f"/api/v1/videos/{video_id}/markers",
                headers=headers,
                json={"position_ms": 60001},
            )
            assert invalid.status_code == 400
            assert (
                await client.delete(
                    f"/api/v1/videos/{video_id}/markers/{marker['id']}", headers=headers
                )
            ).status_code == 204

    asyncio.run(scenario())


def test_collection_tree_supports_arbitrary_depth_move_and_group_promotion(tmp_path: Path):
    app, _, source_id, token = _app(tmp_path)
    collection_id, video_id, _, _ = _seed_taxonomy(app, source_id)
    headers = {"Authorization": f"Bearer {token}"}

    async def scenario():
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            chapter = (
                await client.post(
                    f"/api/v1/collections/{collection_id}/groups",
                    headers=headers,
                    json={"name": "第一章"},
                )
            ).json()
            section = (
                await client.post(
                    f"/api/v1/collections/{collection_id}/groups",
                    headers=headers,
                    json={"name": "第一节", "parent_id": chapter["id"]},
                )
            ).json()
            assigned = await client.post(
                f"/api/v1/collections/{collection_id}/videos",
                headers=headers,
                json={"video_ids": [video_id], "parent_id": section["id"]},
            )
            assert assigned.status_code == 200
            detail = (await client.get(f"/api/v1/collections/{collection_id}", headers=headers)).json()
            assert detail["items"][0]["children"][0]["children"][0]["video_id"] == video_id
            assert detail["videos"][0]["collection_path"] == ["第一章", "第一节"]

            cycle = await client.patch(
                f"/api/v1/collections/{collection_id}/items/{chapter['id']}/move",
                headers=headers,
                json={"parent_id": section["id"], "position": 0},
            )
            assert cycle.status_code == 400
            removed = await client.delete(
                f"/api/v1/collections/{collection_id}/groups/{chapter['id']}", headers=headers
            )
            assert removed.json()["promoted_items"] == 1
            detail = (await client.get(f"/api/v1/collections/{collection_id}", headers=headers)).json()
            assert detail["items"][0]["name"] == "第一节"
            assert detail["videos"][0]["collection_path"] == ["第一节"]

    asyncio.run(scenario())


def test_webdav_upload_move_browse_range_and_delete_guard(tmp_path: Path):
    app, media, source_id, token = _app(tmp_path)
    bearer = {"Authorization": f"Bearer {token}"}
    basic = {
        "Authorization": "Basic "
        + base64.b64encode(f"imagefind:{token}".encode()).decode()
    }
    payload = b"private-webdav-video"

    async def scenario():
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            configured = await client.patch(
                "/api/v1/webdav/config",
                headers={
                    **bearer,
                    "X-Forwarded-Proto": "http",
                    "X-Forwarded-Host": "192.168.5.100:80",
                    "X-Forwarded-Port": "5666",
                },
                json={"enabled": True, "source_id": source_id, "relative_path": "dav"},
            )
            assert configured.status_code == 200
            assert configured.json()["url"] == "http://192.168.5.100:8765/webdav/"
            assert configured.json()["direct_access"] is True
            assert configured.json()["direct_port"] == 8765
            assert configured.json()["gateway_supported"] is False
            unauthenticated = await client.request("PROPFIND", "/webdav/", headers={"Depth": "0"})
            assert unauthenticated.status_code == 401
            assert unauthenticated.headers["www-authenticate"].startswith("Basic realm=")
            assert (await client.request("MKCOL", "/webdav/旅行", headers=basic)).status_code == 201
            uploaded = await client.put(
                "/webdav/旅行/movie.mp4.part", headers=basic, content=payload
            )
            assert uploaded.status_code == 201
            listing = await client.request(
                "PROPFIND", "/webdav/旅行/", headers={**basic, "Depth": "1"}
            )
            assert listing.status_code == 207
            assert "movie.mp4.part" in listing.text
            moved = await client.request(
                "MOVE",
                "/webdav/旅行/movie.mp4.part",
                headers={
                    **basic,
                    "Destination": "http://test/webdav/%E6%97%85%E8%A1%8C/movie.mp4",
                },
            )
            assert moved.status_code == 201
            upload = app.state.services.uploads.list()[0]
            assert upload["origin"] == "webdav"
            assert upload["collection_id"]
            internal = app.state.services.uploads._get(upload["id"])
            assert internal["webdav_path"] == "旅行/movie.mp4"
            assert internal["content_sha256_verified"] is True
            pending = await client.request(
                "HEAD", "/webdav/旅行/movie.mp4", headers=basic
            )
            assert pending.status_code == 200
            assert pending.headers["x-imagefind-upload-state"] == "accepted"
            pending_listing = await client.request(
                "PROPFIND", "/webdav/旅行/", headers={**basic, "Depth": "1"}
            )
            assert "movie.mp4" in pending_listing.text
            app.state.services.uploads.transfer(internal["job_id"], upload["id"])
            accepted = await client.request(
                "HEAD", "/webdav/旅行/movie.mp4", headers=basic
            )
            assert accepted.status_code == 200
            assert accepted.headers["retry-after"] == "2"
            unavailable = await client.get("/webdav/旅行/movie.mp4", headers=basic)
            assert unavailable.status_code == 503
            assert unavailable.headers["retry-after"] == "2"
            assert int(unavailable.headers["content-length"]) == len(unavailable.content)
            assert int(unavailable.headers["content-length"]) != len(payload)
            app.state.services.scanner.refresh_path(
                "refresh", source_id, "dav/旅行/movie.mp4", upload["id"]
            )
            assert (media / "dav" / "旅行" / "movie.mp4").read_bytes() == payload
            ranged = await client.get(
                "/webdav/旅行/movie.mp4", headers={**basic, "Range": "bytes=0-6"}
            )
            assert ranged.status_code == 206
            assert ranged.content == payload[:7]
            head = await client.request(
                "HEAD", "/webdav/旅行/movie.mp4", headers={**basic, "Range": "bytes=0-6"}
            )
            assert head.status_code == 206
            assert head.headers["content-length"] == "7"
            assert head.headers["content-range"] == f"bytes 0-6/{len(payload)}"
            assert head.content == b""
            guarded = await client.request("DELETE", "/webdav/旅行/movie.mp4", headers=basic)
            assert guarded.status_code == 405

            with app.state.services.db.read() as conn:
                video_id = conn.execute(
                    "SELECT id FROM videos WHERE source_id=? AND source_key=?",
                    (source_id, "dav/旅行/movie.mp4"),
                ).fetchone()["id"]
            deleted = await client.delete(
                f"/api/v1/videos/{video_id}?delete_source=true", headers=bearer
            )
            assert deleted.status_code == 200
            assert app.state.services.uploads._get(upload["id"])["status"] == "completed"
            after_delete = await client.request(
                "PROPFIND", "/webdav/旅行/", headers={**basic, "Depth": "1"}
            )
            assert after_delete.status_code == 207
            assert "movie.mp4" not in after_delete.text
            assert (await client.head("/webdav/旅行/movie.mp4", headers=basic)).status_code == 404
            assert (await client.get("/webdav/旅行/movie.mp4", headers=basic)).status_code == 404

    asyncio.run(scenario())


def test_webdav_maps_nested_directories_to_collection_groups(tmp_path: Path):
    app, media, source_id, token = _app(tmp_path)
    bearer = {"Authorization": f"Bearer {token}"}
    basic = {
        "Authorization": "Basic " + base64.b64encode(f"imagefind:{token}".encode()).decode()
    }

    async def scenario():
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            await client.patch(
                "/api/v1/webdav/config",
                headers=bearer,
                json={"enabled": True, "source_id": source_id, "relative_path": "dav"},
            )
            assert (await client.request("MKCOL", "/webdav/课程", headers=basic)).status_code == 201
            assert (
                await client.request("MKCOL", "/webdav/课程/第一章", headers=basic)
            ).status_code == 201
            assert (
                await client.request("MKCOL", "/webdav/课程/第一章/第一节", headers=basic)
            ).status_code == 201
            uploaded = await client.put(
                "/webdav/课程/第一章/第一节/clip.mp4", headers=basic, content=b"nested-video"
            )
            assert uploaded.status_code == 201
            upload = app.state.services.uploads.list()[0]
            assert upload["collection_parent_id"]
            internal = app.state.services.uploads._get(upload["id"])
            app.state.services.uploads.transfer(internal["job_id"], upload["id"])
            app.state.services.scanner.refresh_path(
                "refresh", source_id, "dav/课程/第一章/第一节/clip.mp4", upload["id"]
            )
            assert (media / "dav" / "课程" / "第一章" / "第一节" / "clip.mp4").is_file()
            listing = await client.request(
                "PROPFIND", "/webdav/课程/第一章/第一节/", headers={**basic, "Depth": "1"}
            )
            assert listing.status_code == 207
            assert "clip.mp4" in listing.text
            collections = (await client.get("/api/v1/collections", headers=bearer)).json()
            detail = (
                await client.get(f"/api/v1/collections/{collections[0]['id']}", headers=bearer)
            ).json()
            assert detail["videos"][0]["collection_path"] == ["第一章", "第一节"]

    asyncio.run(scenario())


def test_webdav_root_upload_is_not_added_to_a_collection(tmp_path: Path, monkeypatch):
    app, media, source_id, token = _app(tmp_path)
    bearer = {"Authorization": f"Bearer {token}"}
    basic = {
        "Authorization": "Basic " + base64.b64encode(f"imagefind:{token}".encode()).decode()
    }

    async def scenario():
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            await client.patch(
                "/api/v1/webdav/config",
                headers=bearer,
                json={"enabled": True, "source_id": source_id, "relative_path": "dav"},
            )
            response = await client.put(
                "/webdav/root.mp4", headers=basic, content=b"root-video"
            )
            assert response.status_code == 201
            upload = app.state.services.uploads.list()[0]
            assert upload["collection_id"] is None
            internal = app.state.services.uploads._get(upload["id"])
            assert internal["webdav_path"] == "root.mp4"
            assert internal["content_sha256"] == hashlib.sha256(b"root-video").hexdigest()
            assert internal["content_sha256_verified"] is True
            assert (await client.get("/api/v1/videos", headers=bearer)).json() == []

            def unexpected_rescan(*_args, **_kwargs):
                raise AssertionError("verified WebDAV PUT must not be hashed again during transfer")

            monkeypatch.setattr(app.state.services.uploads, "_content_sha256", unexpected_rescan)
            app.state.services.uploads.transfer(internal["job_id"], upload["id"])
            assert (await client.get("/api/v1/videos", headers=bearer)).json() == []
            app.state.services.scanner.refresh_path(
                "refresh", source_id, "dav/root.mp4", upload["id"]
            )
            assert (media / "dav" / "root.mp4").read_bytes() == b"root-video"
            videos = (await client.get("/api/v1/videos", headers=bearer)).json()
            assert len(videos) == 1
            assert videos[0]["collection_id"] is None
            listing = await client.request(
                "PROPFIND", "/webdav/", headers={**basic, "Depth": "1"}
            )
            assert listing.status_code == 207
            assert "root.mp4" in listing.text

    asyncio.run(scenario())


def test_webdav_content_range_resumes_and_reports_offset(tmp_path: Path):
    app, media, source_id, token = _app(tmp_path)
    bearer = {"Authorization": f"Bearer {token}"}
    basic = {
        "Authorization": "Basic " + base64.b64encode(f"imagefind:{token}".encode()).decode()
    }

    async def scenario():
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            await client.patch(
                "/api/v1/webdav/config",
                headers=bearer,
                json={"enabled": True, "source_id": source_id, "relative_path": "dav"},
            )
            first = await client.put(
                "/webdav/课程/章节/clip.mp4",
                headers={**basic, "Content-Range": "bytes 0-3/8"},
                content=b"abcd",
            )
            assert first.status_code == 204
            assert first.headers["upload-offset"] == "4"
            assert app.state.services.uploads.list() == []
            head = await client.request(
                "HEAD", "/webdav/课程/章节/clip.mp4", headers=basic
            )
            assert head.status_code == 200
            assert head.headers["upload-offset"] == "4"
            assert head.headers["upload-length"] == "8"
            assert head.headers["x-imagefind-upload-state"] == "partial"

            wrong = await client.put(
                "/webdav/课程/章节/clip.mp4",
                headers={**basic, "Content-Range": "bytes 3-7/8"},
                content=b"defgh",
            )
            assert wrong.status_code == 409
            assert wrong.headers["upload-offset"] == "4"
            second = await client.put(
                "/webdav/课程/章节/clip.mp4",
                headers={**basic, "Content-Range": "bytes 4-7/8"},
                content=b"efgh",
            )
            assert second.status_code == 204
            upload = app.state.services.uploads.list()[0]
            assert upload["collection_id"]
            internal = app.state.services.uploads._get(upload["id"])
            assert internal["content_sha256"] is None
            assert internal["content_sha256_verified"] is False
            app.state.services.uploads.transfer(internal["job_id"], upload["id"])
            assert app.state.services.uploads._get(upload["id"])["content_sha256_verified"] is True
            app.state.services.scanner.refresh_path(
                "refresh", source_id, "dav/课程/章节/clip.mp4", upload["id"]
            )
            assert (media / "dav" / "课程" / "章节" / "clip.mp4").read_bytes() == b"abcdefgh"

    asyncio.run(scenario())


def test_webdav_full_retry_without_checksum_reuses_active_upload(tmp_path: Path):
    app, _, source_id, token = _app(tmp_path)
    bearer = {"Authorization": f"Bearer {token}"}
    basic = {
        "Authorization": "Basic " + base64.b64encode(f"imagefind:{token}".encode()).decode()
    }
    payload = b"response-was-lost"

    async def scenario():
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            await client.patch(
                "/api/v1/webdav/config",
                headers=bearer,
                json={"enabled": True, "source_id": source_id, "relative_path": "dav"},
            )
            first = await client.put("/webdav/retry.mp4", headers=basic, content=payload)
            assert first.status_code == 201
            second = await client.put("/webdav/retry.mp4", headers=basic, content=payload)
            assert second.status_code == 204
            assert second.headers["x-imagefind-deduplicated"] == "true"
            assert second.headers["x-imagefind-upload-id"] == first.headers["x-imagefind-upload-id"]
            assert len(app.state.services.uploads.list()) == 1
            with app.state.services.db.read() as conn:
                assert conn.execute("SELECT count(*) FROM webdav_staging").fetchone()[0] == 0

    asyncio.run(scenario())


def test_webdav_same_size_different_content_is_not_deduplicated(tmp_path: Path):
    app, _, source_id, token = _app(tmp_path)
    bearer = {"Authorization": f"Bearer {token}"}
    basic = {
        "Authorization": "Basic " + base64.b64encode(f"imagefind:{token}".encode()).decode()
    }

    async def scenario():
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            await client.patch(
                "/api/v1/webdav/config",
                headers=bearer,
                json={"enabled": True, "source_id": source_id, "relative_path": "dav"},
            )
            first = await client.put("/webdav/changed.mp4", headers=basic, content=b"first")
            second = await client.put("/webdav/changed.mp4", headers=basic, content=b"other")
            assert first.status_code == 201
            assert second.status_code in {201, 204}
            assert second.headers.get("x-imagefind-deduplicated") is None
            assert len(app.state.services.uploads.list()) == 2

    asyncio.run(scenario())


def test_webdav_head_repairs_delayed_offset_and_locked_put_reports_it(tmp_path: Path):
    app, _, source_id, token = _app(tmp_path)
    bearer = {"Authorization": f"Bearer {token}"}
    basic = {
        "Authorization": "Basic " + base64.b64encode(f"imagefind:{token}".encode()).decode()
    }
    virtual = "course/offset.mp4"

    async def scenario():
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            await client.patch(
                "/api/v1/webdav/config",
                headers=bearer,
                json={"enabled": True, "source_id": source_id, "relative_path": "dav"},
            )
            partial = await client.put(
                f"/webdav/{virtual}",
                headers={**basic, "Content-Range": "bytes 0-3/8"},
                content=b"abcd",
            )
            assert partial.status_code == 204
            with app.state.services.db.transaction() as conn:
                conn.execute(
                    "UPDATE webdav_staging SET received_bytes=1,size_bytes=1 WHERE virtual_path=?",
                    (virtual,),
                )
            head = await client.head(f"/webdav/{virtual}", headers=basic)
            assert head.status_code == 200
            assert head.headers["upload-offset"] == "4"

            lock = _path_lock(virtual)
            assert lock.acquire(blocking=False)
            try:
                locked = await client.put(
                    f"/webdav/{virtual}",
                    headers={**basic, "Content-Range": "bytes 4-7/8"},
                    content=b"efgh",
                )
            finally:
                lock.release()
            assert locked.status_code == 423
            assert locked.headers["upload-offset"] == "4"
            assert locked.headers["retry-after"] == "2"

    asyncio.run(scenario())


def test_webdav_same_path_sha256_uses_safe_instant_deduplication(tmp_path: Path):
    app, _, source_id, token = _app(tmp_path)
    bearer = {"Authorization": f"Bearer {token}"}
    basic = {
        "Authorization": "Basic " + base64.b64encode(f"imagefind:{token}".encode()).decode()
    }
    payload = b"deduplicated-video"
    digest = hashlib.sha256(payload).hexdigest()

    async def scenario():
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            await client.patch(
                "/api/v1/webdav/config",
                headers=bearer,
                json={"enabled": True, "source_id": source_id, "relative_path": "dav"},
            )
            first = await client.put("/webdav/same.mp4", headers=basic, content=payload)
            assert first.status_code == 201
            upload = app.state.services.uploads.list()[0]
            internal = app.state.services.uploads._get(upload["id"])
            app.state.services.uploads.transfer(internal["job_id"], upload["id"])
            app.state.services.scanner.refresh_path(
                "refresh", source_id, "dav/same.mp4", upload["id"]
            )

            duplicate = await client.put(
                "/webdav/same.mp4",
                headers={**basic, "X-Content-SHA256": digest},
                content=payload,
            )
            assert duplicate.status_code == 204
            assert duplicate.headers["x-imagefind-deduplicated"] == "true"
            assert len(app.state.services.uploads.list()) == 1

    asyncio.run(scenario())


def test_webdav_buffer_flush_does_not_create_sparse_offset(tmp_path: Path, monkeypatch):
    app, _, source_id, token = _app(tmp_path)
    bearer = {"Authorization": f"Bearer {token}"}
    basic = {
        "Authorization": "Basic " + base64.b64encode(f"imagefind:{token}".encode()).decode()
    }
    chunk_size = 8 * 1024**2
    payload_size = chunk_size + 1024**2

    class Clock:
        values = iter((0.0, 3.0, 4.0))

        @classmethod
        def monotonic(cls):
            return next(cls.values, 4.0)

    monkeypatch.setattr(webdav_module, "time", Clock)

    async def body():
        yield b"a" * chunk_size
        yield b"b" * (payload_size - chunk_size)

    async def scenario():
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            await client.patch(
                "/api/v1/webdav/config",
                headers=bearer,
                json={"enabled": True, "source_id": source_id, "relative_path": "dav"},
            )
            response = await client.put(
                "/webdav/buffer.mp4",
                headers={**basic, "Content-Length": str(payload_size)},
                content=body(),
            )
            assert response.status_code == 201
            upload = app.state.services.uploads.list()[0]
            assert upload["size_bytes"] == payload_size
            internal = app.state.services.uploads._get(upload["id"])
            assert Path(internal["temp_path"]).stat().st_size == payload_size
            assert Path(internal["temp_path"]).read_bytes() == b"a" * chunk_size + b"b" * (
                payload_size - chunk_size
            )

    asyncio.run(scenario())


def test_webdav_resume_truncates_unconfirmed_tail_and_rejects_short_file(tmp_path: Path):
    staging = tmp_path / "resume.part"
    staging.write_bytes(b"confirmed-unconfirmed")
    descriptor = webdav_module._open_staging(staging, len(b"confirmed"))
    try:
        assert webdav_module.os.fstat(descriptor).st_size == len(b"confirmed")
    finally:
        webdav_module.os.close(descriptor)
    assert staging.read_bytes() == b"confirmed"

    with pytest.raises(OSError, match="短于已确认续传偏移"):
        webdav_module._open_staging(staging, len(b"confirmed") + 1)


def test_webdav_stream_does_not_issue_explicit_burst_disk_flush(tmp_path: Path, monkeypatch):
    app, _, source_id, token = _app(tmp_path)
    bearer = {"Authorization": f"Bearer {token}"}
    basic = {
        "Authorization": "Basic " + base64.b64encode(f"imagefind:{token}".encode()).decode()
    }
    sync_calls: list[int] = []
    def unexpected_fsync(file_descriptor: int) -> None:
        sync_calls.append(webdav_module.os.fstat(file_descriptor).st_size)
        raise AssertionError("WebDAV must not synchronously fsync in the PUT request")

    def unexpected_fdatasync(_file_descriptor: int) -> None:
        raise AssertionError("WebDAV must not synchronously fdatasync while receiving the body")

    monkeypatch.setattr(webdav_module.os, "fsync", unexpected_fsync)
    monkeypatch.setattr(webdav_module.os, "fdatasync", unexpected_fdatasync)

    open_flags: list[int] = []
    real_open = webdav_module.os.open

    def recording_open(path, flags, mode=0o777):
        open_flags.append(flags)
        return real_open(path, flags, mode)

    write_sizes: list[int] = []
    writeback_ranges: list[tuple[int, int]] = []
    real_pwrite_all = webdav_module._pwrite_all

    def recording_pwrite_all(file_descriptor: int, data: bytes, offset: int) -> int:
        write_sizes.append(len(data))
        return real_pwrite_all(file_descriptor, data, offset)

    def recording_writeback(_file_descriptor: int, offset: int, length: int) -> bool:
        writeback_ranges.append((offset, length))
        return True

    monkeypatch.setattr(webdav_module.os, "open", recording_open)
    monkeypatch.setattr(webdav_module, "_pwrite_all", recording_pwrite_all)
    monkeypatch.setattr(webdav_module, "_queue_writeback", recording_writeback)

    payload_size = 9 * 1024**2

    async def body():
        yield b"a" * (8 * 1024**2)
        assert sync_calls == []
        yield b"b" * (payload_size - 8 * 1024**2)
        assert sync_calls == []

    async def scenario():
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            await client.patch(
                "/api/v1/webdav/config",
                headers=bearer,
                json={"enabled": True, "source_id": source_id, "relative_path": "dav"},
            )
            response = await client.put(
                "/webdav/sync-once.mp4",
                headers={**basic, "Content-Length": str(payload_size)},
                content=body(),
            )
            assert response.status_code == 201

    asyncio.run(scenario())
    assert sync_calls == []
    assert open_flags
    assert not any(flags & getattr(webdav_module.os, "O_DSYNC", 0) for flags in open_flags)
    assert write_sizes
    assert max(write_sizes) <= webdav_module.WEBDAV_WRITE_QUANTUM
    assert writeback_ranges
    assert sum(length for _, length in writeback_ranges) == payload_size


def test_webdav_propfind_does_not_block_other_requests(tmp_path: Path, monkeypatch):
    if sys.version_info >= (3, 13):
        pytest.skip("Python 3.13 sandbox thread selector deadlock; fnOS ships Python 3.12")
    app, _, source_id, token = _app(tmp_path)
    bearer = {"Authorization": f"Bearer {token}"}
    basic = {
        "Authorization": "Basic " + base64.b64encode(f"imagefind:{token}".encode()).decode()
    }

    async def threaded(_app, function, /, *args, **kwargs):
        return await asyncio.to_thread(function, *args, **kwargs)

    monkeypatch.setattr(webdav_module, "_background_io", threaded)

    def slow_propfind(*_args, **_kwargs):
        time.sleep(0.25)
        return webdav_module.Response(status_code=207)

    monkeypatch.setattr(webdav_module, "_propfind", slow_propfind)

    async def scenario():
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            await client.patch(
                "/api/v1/webdav/config",
                headers=bearer,
                json={"enabled": True, "source_id": source_id, "relative_path": "dav"},
            )
            request = asyncio.create_task(
                client.request("PROPFIND", "/webdav/", headers={**basic, "Depth": "0"})
            )
            await asyncio.sleep(0.02)
            started = time.monotonic()
            status = await client.get("/api/v1/status")
            elapsed = time.monotonic() - started
            assert status.status_code == 200
            assert elapsed < 0.15
            assert (await request).status_code == 207

    asyncio.run(scenario())


def test_webdav_cannot_be_enabled_without_native_direct_access(tmp_path: Path):
    app, _, source_id, token = _app(tmp_path, direct_access=False, port=9876)
    bearer = {"Authorization": f"Bearer {token}"}

    async def scenario():
        transport = httpx.ASGITransport(app=app, root_path="/app/imagefind")
        async with httpx.AsyncClient(
            transport=transport, base_url="http://192.168.5.100:5666"
        ) as client:
            config = await client.get("/api/v1/webdav/config", headers=bearer)
            assert config.status_code == 200
            assert config.json()["url"] == "http://192.168.5.100:9876/webdav/"
            assert config.json()["direct_access"] is False
            assert config.json()["gateway_supported"] is False
            enabled = await client.patch(
                "/api/v1/webdav/config",
                headers=bearer,
                json={"enabled": True, "source_id": source_id, "relative_path": ""},
            )
            assert enabled.status_code == 409
            assert "直接 Web/API 访问" in enabled.json()["detail"]

    asyncio.run(scenario())


def test_webdav_propfind_hrefs_include_gateway_root_path(tmp_path: Path):
    app, _, source_id, token = _app(tmp_path)
    app.state.services.db.set_setting(
        "webdav_server", {"enabled": True, "source_id": source_id, "relative_path": ""}
    )
    basic = {
        "Authorization": "Basic " + base64.b64encode(f"imagefind:{token}".encode()).decode(),
        "Depth": "1",
    }

    async def scenario():
        transport = httpx.ASGITransport(app=app, root_path="/app/imagefind")
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            response = await client.request("PROPFIND", "/webdav/", headers=basic)
            assert response.status_code == 207
            assert "/app/imagefind/webdav/" in response.text

    asyncio.run(scenario())