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

from __future__ import annotations

from pathlib import Path
import re

import pytest
import yaml

from dms.storage.paths import (
    STORAGE_SUBDIRECTORIES,
    UnsafeStoragePathError,
    ensure_storage_directories,
    resolve_storage_path,
)


BACKEND_ROOT = Path(__file__).resolve().parents[2]
OPENAPI_ROOT = BACKEND_ROOT / "openapi"


def test_all_openapi_yaml_files_parse():
    for path in OPENAPI_ROOT.rglob("*.yaml"):
        with path.open("r", encoding="utf-8") as stream:
            assert isinstance(yaml.safe_load(stream), dict)


def test_all_openapi_external_refs_resolve():
    def walk(value, source_path: Path):
        if isinstance(value, dict):
            reference = value.get("$ref")
            if isinstance(reference, str) and not reference.startswith("#"):
                target, _, fragment = reference.partition("#")
                target_path = (source_path.parent / target).resolve()
                assert target_path.is_file(), reference
                target_document = yaml.safe_load(
                    target_path.read_text(encoding="utf-8")
                )
                current = target_document
                if fragment:
                    for part in fragment.lstrip("/").split("/"):
                        assert part in current, reference
                        current = current[part]
            for child in value.values():
                walk(child, source_path)
        elif isinstance(value, list):
            for child in value:
                walk(child, source_path)

    for path in OPENAPI_ROOT.rglob("*.yaml"):
        walk(yaml.safe_load(path.read_text(encoding="utf-8")), path)


def test_flask_methods_match_openapi_exactly(flask_app):
    document = yaml.safe_load(
        (OPENAPI_ROOT / "openapi.yaml").read_text(encoding="utf-8")
    )
    openapi_methods = set()
    operation_ids = []
    for path, item in document["paths"].items():
        if "$ref" in item:
            target, _, fragment = item["$ref"].partition("#")
            referenced = yaml.safe_load(
                (OPENAPI_ROOT / target).read_text(encoding="utf-8")
            )
            item = referenced[fragment.lstrip("/")]
        for method, operation in item.items():
            if method.lower() in {"get", "post", "put", "delete", "patch"}:
                openapi_methods.add((method.upper(), path))
                operation_ids.append(operation["operationId"])
    flask_methods = set()
    for rule in flask_app.url_map.iter_rules():
        if not rule.rule.startswith("/api/v1"):
            continue
        path = re.sub(r"<(?:[^:>]+:)?([^>]+)>", r"{\1}", rule.rule)
        for method in rule.methods - {"HEAD", "OPTIONS"}:
            flask_methods.add((method, path))
    assert flask_methods == openapi_methods
    assert len(operation_ids) == len(set(operation_ids))


def test_openapi_contains_only_q2b_paths():
    document = yaml.safe_load(
        (OPENAPI_ROOT / "openapi.yaml").read_text(encoding="utf-8")
    )
    assert set(document["paths"]) == {
        "/api/v1/health",
        "/api/v1/auth/login",
        "/api/v1/auth/logout",
        "/api/v1/users/me",
        "/api/v1/organizations/tree",
        "/api/v1/users",
        "/api/v1/config/ui-dictionaries",
        "/api/v1/categories/tree",
        "/api/v1/categories",
        "/api/v1/categories/{id}",
        "/api/v1/documents",
        "/api/v1/documents/{id}",
        "/api/v1/documents/batch-import",
        "/api/v1/documents/{id}/preview",
        "/api/v1/documents/{id}/download",
        "/api/v1/main-plans/{id}/sub-plans",
        "/api/v1/attachments",
        "/api/v1/attachments/{id}",
        "/api/v1/attachments/batch-import",
        "/api/v1/attachments/{id}/main-plans",
        "/api/v1/main-plans/{id}/attachments",
        "/api/v1/main-plans/{id}/attachments/bind",
        "/api/v1/main-plans/{id}/attachments/{attachmentId}",
        "/api/v1/documents/{id}/permissions",
        "/api/v1/audit/logs",
        "/api/v1/statistics/documents",
        "/api/v1/audit/statistics/trend",
        "/api/v1/audit/statistics/actions",
        "/api/v1/audit/statistics/users",
        "/api/v1/recycle-bin/documents",
        "/api/v1/recycle-bin/documents/{id}/restore",
    }
    assert document["info"]["version"] == "1.5.0"
    assert "/api/v1/documents/{id}/restore" not in document["paths"]
    assert document["paths"]["/api/v1/health"]["get"]["security"] == []
    login_path = yaml.safe_load(
        (OPENAPI_ROOT / "paths" / "auth.yaml").read_text(encoding="utf-8")
    )
    assert login_path["Login"]["post"]["security"] == []


def test_openapi_string_id_component():
    schemas = yaml.safe_load(
        (OPENAPI_ROOT / "components" / "schemas.yaml").read_text(encoding="utf-8")
    )
    assert schemas["StringId"] == {
        "type": "string",
        "pattern": "^[0-9]+$",
        "description": "数据库BIGINT在JSON中的安全十进制字符串表达。",
    }


def test_openapi_health_response_matches_contract():
    schemas = yaml.safe_load(
        (OPENAPI_ROOT / "components" / "schemas.yaml").read_text(encoding="utf-8")
    )
    health = schemas["HealthResponse"]
    assert set(health["required"]) == {"code", "message", "data", "requestId"}
    assert "timestamp" not in health["properties"]


def test_openapi_has_all_fixed_enum_groups():
    schemas = yaml.safe_load(
        (OPENAPI_ROOT / "components" / "schemas.yaml").read_text(encoding="utf-8")
    )
    expected = {
        "RoleCode",
        "AllowedModule",
        "DocumentType",
        "DocumentStatus",
        "SecurityLevel",
        "VisibilityType",
        "AttachmentType",
        "SubjectType",
        "AllowedAction",
        "CategoryType",
        "EnabledStatus",
        "AuditResult",
        "AuditAction",
        "AuditTarget",
    }
    assert expected <= set(schemas)


def test_storage_directory_skeleton(tmp_path: Path):
    root = ensure_storage_directories(tmp_path / "storage")
    assert {path.name for path in root.iterdir()} == set(STORAGE_SUBDIRECTORIES)


@pytest.mark.parametrize(
    "unsafe_path",
    [
        "../outside.pdf",
        "original/../../outside.pdf",
        "/absolute/file.pdf",
        "C:\\Windows\\system.ini",
        "",
    ],
)
def test_storage_path_cannot_escape_root(tmp_path: Path, unsafe_path: str):
    with pytest.raises(UnsafeStoragePathError):
        resolve_storage_path(unsafe_path, tmp_path)


def test_storage_path_resolves_safe_relative_path(tmp_path: Path):
    resolved = resolve_storage_path(
        "original/2026/07/1/file.pdf",
        tmp_path,
    )
    assert resolved == (tmp_path / "original/2026/07/1/file.pdf").resolve()