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"