DataOps-platform/tests/data_research/test_quality_issue_api.py

from __future__ import annotations

from datetime import UTC, datetime, timedelta

import pytest

ISSUE_UID = "01900000-0000-7000-8000-000000000901"
VIOLATION_UID = "01900000-0000-7000-8000-000000000902"
RUN_UID = "01900000-0000-7000-8000-000000000903"
ASSET_UID = "01900000-0000-7000-8000-000000000904"
ASSIGNEE_UID = "01900000-0000-7000-8000-000000000905"


class FakeQualityIssueService:
    def __init__(self):
        from app.core.data_research.quality_issues import (
            QualityIssueImportResult,
            QualityIssueRecord,
            QualityIssueRemediationRecord,
            QualityIssueTimelineRecord,
        )

        now = datetime(2026, 7, 29, 10, tzinfo=UTC)
        self.issue = QualityIssueRecord(
            uid=ISSUE_UID,
            issue_code="QI-000000000901",
            source_violation_uid=VIOLATION_UID,
            source_run_uid=RUN_UID,
            rule_code="asset_context_complete",
            severity="error",
            priority="high",
            asset_uid=ASSET_UID,
            field_name="location",
            source_uid=None,
            source_mapping_uid=None,
            message="设备位置缺失",
            evidence={
                "asset_uid": ASSET_UID,
                "missing_fields": ["location"],
            },
            recurrence_key="a" * 64,
            occurrence_number=2,
            status="assigned",
            assignee_uid=ASSIGNEE_UID,
            due_at=now + timedelta(days=2),
            current_version=2,
            created_by="editor-1",
            updated_by="editor-1",
            created_at=now,
            updated_at=now,
        )
        self.remediation = QualityIssueRemediationRecord(
            uid="01900000-0000-7000-8000-000000000906",
            issue_uid=ISSUE_UID,
            round_number=1,
            summary="已完成补录",
            evidence_refs=(),
            submitted_by="editor-1",
            submitted_at=now,
        )
        self.event = QualityIssueTimelineRecord(
            uid="01900000-0000-7000-8000-000000000907",
            issue_uid=ISSUE_UID,
            action="assigned",
            from_status="open",
            to_status="assigned",
            actor_uid="editor-1",
            note=None,
            payload={"assignee_uid": ASSIGNEE_UID},
            created_at=now,
        )
        self.import_result = QualityIssueImportResult(
            records=(self.issue,),
            created_count=1,
            existing_count=0,
        )
        self.actions = []

    def issues(self, **filters):
        self.actions.append(("issues", filters))
        return [self.issue], 1

    def statistics(self):
        self.actions.append(("statistics",))
        return {
            "total": 2,
            "open": 1,
            "closed": 1,
            "overdue": 0,
            "recurrent": 1,
        }

    def assignees(self):
        self.actions.append(("assignees",))
        return [
            {
                "uid": ASSIGNEE_UID,
                "username": "assignee",
                "display_name": "整改负责人",
            }
        ]

    def import_violations(self, **payload):
        self.actions.append(("import", payload))
        return self.import_result

    def get(self, issue_uid):
        self.actions.append(("get", issue_uid))
        return self.issue, self.remediation

    def timeline(self, issue_uid):
        self.actions.append(("timeline", issue_uid))
        return [self.event]

    def assign(self, issue_uid, **payload):
        self.actions.append(("assign", issue_uid, payload))
        return self.issue

    def start(self, issue_uid, **payload):
        self.actions.append(("start", issue_uid, payload))
        return self.issue

    def submit(self, issue_uid, **payload):
        self.actions.append(("submit", issue_uid, payload))
        return self.issue

    def review(self, issue_uid, **payload):
        self.actions.append(("review", issue_uid, payload))
        return self.issue

    def reopen(self, issue_uid, **payload):
        self.actions.append(("reopen", issue_uid, payload))
        return self.issue


