DataOps-platform/tests/integration/test_rbac_api.py

from __future__ import annotations

import os
import uuid

import pytest
from sqlalchemy import create_engine, text


pytestmark = pytest.mark.integration


@pytest.fixture()
def rbac_client(monkeypatch):
    database_url = os.environ.get("TEST_DATABASE_URL")
    admin_password = os.environ.get("TEST_ADMIN_PASSWORD")
    if not database_url or not admin_password:
        pytest.skip("TEST_DATABASE_URL and TEST_ADMIN_PASSWORD are required")
    monkeypatch.setenv("DATABASE_URL", database_url)
    monkeypatch.setenv("SECRET_KEY", "dataops-local-test-secret-key")
    from app import create_app

    app = create_app()
    app.config.update(TESTING=True)
    return app.test_client(), database_url, admin_password


def _login(client, username, password):
    response = client.post(
        "/api/system/auth/login", json={"username": username, "password": password}
    )
    return response, (response.get_json().get("data") or {}).get("token")


def test_admin_manages_users_and_viewer_is_forbidden(rbac_client):
    client, database_url, admin_password = rbac_client
    username = f"viewer_{uuid.uuid4().hex[:10]}"
    user_id = None
    try:
        login, admin_token = _login(client, "admin", admin_password)
        assert login.status_code == 200
        assert admin_token

        me = client.get(
            "/api/system/auth/me",
            headers={"Authorization": f"Bearer {admin_token}"},
        )
        assert me.status_code == 200
        assert "users:manage" in me.get_json()["data"]["permissions"]

        created = client.post(
            "/api/system/users",
            json={
                "username": username,
                "password": "ViewerPass123",
                "roles": ["viewer"],
            },
            headers={"Authorization": f"Bearer {admin_token}"},
        )
        assert created.status_code == 201
        user_id = created.get_json()["data"]["id"]

        viewer_login, viewer_token = _login(client, username, "ViewerPass123")
        assert viewer_login.status_code == 200
        forbidden = client.get(
            "/api/system/users",
            headers={"Authorization": f"Bearer {viewer_token}"},
        )
        assert forbidden.status_code == 403

        disabled = client.put(
            f"/api/system/users/{user_id}",
            json={"status": "disabled"},
            headers={"Authorization": f"Bearer {admin_token}"},
        )
        assert disabled.status_code == 200
        rejected, _ = _login(client, username, "ViewerPass123")
        assert rejected.status_code == 401

        invalid = client.get(
            "/api/system/auth/me",
            headers={"Authorization": "Bearer invalid-token"},
        )
        assert invalid.status_code == 401
    finally:
        if user_id:
            engine = create_engine(database_url)
            with engine.begin() as connection:
                connection.execute(
                    text("DELETE FROM public.users WHERE id = CAST(:id AS uuid)"),
                    {"id": user_id},
                )
            engine.dispose()