hyungi_document_server/tests/test_s1_dedup_shape.py

"""S1-ADD (plan ds-s1-backend-1) A-4 — call-shape regression + md_status 매핑 동작 검증.

검증 대상 (값이 아니라 *동작*):
  1. ★ DB md_status='success' → 응답 'completed' 단방향 매핑 (P0-3 silent-fallback 함정 가드의 backend 절반).
     - partial/pending/failed/skipped/None 은 그대로 통과 ('success' 만 매핑).
  2. [S1-ADD] 3필드(original_filename / duplicate_of / duplicate_count) 디코드 + 기본값(duplicate_count=0).
  3. (있으면) ds-app contract fixtures 가 응답 모델로 디코드 — 계약 shape 비파괴.

주의 — 이 테스트는 backend 직렬화 절반만 커버한다.
  앱이 'completed' 를 실제 md-first 렌더 분기로 태우는지(¬extracted_text)는 S3 fixture-render 테스트가 책임진다
  (A 그룹 close = 본 backend green AND S3 render green, owner 명기 — plan A-4).

실행 환경: app/ 의존성 설치된 컨텍스트(devsbx/GPU). 순수 단위(DB 불요).
"""

from __future__ import annotations

import json
import logging
import os
import sys
from datetime import datetime, timezone
from pathlib import Path

import pytest

# logs/ 가 운영 daemon(root) 소유일 때 import-time FileHandler PermissionError 방어 (test 한정).
_orig_file_handler = logging.FileHandler


def _safe_file_handler(filename, *args, **kwargs):  # type: ignore[no-untyped-def]
    try:
        return _orig_file_handler(filename, *args, **kwargs)
    except PermissionError:
        return logging.NullHandler()


logging.FileHandler = _safe_file_handler  # type: ignore[assignment]

# api.documents import 가 SQLAlchemy engine init 를 트리거 — dummy DATABASE_URL (실제 connect X).
os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://test:test@localhost:5432/test")

# tests/ → 프로젝트 루트 → app/
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "app"))

from api.documents import (  # noqa: E402
    DocumentDetailResponse,
    DocumentListResponse,
    DocumentResponse,
)

_NOW = datetime(2026, 6, 4, 8, 0, 0, tzinfo=timezone.utc)


def _base_detail(**overrides) -> dict:
    """DocumentDetailResponse 가 요구하는 전 필드(필수 포함) 완비 dict. overrides 로 일부 교체."""
    d = {
        "id": 4912,
        "file_path": "Engineering/ASME/x.pdf",
        "file_format": "pdf",
        "file_size": 1338920,
        "file_type": "document",
        "title": "x",
        "ai_domain": "Engineering",
        "ai_sub_group": "압력용기",
        "ai_tags": ["ASME"],
        "ai_summary": "요약",
        "document_type": "standard",
        "importance": "high",
        "ai_confidence": 0.9,
        "user_note": None,
        "user_tags": None,
        "pinned": True,
        "ask_includable": True,
        "derived_path": None,
        "original_format": "pdf",
        "conversion_status": "completed",
        "is_read": True,
        "review_status": "approved",
        "edit_url": None,
        "preview_status": "ready",
        "source_channel": "upload",
        "data_origin": "external",
        "doc_purpose": "reference",
        "extracted_at": _NOW,
        "ai_processed_at": _NOW,
        "embedded_at": _NOW,
        "created_at": _NOW,
        "updated_at": _NOW,
        # detail 전용
        "extracted_text": "원문 폴백 텍스트",
        "md_content": "# 제목\n본문",
        "md_frontmatter": {},
        "md_status": "success",
        "md_extraction_engine": "marker",
        "md_generated_at": _NOW,
    }
    d.update(overrides)
    return d


# ── 1. ★ md_status 단방향 매핑 (success → completed) ──────────────────────────

def test_db_success_serializes_as_completed():
    m = DocumentDetailResponse.model_validate(_base_detail(md_status="success"))
    assert m.md_status == "completed", "DB 'success' 는 응답에서 'completed' 로 매핑돼야 함(MD-first 렌더 트리거)"
    # model_dump(직렬화) 까지 확인 — 앱이 받는 실제 값.
    assert m.model_dump()["md_status"] == "completed"


@pytest.mark.parametrize("raw", ["pending", "processing", "partial", "failed", "skipped", None])
def test_non_success_statuses_pass_through(raw):
    m = DocumentDetailResponse.model_validate(_base_detail(md_status=raw))
    assert m.md_status == raw, f"'{raw}' 는 매핑 대상 아님 — 그대로 통과해야 함"


def test_mapping_is_read_only_not_a_write_path():
    # 이 모델은 응답 직렬화 전용 — write(ORM) 경로가 'completed' 를 DB 로 되쓰지 않는지의 1차 방어선.
    # 'completed' 입력이 들어와도(예: fixture) 그대로 'completed' (재매핑 없음, 멱등).
    m = DocumentDetailResponse.model_validate(_base_detail(md_status="completed"))
    assert m.md_status == "completed"


# ── 2. [S1-ADD] 3필드 디코드 + 기본값 ────────────────────────────────────────

def test_s1add_fields_default_on_list_response():
    # DocumentResponse(리스트 행)에도 3필드 존재 — 미제공 시 기본값.
    base = {k: v for k, v in _base_detail().items()
            if k not in {"extracted_text", "md_content", "md_frontmatter", "md_status",
                         "md_extraction_engine", "md_generated_at"}}
    m = DocumentResponse.model_validate(base)
    assert m.duplicate_count == 0
    assert m.duplicate_of is None
    assert m.original_filename is None


def test_s1add_fields_roundtrip_values():
    m = DocumentDetailResponse.model_validate(
        _base_detail(original_filename="보고서.docx", duplicate_of=4912, duplicate_count=2)
    )
    assert m.original_filename == "보고서.docx"
    assert m.duplicate_of == 4912
    assert m.duplicate_count == 2


# ── 3. ds-app contract fixtures 디코드 (있으면) ──────────────────────────────

_FIXDIR = Path(os.path.expanduser("~/Documents/code/ds-app/contract/fixtures"))


@pytest.mark.skipif(not _FIXDIR.exists(), reason="ds-app contract fixtures 미존재(독립 repo) — 디코드 회귀 skip")
@pytest.mark.parametrize("fname", ["document_detail.json", "document_detail_pending_md.json"])
def test_contract_detail_fixture_decodes(fname):
    payload = json.loads((_FIXDIR / fname).read_text())
    m = DocumentDetailResponse.model_validate(payload)
    # fixture 의 md_status 는 이미 API 어휘('completed'/'pending') — 매핑 멱등.
    assert m.md_status == payload["md_status"]
    # [S1-ADD] 필드가 fixture 에 있으면 디코드 일치.
    if "duplicate_count" in payload:
        assert m.duplicate_count == payload["duplicate_count"]


@pytest.mark.skipif(not _FIXDIR.exists(), reason="ds-app contract fixtures 미존재")
def test_contract_list_fixture_decodes():
    payload = json.loads((_FIXDIR / "documents_list.json").read_text())
    m = DocumentListResponse.model_validate(payload)
    assert m.total == payload["total"]
    assert len(m.items) == len(payload["items"])