from __future__ import annotations from datetime import UTC, datetime, timedelta import pytest VERSION_UID = "01900000-0000-7000-8000-000000000801" RUN_UID = "01900000-0000-7000-8000-000000000802" ASSET_UID = "01900000-0000-7000-8000-000000000803" SOURCE_UID = "01900000-0000-7000-8000-000000000804" MAPPING_UID = "01900000-0000-7000-8000-000000000805" class FakeDeviceQualityService: def __init__(self): from app.core.data_research.device_quality import ( DEFAULT_DEVICE_QUALITY_RULES, DeviceQualityAssetScoreRecord, DeviceQualityPolicyVersionRecord, DeviceQualityRuleResultRecord, DeviceQualityRunRecord, DeviceQualityViolationRecord, validate_device_quality_rules, ) now = datetime(2026, 7, 29, 17, 0, tzinfo=UTC) self.version = DeviceQualityPolicyVersionRecord( uid=VERSION_UID, profile_uid="01900000-0000-7000-8000-000000000806", version=1, status="published", rules=validate_device_quality_rules( DEFAULT_DEVICE_QUALITY_RULES ), content_hash="a" * 64, created_by="editor-1", created_at=now, published_by="admin-1", published_at=now, ) self.run_record = DeviceQualityRunRecord( uid=RUN_UID, policy_version_uid=VERSION_UID, policy_hash="a" * 64, source_uid=SOURCE_UID, status="success", total_assets=1, total_violations=1, score=85.0, created_by="editor-1", created_at=now, ) self.rule_result = DeviceQualityRuleResultRecord( uid="01900000-0000-7000-8000-000000000807", run_uid=RUN_UID, rule_code="asset_context_complete", severity="error", weight=15.0, status="violated", evaluated_count=1, violation_count=1, sampled_count=1, pass_rate=0.0, weighted_score=0.0, created_at=now, ) self.violation = DeviceQualityViolationRecord( uid="01900000-0000-7000-8000-000000000808", run_uid=RUN_UID, rule_code="asset_context_complete", severity="error", asset_uid=ASSET_UID, field_name="location", source_uid=SOURCE_UID, source_mapping_uid=MAPPING_UID, message="设备台账位置、组织或责任人不完整", evidence={ "asset_uid": ASSET_UID, "source_mapping_uid": MAPPING_UID, "missing_fields": ["location"], }, created_at=now, expires_at=now + timedelta(days=30), ) self.asset_score = DeviceQualityAssetScoreRecord( uid="01900000-0000-7000-8000-000000000809", run_uid=RUN_UID, asset_uid=ASSET_UID, asset_type="device", status="violated", evaluated_rule_count=3, violation_count=1, score=62.5, created_at=now, ) self.actions = [] def profile(self): self.actions.append(("profile",)) return { "profile_uid": self.version.profile_uid, "name": "设备台账与故障质量策略", "latest_version": self.version, "active_version": self.version, } def bootstrap(self, *, actor_uid): self.actions.append(("bootstrap", actor_uid)) return self.version def revise(self, *, rules, expected_version, actor_uid): self.actions.append( ("revise", rules, expected_version, actor_uid) ) return self.version def versions(self): self.actions.append(("versions",)) return [self.version] def publish(self, version_uid, *, actor_uid): self.actions.append(("publish", version_uid, actor_uid)) return self.version def run(self, *, actor_uid, source_uid=None): self.actions.append(("run", actor_uid, source_uid)) return self.run_record def runs(self, *, page, page_size): self.actions.append(("runs", page, page_size)) return [self.run_record], 1 def get_run(self, run_uid): self.actions.append(("get_run", run_uid)) return self.run_record, [self.rule_result] def violations(self, run_uid, *, rule_code, page, page_size): self.actions.append( ("violations", run_uid, rule_code, page, page_size) ) return [self.violation], 1 def asset_scores(self, run_uid, *, page, page_size): self.actions.append(("asset_scores", run_uid, page, page_size)) return [self.asset_score], 1 @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 = FakeDeviceQualityService() 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_device_quality_service", lambda: service, raising=False, ) app = create_app() app.config.update(TESTING=True) return app.test_client(), service def test_viewer_reads_profile_runs_and_evidence_but_cannot_bootstrap(client): http, service = client headers = {"Authorization": "Bearer viewer"} profile = http.get( "/api/development/v1/device-quality/profile", headers=headers, ) versions = http.get( "/api/development/v1/device-quality/profile/versions", headers=headers, ) runs = http.get( "/api/development/v1/device-quality/runs?page=1&page_size=20", headers=headers, ) detail = http.get( f"/api/development/v1/device-quality/runs/{RUN_UID}", headers=headers, ) violations = http.get( f"/api/development/v1/device-quality/runs/{RUN_UID}/violations" "?rule_code=asset_context_complete&page=1&page_size=20", headers=headers, ) scores = http.get( f"/api/development/v1/device-quality/runs/{RUN_UID}/asset-scores" "?page=1&page_size=20", headers=headers, ) denied = http.post( "/api/development/v1/device-quality/bootstrap", headers=headers, json={}, ) assert profile.status_code == 200 assert profile.get_json()["data"]["active_version"]["version"] == 1 assert len(profile.get_json()["data"]["active_version"]["rules"]) == 7 assert versions.get_json()["data"]["total"] == 1 assert runs.get_json()["data"]["records"][0]["score"] == 85.0 assert detail.get_json()["data"]["rule_results"][0]["pass_rate"] == 0.0 assert violations.get_json()["data"]["records"][0]["asset_uid"] == ( ASSET_UID ) assert violations.get_json()["data"]["records"][0]["evidence"] == { "asset_uid": ASSET_UID, "missing_fields": ["location"], "source_mapping_uid": MAPPING_UID, } assert scores.get_json()["data"]["records"][0]["score"] == 62.5 assert denied.status_code == 403 assert service.actions[0] == ("profile",) def test_editor_can_bootstrap_revise_and_execute_but_cannot_publish(client): http, service = client headers = {"Authorization": "Bearer editor"} bootstrapped = http.post( "/api/development/v1/device-quality/bootstrap", headers=headers, json={}, ) revised = http.post( "/api/development/v1/device-quality/profile/versions", headers=headers, json={ "expected_version": 1, "rules": list(service.version.rules), }, ) executed = http.post( "/api/development/v1/device-quality/runs", headers=headers, json={"source_uid": SOURCE_UID}, ) denied = http.post( f"/api/development/v1/device-quality/profile/versions/" f"{VERSION_UID}/publish", headers=headers, json={}, ) assert bootstrapped.status_code == 201 assert revised.status_code == 201 assert executed.status_code == 201 assert executed.get_json()["data"]["policy_version_uid"] == VERSION_UID assert denied.status_code == 403 assert ("bootstrap", "editor-1") in service.actions assert ("run", "editor-1", SOURCE_UID) in service.actions def test_admin_publishes_exact_version_through_runtime_gate(client): http, service = client response = http.post( f"/api/development/v1/device-quality/profile/versions/" f"{VERSION_UID}/publish", headers={"Authorization": "Bearer admin"}, json={}, ) assert response.status_code == 200 assert response.get_json()["data"]["status"] == "published" assert service.actions[-1] == ("publish", VERSION_UID, "admin-1")