"""Unit tests for the AI 改写 prompt builder (app.gateway.routers.writing).

Focus: the 仿写 (imitate) action injects the reference sample into the prompt,
and existing actions are unaffected.
"""

from pathlib import Path
from types import SimpleNamespace

import pytest
from fastapi import HTTPException

from app.gateway.document_rewrite_diagnostics import build_document_rewrite_diagnostic
from app.gateway.routers.writing import (
    _REFERENCE_MAX_CHARS,
    RewriteRequest,
    SelectionInfo,
    _release_document_rewrite,
    _reserve_document_rewrite,
    build_rewrite_messages,
)


def test_document_rewrite_diagnostic_is_copyable_and_redacts_secrets():
    try:
        raise RuntimeError("upstream rejected api_key=secret-value and Bearer abc.def")
    except RuntimeError as exc:
        diagnostic = build_document_rewrite_diagnostic(
            job_id="job-1",
            stage="generation",
            error_code="RuntimeError",
            message=str(exc),
            exc=exc,
        )

    assert diagnostic["jobId"] == "job-1"
    assert diagnostic["exceptionType"] == "RuntimeError"
    assert "secret-value" not in diagnostic["message"]
    assert "abc.def" not in diagnostic["message"]
    assert "[REDACTED]" in diagnostic["message"]
    assert any("test_writing_rewrite_prompt.py" in line for line in diagnostic["traceTail"])


@pytest.mark.asyncio
async def test_durable_rewrite_start_returns_structured_diagnostic(monkeypatch):
    from app.gateway.routers import writing

    async def broken_path(*_args):
        raise OSError("docker volume is read-only")

    async def current_user(*_args):
        return "user-a"

    monkeypatch.setattr(writing, "aresolve_thread_virtual_path", broken_path)
    monkeypatch.setattr(writing, "get_current_user", current_user)
    request = SimpleNamespace(
        _deerflow_test_bypass_auth=True,
        app=SimpleNamespace(state=SimpleNamespace()),
    )
    body = writing.DocumentRewriteRequest(
        threadId="thread-a",
        path="/mnt/user-data/report.md",
        instruction="统一表达风格",
    )

    with pytest.raises(HTTPException) as captured:
        await writing.start_durable_document_rewrite_job(body=body, request=request)

    assert captured.value.status_code == 500
    assert captured.value.detail["message"] == "docker volume is read-only"
    diagnostic = captured.value.detail["diagnostic"]
    assert diagnostic["stage"] == "start"
    assert diagnostic["errorCode"] == "OSError"
    assert diagnostic["exceptionType"] == "OSError"


def _req(**kwargs) -> RewriteRequest:
    base = {
        "selection": SelectionInfo(**{"from": 0, "to": 4, "text": "原始文字"}),
    }
    base.update(kwargs)
    return RewriteRequest(**base)


def test_imitate_injects_reference_text():
    body = _req(action="imitate", reference_text="这是一段范文，节奏明快，排比有力。")
    action, messages = build_rewrite_messages(body)

    assert action == "imitate"
    human = messages[1].content
    assert "范文" in human
    assert "这是一段范文，节奏明快，排比有力。" in human
    assert "原始文字" in human


def test_imitate_without_reference_has_no_reference_block():
    body = _req(action="imitate", reference_text="   ")
    action, messages = build_rewrite_messages(body)

    assert action == "imitate"
    # No reference → the 范文 block is not appended
    assert "【范文】" not in messages[1].content


def test_imitate_reference_is_truncated():
    long_ref = "甲" * (_REFERENCE_MAX_CHARS + 500)
    body = _req(action="imitate", reference_text=long_ref)
    _, messages = build_rewrite_messages(body)

    human = messages[1].content
    # The reference is capped — the 500 extra chars are dropped
    assert human.count("甲") == _REFERENCE_MAX_CHARS


def test_polish_action_unaffected_by_reference_field():
    # A stray reference_text on a non-imitate action must be ignored
    body = _req(action="polish", reference_text="不应出现的范文")
    action, messages = build_rewrite_messages(body)

    assert action == "polish"
    assert "范文" not in messages[1].content
    assert "不应出现的范文" not in messages[1].content


def test_unknown_action_falls_back_to_polish():
    body = _req(action="not-a-real-action")
    action, _ = build_rewrite_messages(body)
    assert action == "polish"


def test_imitate_follow_up_preserves_reference_and_previous_result():
    body = _req(
        action="imitate",
        reference_text="范文样例",
        previous_result="上一版仿写结果",
        follow_up_instruction="再正式一点",
    )
    _, messages = build_rewrite_messages(body)

    # system, human(original w/ reference), ai(previous), human(follow-up)
    assert len(messages) == 4
    assert "范文样例" in messages[1].content
    assert messages[2].content == "上一版仿写结果"
    assert messages[3].content == "再正式一点"


@pytest.mark.asyncio
async def test_full_document_rewrite_reservation_rejects_a_second_live_request(tmp_path: Path):
    """Two streams must never generate competing commits for one real file."""
    path = tmp_path / "report.md"
    path.write_text("# report", encoding="utf-8")

    assert await _reserve_document_rewrite(path) is True
    try:
        assert await _reserve_document_rewrite(path) is False
    finally:
        await _release_document_rewrite(path)

    assert await _reserve_document_rewrite(path) is True
    await _release_document_rewrite(path)


@pytest.mark.asyncio
async def test_full_document_rewrite_keeps_committed_file_when_undo_snapshot_fails(
    monkeypatch,
    tmp_path: Path,
):
    """Undo persistence is best effort and must not restore a valid rewrite."""
    from app.gateway.routers import writing

    path = tmp_path / "report.md"
    path.write_text("# 原文\n\n正文", encoding="utf-8")

    class _Model:
        async def astream(self, *_args, **_kwargs):
            yield "# 改写后\n\n新正文"

    class _BrokenVersionStore:
        async def create(self, **_kwargs):
            raise RuntimeError("version store unavailable")

    class _Config:
        models: list[object] = []

        @staticmethod
        def get_model_config(_name):
            return None

    async def resolve_path(*_args):
        return path

    async def current_user(*_args):
        return "user-a"

    async def requirements(*_args, **_kwargs):
        yield "保留事实。"

    monkeypatch.setattr(writing, "aresolve_thread_virtual_path", resolve_path)
    monkeypatch.setattr(writing, "get_current_user", current_user)
    monkeypatch.setattr(writing, "create_chat_model", lambda **_kwargs: _Model())
    monkeypatch.setattr(writing, "_astream_rewrite_requirements", requirements)
    monkeypatch.setattr(writing, "_extract_chunk_parts", lambda chunk: ("", chunk))
    response = await writing.rewrite_document(
        body=writing.DocumentRewriteRequest(
            threadId="thread-a",
            path="/mnt/user-data/report.md",
            instruction="统一表达风格",
        ),
        request=SimpleNamespace(
            _deerflow_test_bypass_auth=True,
            app=SimpleNamespace(state=SimpleNamespace(document_rewrite_version_store=_BrokenVersionStore())),
        ),
        config=_Config(),
    )

    events = "".join([frame async for frame in response.body_iterator])

    assert path.read_text(encoding="utf-8") == "# 改写后\n\n新正文"
    assert "event: committed" in events
    assert '"versionSnapshotFailed": true' in events
    assert "event: error" not in events