@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

    service = FakeQualityIssueService()

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

    monkeypatch.setattr(permissions, "authenticate_request", identity)
    monkeypatch.setattr(
        routes,
        "get_quality_issue_service",
        lambda: service,
        raising=False,
    )
    app = create_app()
    app.config.update(TESTING=True)
    return app.test_client(), service


def test_viewer_reads_issue_register_detail_timeline_and_statistics(client):
    http, service = client
    headers = {"Authorization": "Bearer viewer"}

    listing = http.get(
        "/api/development/v1/device-quality/issues"
        "?status=assigned&overdue_only=true&page=1&page_size=20",
        headers=headers,
    )
    detail = http.get(
        f"/api/development/v1/device-quality/issues/{ISSUE_UID}",
        headers=headers,
    )
    timeline = http.get(
        f"/api/development/v1/device-quality/issues/{ISSUE_UID}/timeline",
        headers=headers,
    )
    statistics = http.get(
        "/api/development/v1/device-quality/issues/statistics",
        headers=headers,
    )
    assignees = http.get(
        "/api/development/v1/device-quality/issues/assignees",
        headers=headers,
    )
    denied = http.post(
        "/api/development/v1/device-quality/issues/import",
        headers=headers,
        json={"violation_uids": [VIOLATION_UID], "priority": "high"},
    )

    assert listing.status_code == 200
    assert listing.get_json()["data"]["records"][0]["occurrence_number"] == 2
    assert detail.get_json()["data"]["latest_remediation"]["round_number"] == 1
    assert timeline.get_json()["data"]["records"][0]["action"] == "assigned"
    assert statistics.get_json()["data"]["recurrent"] == 1
    assert assignees.get_json()["data"]["records"][0]["display_name"] == (
        "整改负责人"
    )
    assert denied.status_code == 403
    assert service.actions[0][0] == "issues"


def test_editor_imports_assigns_starts_submits_and_reopens_but_not_reviews(
    client,
):
    http, service = client
    headers = {"Authorization": "Bearer editor"}
    base = f"/api/development/v1/device-quality/issues/{ISSUE_UID}"

    imported = http.post(
        "/api/development/v1/device-quality/issues/import",
        headers=headers,
        json={
            "violation_uids": [VIOLATION_UID],
            "priority": "high",
            "due_at": "2026-08-01T12:00:00+08:00",
        },
    )
    assigned = http.post(
        f"{base}/assign",
        headers=headers,
        json={
            "assignee_uid": ASSIGNEE_UID,
            "due_at": None,
            "expected_version": 1,
        },
    )
    started = http.post(
        f"{base}/start",
        headers=headers,
        json={"expected_version": 2},
    )
    submitted = http.post(
        f"{base}/submit",
        headers=headers,
        json={
            "summary": "已补录",
            "evidence_refs": [],
            "expected_version": 3,
        },
    )
    reopened = http.post(
        f"{base}/reopen",
        headers=headers,
        json={
            "reason": "现场复发",
            "due_at": None,
            "expected_version": 5,
        },
    )
    denied = http.post(
        f"{base}/review",
        headers=headers,
        json={
            "verification_result": "passed",
            "note": "复验通过",
            "expected_version": 4,
        },
    )

    assert imported.status_code == 201
    assert assigned.status_code == 200
    assert started.status_code == 200
    assert submitted.status_code == 200
    assert reopened.status_code == 200
    assert denied.status_code == 403
    assert (service.actions[0][0],) == ("import",)


def test_admin_reviews_with_verification_evidence(client):
    http, service = client
    response = http.post(
        f"/api/development/v1/device-quality/issues/{ISSUE_UID}/review",
        headers={"Authorization": "Bearer admin"},
        json={
            "verification_result": "passed",
            "note": "人工核对与复验通过",
            "verification_run_uid": RUN_UID,
            "expected_version": 4,
        },
    )

    assert response.status_code == 200
    assert service.actions[-1][0] == "review"
    assert service.actions[-1][2]["verification_run_uid"] == RUN_UID
    assert service.actions[-1][2]["actor_uid"] == "admin-1"