document-management/backend/tests/dms/test_q2b_updated_range.py

from __future__ import annotations

from datetime import datetime
from urllib.parse import urlencode

import pytest
import yaml
from sqlalchemy import event, func, select

from dms.common.errors import InvalidArgumentError
from dms.common.time_range import parse_datetime, parse_updated_range


def auth(token: str) -> dict[str, str]:
    return {"Authorization": f"Bearer {token}"}


def payload(response):
    return response.get_json()["data"]


TIMES = {
    "B4_TEST_草稿主案": datetime(2026, 1, 5, 0, 0),
    "B4_TEST_绝密主案": datetime(2026, 1, 6, 0, 0),
    "B4_TEST_无授权主案": datetime(2026, 1, 7, 0, 0),
    "B4_TEST_全部主案": datetime(2026, 1, 10, 0, 0),
    "B4_TEST_规范附件": datetime(2026, 1, 15, 0, 0),
    "B4_TEST_组织主案": datetime(2026, 1, 20, 0, 0),
    "B4_TEST_动态继承子案": datetime(2026, 1, 25, 0, 0),
    "B4_TEST_自定义主案": datetime(2026, 1, 30, 0, 0),
    "B4_TEST_表格附件": datetime(2026, 2, 5, 0, 0),
}


@pytest.fixture()
def q2b_app(b4_app):
    from dms.extensions import db
    from dms.models import AttachmentBinding, Document

    with b4_app.app_context():
        documents = db.session.scalars(
            select(Document).where(Document.document_name.in_(TIMES))
        ).all()
        assert len(documents) == len(TIMES)
        for document in documents:
            document.updated_at = TIMES[document.document_name]
        # 故意让挂载关系时间全部晚于筛选范围，证明筛选依据是附件文档。
        for binding in db.session.scalars(select(AttachmentBinding)).all():
            binding.updated_at = datetime(2035, 1, 1, 0, 0)
        db.session.commit()
    return b4_app


@pytest.fixture()
def q2b_client(q2b_app):
    return q2b_app.test_client()


def query(client, path: str, token: str, **params):
    return client.get(
        f"{path}?{urlencode(params)}",
        headers=auth(token),
    )


def names(response) -> list[str]:
    return [item["documentName"] for item in payload(response)["items"]]


def test_common_parser_requires_timezone_and_normalizes_to_utc():
    assert parse_datetime(
        "2026-01-20T08:00:00+08:00", name="updatedFrom"
    ) == datetime(2026, 1, 20, 0, 0)
    assert parse_datetime(
        "2026-01-20T00:00:00Z", name="updatedFrom"
    ) == datetime(2026, 1, 20, 0, 0)
    with pytest.raises(InvalidArgumentError):
        parse_datetime("2026-01-20T00:00:00", name="updatedFrom")
    with pytest.raises(InvalidArgumentError):
        parse_updated_range(
            {
                "updatedFrom": "2026-01-21T00:00:00Z",
                "updatedTo": "2026-01-20T00:00:00Z",
            }
        )


def test_documents_updated_range_single_and_closed_boundaries(
    q2b_client, token_for
):
    token = token_for("admin")
    start = query(
        q2b_client,
        "/api/v1/documents",
        token,
        documentType="MAIN",
        updatedFrom="2026-01-30T00:00:00Z",
        pageSize=100,
    )
    end = query(
        q2b_client,
        "/api/v1/documents",
        token,
        documentType="MAIN",
        updatedTo="2026-01-10T00:00:00Z",
        pageSize=100,
    )
    closed = query(
        q2b_client,
        "/api/v1/documents",
        token,
        documentType="MAIN",
        updatedFrom="2026-01-20T00:00:00Z",
        updatedTo="2026-01-30T00:00:00Z",
        sortBy="updatedAt",
        sortDirection="asc",
        pageSize=100,
    )

    assert names(start) == ["B4_TEST_自定义主案"]
    assert "B4_TEST_全部主案" in names(end)
    assert "B4_TEST_组织主案" not in names(end)
    assert names(closed) == ["B4_TEST_组织主案", "B4_TEST_自定义主案"]
    assert all(item["id"].isdecimal() for item in payload(closed)["items"])


