group_fqcd_jr/tests/unit/service/test_knowledge_reference.py

"""K001 知识引用解析（B1）单元测试：签名、过期、跨用户、发布状态、脱敏。"""

import json
from datetime import UTC, date, datetime, timedelta
from typing import Any

import pytest

from app.core.contracts import RequestContext
from app.core.errors import ForbiddenAgentError, RecoverableAgentError, ResourceNotFoundError
from app.service.authorization_service import AuthorizationService
from app.service.knowledge_service import (
    SIGNING_SECRET_ENV,
    KnowledgeReferenceService,
    build_reference_token,
)

SECRET = "unit-test-signing-secret"
PERMISSION = "knowledge:reference:read"


class FakeResult:
    """最小 SQLAlchemy Result 替身，只提供 mappings().first()。"""

    def __init__(self, row: dict[str, Any] | None) -> None:
        self._row = row

    def mappings(self) -> "FakeResult":
        return self

    def first(self) -> dict[str, Any] | None:
        return self._row


class FakeSession:
    def __init__(self, row: dict[str, Any] | None) -> None:
        self._row = row

    async def __aenter__(self) -> "FakeSession":
        return self

    async def __aexit__(self, *args: object) -> None:
        return None

    async def execute(self, *args: object, **kwargs: object) -> FakeResult:
        return FakeResult(self._row)


def metadata_row(**overrides: Any) -> dict[str, Any]:
    row: dict[str, Any] = {
        "id": 42,
        "knowledge_type": "faq",
        "title": "场内基金申购费率说明",
        "version": "v2",
        "milvus_collection": "fin_faq_collection",
        "tags": json.dumps(["费率", "申购"]),
        "effective_date": date(2026, 1, 1),
        "expire_date": date(2030, 1, 1),
        "review_status": "published",
        "status": "active",
        "content_text": "内部正文：不得通过引用接口返回的整份文档内容。",
    }
    row.update(overrides)
    return row


def service_with_row(row: dict[str, Any] | None) -> KnowledgeReferenceService:
    return KnowledgeReferenceService(session_factory=lambda: FakeSession(row))


def context(**overrides: Any) -> RequestContext:
    values: dict[str, Any] = {
        "user_id": "7",
        "trace_id": "trace-knowledge",
        "permissions": (PERMISSION,),
    }
    values.update(overrides)
    return RequestContext(**values)


def token_for(user_id: str = "7", knowledge_id: int = 42, **kwargs: Any) -> str:
    expires_at = kwargs.pop("expires_at", datetime.now(UTC) + timedelta(minutes=30))
    return build_reference_token(
        user_id=user_id, knowledge_id=knowledge_id, expires_at=expires_at, secret=SECRET, **kwargs
    )


@pytest.fixture(autouse=True)
def signing_secret(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setenv(SIGNING_SECRET_ENV, SECRET)


async def test_valid_token_resolves_masked_metadata() -> None:
    result = await service_with_row(metadata_row()).resolve(context(), token_for())
    assert result["knowledge_id"] == 42
    assert result["title"] == "场内基金申购费率说明"
    assert result["collection"] == "fin_faq_collection"
    assert result["tags"] == ["费率", "申购"]
    assert result["effective_date"] == "2026-01-01"
    assert result["expire_date"] == "2030-01-01"
    assert result["content_included"] is False
    assert "content_text" not in result


async def test_full_document_text_is_never_returned() -> None:
    result = await service_with_row(metadata_row()).resolve(context(), token_for())
    assert "内部正文" not in json.dumps(result, ensure_ascii=False)


@pytest.mark.parametrize(
    "token_kind",
    ["wrong_signature", "expired", "cross_user", "malformed", "not_a_number"],
)
async def test_invalid_tokens_are_rejected(token_kind: str) -> None:
    valid = token_for()
    tokens = {
        "wrong_signature": valid[:-1] + ("0" if valid[-1] != "0" else "1"),
        "expired": token_for(expires_at=datetime.now(UTC) - timedelta(seconds=1)),
        "cross_user": token_for(user_id="8"),
        "malformed": "kr1.7.42.abc.zzzz",
        "not_a_number": "kr1.7.not-a-number.9999999999." + "a" * 64,
    }
    with pytest.raises(ResourceNotFoundError):
        await service_with_row(metadata_row()).resolve(context(), tokens[token_kind])


async def test_token_without_prefix_is_rejected() -> None:
    token = token_for().replace("kr1.", "kr2.", 1)
    with pytest.raises(ResourceNotFoundError):
        await service_with_row(metadata_row()).resolve(context(), token)


@pytest.mark.parametrize(
    "overrides",
    [
        {"review_status": "pending"},
        {"review_status": "approved"},
        {"review_status": "archived"},
        {"status": "inactive"},
        {"status": "deleted"},
        {"expire_date": date(2020, 1, 1)},
        {"effective_date": date(2099, 1, 1)},
    ],
)
async def test_unpublished_or_invalid_knowledge_is_rejected(overrides: dict[str, Any]) -> None:
    service = service_with_row(metadata_row(**overrides))
    with pytest.raises(ResourceNotFoundError):
        await service.resolve(context(), token_for())


async def test_unknown_knowledge_id_is_rejected() -> None:
    with pytest.raises(ResourceNotFoundError) as excinfo:
        await service_with_row(None).resolve(context(), token_for())
    # 文档 §3.6 未给通用 404 单独定义码，实现统一复用 SESSION_NOT_FOUND。
    assert excinfo.value.code == "SESSION_NOT_FOUND"
    assert excinfo.value.status_code == 404


async def test_knowledge_without_expire_date_is_allowed() -> None:
    service = service_with_row(metadata_row(expire_date=None, version=None, tags=None))
    result = await service.resolve(context(), token_for())
    assert result["expire_date"] is None
    assert result["version"] is None
    assert result["tags"] == []


async def test_missing_signing_secret_fails_closed(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.delenv(SIGNING_SECRET_ENV, raising=False)
    with pytest.raises(RecoverableAgentError, match=SIGNING_SECRET_ENV):
        await service_with_row(metadata_row()).resolve(context(), token_for())


async def test_permission_is_checked_before_token_validation(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    async def deny(_context: RequestContext, _permission: str, *, admin: bool = False) -> None:
        raise ForbiddenAgentError("缺少操作权限")

    monkeypatch.setattr(AuthorizationService, "require", deny)
    with pytest.raises(ForbiddenAgentError):
        await service_with_row(metadata_row()).resolve(context(), token_for())