DataOps-platform/tests/data_research/test_evidence_api.py

from __future__ import annotations

import io

import pytest


class EvidenceService:
    def preview(self, uid):
        assert uid == "evidence-1"
        return {
            "uid": uid,
            "locator": {"kind": "pdf.page", "page": 2},
            "excerpt": "客户编号 13800138000，token=secret-value",
            "confidence": 0.88,
        }

    def download(self, uid):
        assert uid == "evidence-1"
        return b"original-secret-source", "evidence.pdf", "application/pdf"


class ArtifactService:
    def __init__(self):
        self.calls = []

    def store(self, source_uid, filename, media_type, content, parser_version):
        self.calls.append((source_uid, filename, media_type, content, parser_version))
        record = type(
            "Artifact",
            (),
            {
                "uid": "artifact-1",
                "source_uid": source_uid,
                "filename": filename,
                "media_type": media_type,
                "size_bytes": len(content),
                "content_hash": "a" * 64,
                "parser_version": parser_version,
            },
        )()
        return record, True


@pytest.fixture()
def client(monkeypatch):
    from flask import request

    from app import create_app
    from app.api.data_development import routes
    from app.core.system import permissions

    artifacts = ArtifactService()

    def identity():
        token = request.headers.get("Authorization", "")
        if token == "Bearer viewer":
            return {"id": "viewer-1", "roles": ["viewer"]}
        if token == "Bearer editor":
            return {"id": "editor-1", "roles": ["editor"]}
        if token == "Bearer admin":
            return {"id": "admin-1", "roles": ["admin"]}
        return None

    monkeypatch.setattr(permissions, "authenticate_request", identity)
    monkeypatch.setattr(routes, "get_evidence_service", lambda: EvidenceService())
    monkeypatch.setattr(routes, "get_artifact_service", lambda: artifacts)
    app = create_app()
    app.config.update(TESTING=True)
    return app.test_client(), artifacts


def test_preview_is_redacted_and_original_download_is_admin_only(client):
    http, _artifacts = client
    preview = http.get(
        "/api/development/v1/evidence/evidence-1",
        headers={"Authorization": "Bearer viewer"},
    )
    denied = http.get(
        "/api/development/v1/evidence/evidence-1?download=1",
        headers={"Authorization": "Bearer viewer"},
    )
    downloaded = http.get(
        "/api/development/v1/evidence/evidence-1?download=1",
        headers={"Authorization": "Bearer admin"},
    )

    assert preview.status_code == 200
    text = preview.get_data(as_text=True)
    assert "13800138000" not in text and "secret-value" not in text
    assert "138****8000" in text and "[redacted]" in text
    assert denied.status_code == 403
    assert downloaded.data == b"original-secret-source"


def test_editor_uploads_allowed_file_without_returning_storage_credentials(client):
    http, artifacts = client
    response = http.post(
        "/api/development/v1/sources/files",
        headers={"Authorization": "Bearer editor"},
        data={
            "source_uid": "source-1",
            "parser_version": "csv-v1",
            "file": (io.BytesIO(b"id,name\n1,A"), "governance.csv", "text/csv"),
        },
        content_type="multipart/form-data",
    )

    assert response.status_code == 201
    assert response.get_json()["data"]["uid"] == "artifact-1"
    assert "storage_ref" not in response.get_data(as_text=True)
    assert artifacts.calls[0][3] == b"id,name\n1,A"