def test_documents_keyword_offset_timezone_and_filter_before_pagination(
    q2b_client, token_for
):
    token = token_for("admin")
    combined = query(
        q2b_client,
        "/api/v1/documents",
        token,
        documentType="MAIN",
        keyword="组织",
        updatedFrom="2026-01-20T08:00:00+08:00",
        updatedTo="2026-01-20T08:00:00+08:00",
        pageSize=100,
    )
    second_page = query(
        q2b_client,
        "/api/v1/documents",
        token,
        documentType="MAIN",
        updatedFrom="2026-01-20T00:00:00Z",
        updatedTo="2026-01-30T00:00:00Z",
        sortBy="updatedAt",
        sortDirection="asc",
        page=2,
        pageSize=1,
    )

    assert names(combined) == ["B4_TEST_组织主案"]
    assert payload(second_page)["total"] == 2
    assert names(second_page) == ["B4_TEST_自定义主案"]


def test_direct_sub_plan_range_exact_boundary_and_status(
    q2b_app, q2b_client, token_for
):
    main_id = q2b_app.config["B4_IDS"]["main_custom"]
    response = query(
        q2b_client,
        f"/api/v1/main-plans/{main_id}/sub-plans",
        token_for("user"),
        keyword="动态",
        status="PUBLISHED",
        updatedFrom="2026-01-25T00:00:00Z",
        updatedTo="2026-01-25T00:00:00Z",
        pageSize=100,
    )
    outside = query(
        q2b_client,
        f"/api/v1/main-plans/{main_id}/sub-plans",
        token_for("admin"),
        status="PUBLISHED",
        updatedFrom="2026-01-25T00:00:00.001Z",
        pageSize=100,
    )

    assert response.status_code == 200
    assert names(response) == ["B4_TEST_动态继承子案"]
    assert names(outside) == []


def test_mounted_attachment_range_uses_attachment_not_binding_time(
    q2b_app, q2b_client, token_for
):
    main_id = q2b_app.config["B4_IDS"]["main_all"]
    first = query(
        q2b_client,
        f"/api/v1/main-plans/{main_id}/attachments",
        token_for("user"),
        keyword="规范",
        attachmentType="WORK_STANDARD",
        updatedFrom="2026-01-15T00:00:00Z",
        updatedTo="2026-01-15T00:00:00Z",
        pageSize=100,
    )
    second = query(
        q2b_client,
        f"/api/v1/main-plans/{main_id}/attachments",
        token_for("admin"),
        attachmentType="TABLE",
        updatedFrom="2026-02-01T00:00:00Z",
        pageSize=100,
    )
    excluded = query(
        q2b_client,
        f"/api/v1/main-plans/{main_id}/attachments",
        token_for("admin"),
        updatedTo="2026-01-14T23:59:59Z",
        pageSize=100,
    )

    assert names(first) == ["B4_TEST_规范附件"]
    assert names(second) == ["B4_TEST_表格附件"]
    assert names(excluded) == []
    assert payload(first)["items"][0]["bindingId"].isdecimal()


@pytest.mark.parametrize(
    "params",
    [
        {"updatedFrom": ""},
        {"updatedTo": "not-a-time"},
        {"updatedFrom": "2026-01-01T00:00:00"},
        {
            "updatedFrom": "2026-02-01T00:00:00Z",
            "updatedTo": "2026-01-01T00:00:00Z",
        },
    ],
)
@pytest.mark.parametrize("endpoint", ["documents", "sub-plans", "attachments"])
def test_invalid_ranges_return_uniform_error(
    q2b_app, q2b_client, token_for, endpoint, params
):
    ids = q2b_app.config["B4_IDS"]
    paths = {
        "documents": "/api/v1/documents",
        "sub-plans": f"/api/v1/main-plans/{ids['main_custom']}/sub-plans",
        "attachments": f"/api/v1/main-plans/{ids['main_all']}/attachments",
    }
    response = query(q2b_client, paths[endpoint], token_for("admin"), **params)

    assert response.status_code == 400
    body = response.get_json()
    assert body["code"] == "INVALID_ARGUMENT"
    assert body["requestId"] == response.headers["X-Request-Id"]


