DataOps-platform/tests/integration/test_responsibility_postgres.py

from __future__ import annotations

import os
import uuid

import pytest
from sqlalchemy import create_engine, text


pytestmark = pytest.mark.integration


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


def test_device_responsibility_matrix_is_versioned_audited_and_readable(
    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)
    client = app.test_client()
    username = f"asset_manager_{uuid.uuid4().hex[:10]}"
    resource_uid = f"device-{uuid.uuid4().hex[:12]}"
    user_id = None
    engine = create_engine(database_url)
    try:
        login, admin_token = _login(client, "admin", admin_password)
        assert login.status_code == 200
        assert admin_token

        created = client.post(
            "/api/system/users",
            json={
                "username": username,
                "display_name": "设备资产管理员",
                "password": "AssetManager123",
                "roles": ["viewer"],
            },
            headers={"Authorization": f"Bearer {admin_token}"},
        )
        assert created.status_code == 201
        user_id = created.get_json()["data"]["id"]

        matrix = {
            "assignments": [
                {
                    "user_id": user_id,
                    "responsibility_role": "asset_manager",
                    "raci_role": "accountable",
                }
            ]
        }
        updated = client.put(
            f"/api/system/responsibilities/device_asset/{resource_uid}",
            json=matrix,
            headers={
                "Authorization": f"Bearer {admin_token}",
                "If-Match": '"0"',
            },
        )
        assert updated.status_code == 200
        assert updated.headers["ETag"] == '"1"'

        viewer_login, viewer_token = _login(
            client,
            username,
            "AssetManager123",
        )
        assert viewer_login.status_code == 200
        readable = client.get(
            f"/api/system/responsibilities/device_asset/{resource_uid}",
            headers={"Authorization": f"Bearer {viewer_token}"},
        )
        assert readable.status_code == 200
        assert readable.get_json()["data"]["assignments"][0]["user_id"] == user_id

        forbidden = client.put(
            f"/api/system/responsibilities/device_asset/{resource_uid}",
            json=matrix,
            headers={
                "Authorization": f"Bearer {viewer_token}",
                "If-Match": '"1"',
            },
        )
        assert forbidden.status_code == 403

        stale = client.put(
            f"/api/system/responsibilities/device_asset/{resource_uid}",
            json=matrix,
            headers={
                "Authorization": f"Bearer {admin_token}",
                "If-Match": '"0"',
            },
        )
        assert stale.status_code == 409

        with engine.connect() as connection:
            audit_count = connection.execute(
                text(
                    """
                    SELECT COUNT(*)
                    FROM public.governance_responsibility_audit_events
                    WHERE resource_type = 'device_asset'
                      AND resource_uid = :resource_uid
                    """
                ),
                {"resource_uid": resource_uid},
            ).scalar_one()
        assert audit_count == 1
    finally:
        with engine.begin() as connection:
            connection.execute(
                text(
                    """
                    DELETE FROM public.governance_responsibility_scopes
                    WHERE resource_type = 'device_asset'
                      AND resource_uid = :resource_uid
                    """
                ),
                {"resource_uid": resource_uid},
            )
            if user_id:
                connection.execute(
                    text(
                        "DELETE FROM public.users WHERE id = CAST(:user_id AS uuid)"
                    ),
                    {"user_id": user_id},
                )
        engine.dispose()