DataOps-platform/tests/test_security_governance_api.py

from __future__ import annotations

USER_UID = "01900000-0000-7000-8000-000000018801"
FINDING_UID = "01900000-0000-7000-8000-000000018802"


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

    def dashboard(self):
        return {"pending_classification_reviews": 1, "open_vulnerability_count": 2}

    def list_classification_findings(self, **filters):
        self.calls.append(("findings", filters))
        return [{"uid": FINDING_UID, "status": "pending_review"}]

    def scan_sensitive_sample(self, payload, actor_uid):
        self.calls.append(("scan", payload, actor_uid))
        return {"uid": "scan-1", "sample_retained": False, "findings": []}

    def evaluate_access(self, payload):
        self.calls.append(("evaluate", payload))
        return {"uid": "decision-1", "decision": "denied", "reason_code": "default_deny"}

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


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


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

    service = FakeSecurityService()
    monkeypatch.setattr(security_governance, "_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_security_governance_role_boundaries_and_server_bound_identity(monkeypatch):
    client, service = _client(monkeypatch)
    assert client.get(
        "/api/system/security-governance/dashboard", headers=_headers("viewer")
    ).status_code == 200
    assert client.get(
        "/api/system/security-governance/findings?status=pending_review",
        headers=_headers("viewer"),
    ).status_code == 200

    forbidden = client.post(
        "/api/system/security-governance/scans", json={}, headers=_headers("viewer")
    )
    assert forbidden.status_code == 403
    assert client.post(
        "/api/system/security-governance/scans", json={}, headers=_headers("editor")
    ).status_code == 201

    decision = client.post(
        "/api/system/security-governance/access/evaluate",
        json={"user_uid": "forged", "roles": ["admin"]},
        headers=_headers("editor"),
    )
    assert decision.status_code == 201
    body = service.calls[-1][1]
    assert body["user_uid"] == USER_UID
    assert body["roles"] == ["editor"]


def test_independent_review_is_admin_only_and_requires_etag(monkeypatch):
    client, service = _client(monkeypatch)
    path = f"/api/system/security-governance/findings/{FINDING_UID}/review"
    assert client.post(path, json={}, headers=_headers("editor")).status_code == 403
    assert client.post(path, json={}, headers=_headers("admin")).status_code == 400
    response = client.post(
        path,
        json={"decision": "confirm"},
        headers=_headers("admin", **{"If-Match": '"1"'}),
    )
    assert response.status_code == 200
    assert response.headers["ETag"] == '"2"'
    assert service.calls[-1][3] == 1