def test_role_boundaries_and_queries_do_not_increment_counts_or_audit(
    q2b_app, q2b_client, token_for
):
    from dms.extensions import db
    from dms.models import AuditLog, Document

    ids = q2b_app.config["B4_IDS"]
    tokens = {name: token_for(name) for name in ("admin", "user")}
    with q2b_app.app_context():
        before_counts = {
            item.id: (item.view_count, item.download_count)
            for item in db.session.scalars(select(Document)).all()
        }
        before_audits = db.session.scalar(select(func.count()).select_from(AuditLog))

    admin = query(
        q2b_client,
        "/api/v1/documents",
        tokens["admin"],
        updatedFrom="2026-01-01T00:00:00Z",
    )
    user = query(
        q2b_client,
        f"/api/v1/main-plans/{ids['main_custom']}/sub-plans",
        tokens["user"],
        updatedTo="2026-12-31T23:59:59Z",
    )
    assert admin.status_code == 200
    assert user.status_code == 200
    with q2b_app.app_context():
        after_counts = {
            item.id: (item.view_count, item.download_count)
            for item in db.session.scalars(select(Document)).all()
        }
        after_audits = db.session.scalar(select(func.count()).select_from(AuditLog))
    assert after_counts == before_counts
    assert after_audits == before_audits


def test_time_filters_are_present_in_database_queries(
    q2b_app, q2b_client, token_for
):
    from dms.extensions import db

    ids = q2b_app.config["B4_IDS"]
    statements: list[str] = []

    def record(_conn, _cursor, statement, _params, _context, _many):
        statements.append(statement.lower())

    with q2b_app.app_context():
        event.listen(db.engine, "before_cursor_execute", record)
        try:
            token = token_for("admin")
            for path in (
                "/api/v1/documents",
                f"/api/v1/main-plans/{ids['main_custom']}/sub-plans",
                f"/api/v1/main-plans/{ids['main_all']}/attachments",
            ):
                response = query(
                    q2b_client,
                    path,
                    token,
                    updatedFrom="2026-01-01T00:00:00Z",
                    updatedTo="2026-12-31T23:59:59Z",
                )
                assert response.status_code == 200
        finally:
            event.remove(db.engine, "before_cursor_execute", record)

    filtered = [
        statement
        for statement in statements
        if "doc_document.updated_at >=" in statement
        and "doc_document.updated_at <=" in statement
    ]
    assert len(filtered) >= 3
    mounted = next(
        statement for statement in filtered if "doc_attachment_binding" in statement
    )
    normalized_sql = " ".join(mounted.split())
    where_clause = normalized_sql.split(" where ", maxsplit=1)[1]
    assert "doc_attachment_binding.updated_at" not in where_clause


def test_openapi_declares_q2b_parameters_without_new_paths():
    from pathlib import Path

    root = Path(__file__).resolve().parents[2] / "openapi"
    document = yaml.safe_load((root / "openapi.yaml").read_text(encoding="utf-8"))
    paths = yaml.safe_load(
        (root / "paths" / "documents.yaml").read_text(encoding="utf-8")
    )

    assert document["info"]["version"] == "1.5.0"
    for operation in ("Documents", "SubPlans", "MainPlanAttachments"):
        references = {
            parameter.get("$ref")
            for parameter in paths[operation]["get"]["parameters"]
            if "$ref" in parameter
        }
        assert "../components/parameters.yaml#/UpdatedFrom" in references
        assert "../components/parameters.yaml#/UpdatedTo" in references
    assert "/api/v1/documents/search" not in document["paths"]