from __future__ import annotations

from pathlib import Path
from types import SimpleNamespace

import pytest

from app.gateway.routers import artifacts
from deerflow.uploads.policy import default_upload_policy, set_runtime_upload_policy


@pytest.fixture(autouse=True)
def _restore_upload_policy():
    set_runtime_upload_policy(default_upload_policy())
    yield
    set_runtime_upload_policy(default_upload_policy())


@pytest.mark.asyncio
async def test_artifact_preview_reads_small_text_without_unbounded_artifact_response(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    source = tmp_path / "notes.txt"
    source.write_text("第一行\n第二行", encoding="utf-8")

    async def resolve(_thread_id: str, _path: str) -> Path:
        return source

    monkeypatch.setattr(artifacts, "aresolve_thread_virtual_path", resolve)
    response = await artifacts.preview_artifact.__wrapped__(
        thread_id="thread-1",
        request=SimpleNamespace(),
        path="mnt/user-data/uploads/notes.txt",
        max_chars=1024,
    )

    assert response.content.splitlines() == ["第一行", "第二行"]
    assert response.truncated is False
    assert response.strategy == "complete-text"
    assert response.total_bytes == source.stat().st_size


@pytest.mark.asyncio
async def test_artifact_preview_samples_large_text_with_full_file_statistics(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    source = tmp_path / "words.txt"
    source.write_text("".join(f"word-{index:04d}\n" for index in range(1, 1001)), encoding="utf-8")
    policy = default_upload_policy()
    policy["large_file"] = {
        "threshold_bytes": 100,
        "max_result_chars": 4096,
        "hard_max_result_chars": 4096,
    }
    set_runtime_upload_policy(policy)

    async def resolve(_thread_id: str, _path: str) -> Path:
        return source

    monkeypatch.setattr(artifacts, "aresolve_thread_virtual_path", resolve)
    response = await artifacts.preview_artifact.__wrapped__(
        thread_id="thread-1",
        request=SimpleNamespace(),
        path="mnt/user-data/uploads/words.txt",
        max_chars=4096,
    )

    assert response.truncated is True
    assert response.stats is not None
    assert response.stats["lines"] == 1000
    assert "流式扫描" in response.content
    assert "word-0001" in response.content
    assert len(response.content) <= 4096
    assert "sampl" in response.strategy


@pytest.mark.asyncio
async def test_artifact_preview_rejects_binary_content(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    source = tmp_path / "payload.bin"
    source.write_bytes(b"abc\x00def")

    async def resolve(_thread_id: str, _path: str) -> Path:
        return source

    monkeypatch.setattr(artifacts, "aresolve_thread_virtual_path", resolve)
    with pytest.raises(artifacts.HTTPException) as raised:
        await artifacts.preview_artifact.__wrapped__(
            thread_id="thread-1",
            request=SimpleNamespace(),
            path="mnt/user-data/uploads/payload.bin",
            max_chars=1024,
        )
    assert raised.value.status_code == 415