leaudit-platform-backend/tests/release/test_g4_documents.py

from __future__ import annotations

import time

import fitz
import pytest

from .conftest import TenantSeed
from .helpers import ReleaseApiClient


def _document_ids(items: list[dict]) -> set[int]:
    return {int(item["documentId"]) for item in items if item.get("documentId") is not None}


def _sample_pdf_bytes(text: str) -> bytes:
    document = fitz.open()
    page = document.new_page()
    page.insert_text((72, 72), text)
    return document.tobytes()


@pytest.mark.release
def test_g4_document_list_and_detail_respect_tenant_boundary(
    tenant_admin_api: ReleaseApiClient,
    tenant_admin_api_b: ReleaseApiClient,
    tenant_a: TenantSeed,
    tenant_b: TenantSeed,
    make_release_document,
) -> None:
    own_doc = make_release_document(
        client=tenant_admin_api,
        tenant=tenant_a,
        file_name=f"pytest-g4-a-{int(time.time())}.txt",
    )
    other_doc = make_release_document(
        client=tenant_admin_api_b,
        tenant=tenant_b,
        file_name=f"pytest-g4-b-{int(time.time())}.txt",
    )

    own_list = ReleaseApiClient.json_data(tenant_admin_api.get("/api/documents/list?page=1&pageSize=100"))
    own_ids = _document_ids(own_list["documents"])
    assert own_doc.document_id in own_ids
    assert other_doc.document_id not in own_ids

    own_detail = ReleaseApiClient.json_data(tenant_admin_api.get(f"/api/documents/{own_doc.document_id}"))
    assert int(own_detail["documentId"]) == own_doc.document_id
    assert str(own_detail.get("tenantCode") or "") == tenant_a.tenant_code

    cross_detail = tenant_admin_api.get(f"/api/documents/{other_doc.document_id}", expected_status=404)
    assert "无权访问" in cross_detail.text or "不存在" in cross_detail.text


@pytest.mark.release
def test_g4_document_update_append_and_delete_reject_cross_tenant_access(
    tenant_admin_api: ReleaseApiClient,
    tenant_admin_api_b: ReleaseApiClient,
    tenant_a: TenantSeed,
    make_release_document,
) -> None:
    own_doc = make_release_document(
        client=tenant_admin_api,
        tenant=tenant_a,
        file_name=f"pytest-g4-own-{int(time.time())}.pdf",
        content=_sample_pdf_bytes("pytest g4 own document"),
        content_type="application/pdf",
    )

    update_response = tenant_admin_api.put(
        f"/api/documents/{own_doc.document_id}",
        json={"remark": "pytest g4 own update"},
        expected_status=200,
    )
    update_data = ReleaseApiClient.json_data(update_response)
    assert int(update_data["documentId"]) == own_doc.document_id

    append_response = tenant_admin_api.post(
        f"/api/documents/{own_doc.document_id}/attachments",
        data={"mergeMode": "new", "remark": "pytest attachment"},
        files=[("files", ("attachment.pdf", _sample_pdf_bytes("pytest attachment"), "application/pdf"))],
        expected_status=200,
    )
    append_data = ReleaseApiClient.json_data(append_response)
    new_document_id = int(append_data["documentId"])
    assert new_document_id != own_doc.document_id
    assert str(append_data.get("tenantCode") or "") == tenant_a.tenant_code
    assert int(append_data.get("previousVersionId") or 0) == own_doc.document_id

    cross_update = tenant_admin_api_b.put(
        f"/api/documents/{new_document_id}",
        json={"remark": "should fail"},
        expected_status=404,
    )
    assert "无权访问" in cross_update.text or "不存在" in cross_update.text

    cross_append = tenant_admin_api_b.post(
        f"/api/documents/{new_document_id}/attachments",
        data={"mergeMode": "new", "remark": "should fail"},
        files=[("files", ("cross.pdf", _sample_pdf_bytes("cross tenant"), "application/pdf"))],
        expected_status=404,
    )
    assert "无权访问" in cross_append.text or "不存在" in cross_append.text

    cross_delete = tenant_admin_api_b.delete(f"/api/documents/{new_document_id}", expected_status=404)
    assert "无权访问" in cross_delete.text or "不存在" in cross_delete.text


@pytest.mark.release
def test_g4_common_user_only_sees_self_created_documents(
    common_api_a: ReleaseApiClient,
    tenant_admin_api: ReleaseApiClient,
    tenant_a: TenantSeed,
    make_release_document,
) -> None:
    user_doc = make_release_document(
        client=common_api_a,
        tenant=tenant_a,
        file_name=f"pytest-g4-common-self-{int(time.time())}.txt",
    )
    admin_doc = make_release_document(
        client=tenant_admin_api,
        tenant=tenant_a,
        file_name=f"pytest-g4-common-other-{int(time.time())}.txt",
    )

    own_list = ReleaseApiClient.json_data(common_api_a.get("/api/documents/list?page=1&pageSize=100"))
    own_ids = _document_ids(own_list["documents"])
    assert user_doc.document_id in own_ids
    assert admin_doc.document_id not in own_ids

    own_detail = ReleaseApiClient.json_data(common_api_a.get(f"/api/documents/{user_doc.document_id}"))
    assert int(own_detail["documentId"]) == user_doc.document_id

    hidden_detail = common_api_a.get(f"/api/documents/{admin_doc.document_id}", expected_status=404)
    assert "无权访问" in hidden_detail.text or "不存在" in hidden_detail.text


@pytest.mark.release
def test_g4_govdoc_list_endpoint_does_not_fail_when_backfilling_version_groups(
    tenant_admin_api: ReleaseApiClient,
) -> None:
    response = tenant_admin_api.get("/api/govdoc/documents?page=1&pageSize=10", expected_status=200)
    payload = response.json()
    assert payload.get("code") in {0, 200}, payload
    assert "data" in payload, payload