DataOps-platform/tests/test_work_center_api.py

from __future__ import annotations

USER_UID = "01900000-0000-7000-8000-000000007901"
TASK_UID = "01900000-0000-7000-8000-000000007902"
WORKFLOW_UID = "01900000-0000-7000-8000-000000007903"


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

    def list_tasks(self, **filters):
        self.calls.append(("list_tasks", filters))
        return [{"uid": TASK_UID, "subject_type": "quality_issue"}]

    def create_task(self, payload, actor_uid):
        self.calls.append(("create_task", payload, actor_uid))
        return {"uid": TASK_UID, "current_version": 1, **payload}

    def review_task(self, uid, payload, expected_version, actor_uid):
        self.calls.append(("review", uid, payload, expected_version, actor_uid))
        return {"uid": uid, "status": "approved", "current_version": 2}

    def list_workflows(self):
        return []

    def create_workflow(self, payload, actor_uid):
        self.calls.append(("workflow", payload, actor_uid))
        return {"uid": WORKFLOW_UID, "current_version": 1}

    def dashboard(self, at=None):
        return {"pending_count": 1, "overdue_count": 0, "closure_rate": 0}

    def get_notification_preferences(self, user_uid):
        return {"user_uid": user_uid, "revision": 0, "enabled_channels": ["in_app"]}

    def replace_notification_preferences(self, payload, expected_revision, actor_uid):
        self.calls.append(("preferences", payload, expected_revision, actor_uid))
        return {"user_uid": actor_uid, "revision": expected_revision + 1, **payload}


def _headers(role, **extra):
    return {"Authorization": f"Bearer {role}", **extra}


def _client(monkeypatch):
    from app import create_app
    from app.api.system import work_center

    service = FakeWorkCenterService()
    monkeypatch.setattr(work_center, "_service", lambda: service)
    monkeypatch.setattr(
        "app.core.system.auth.load_identity_from_token",
        lambda token, secret: (
            {"id": USER_UID, "username": token, "roles": [token]}
            if token in {"viewer", "editor", "admin"}
            else None
        ),
    )
    app = create_app()
    app.config.update(TESTING=True)
    return app.test_client(), service


def test_unified_inbox_is_readable_and_filters_non_manager_scope(monkeypatch):
    client, service = _client(monkeypatch)
    response = client.get(
        "/api/system/work-center/tasks?subject_type=quality_issue",
        headers=_headers("viewer"),
    )
    assert response.status_code == 200
    filters = service.calls[-1][1]
    assert filters["requester_uid"] == USER_UID
    assert filters["can_manage"] is False
    assert filters["subject_type"] == "quality_issue"


def test_editor_can_create_and_review_versioned_task_but_not_manage_flow(monkeypatch):
    client, service = _client(monkeypatch)
    created = client.post(
        "/api/system/work-center/tasks",
        json={"title": "quality remediation"},
        headers=_headers("editor"),
    )
    assert created.status_code == 201
    assert created.headers["ETag"] == '"1"'

    missing_version = client.post(
        f"/api/system/work-center/tasks/{TASK_UID}/review",
        json={"decision": "approve", "reason": "ready"},
        headers=_headers("editor"),
    )
    assert missing_version.status_code == 428
    reviewed = client.post(
        f"/api/system/work-center/tasks/{TASK_UID}/review",
        json={"decision": "approve", "reason": "ready"},
        headers=_headers("editor", **{"If-Match": '"1"'}),
    )
    assert reviewed.status_code == 200
    assert reviewed.headers["ETag"] == '"2"'
    assert service.calls[-1][3] == 1

    forbidden = client.post(
        "/api/system/work-center/workflows",
        json={"code": "FLOW"},
        headers=_headers("editor"),
    )
    assert forbidden.status_code == 403


def test_admin_manages_workflow_and_user_versions_preferences(monkeypatch):
    client, service = _client(monkeypatch)
    workflow = client.post(
        "/api/system/work-center/workflows",
        json={"code": "FLOW"},
        headers=_headers("admin"),
    )
    assert workflow.status_code == 201
    preference = client.put(
        "/api/system/work-center/preferences",
        json={"enabled_channels": ["in_app"]},
        headers=_headers("editor", **{"If-Match": '"0"'}),
    )
    assert preference.status_code == 200
    assert preference.headers["ETag"] == '"1"'
    assert service.calls[-1][2] == 0