DataOps-platform/tests/integration/test_phase3_wp06_http_commit_postgres.py

"""Real Flask/SQLAlchemy commit boundary checks for trusted-delivery HTTP."""

from __future__ import annotations

import os
import uuid
from datetime import UTC, datetime, timedelta

import pytest
from sqlalchemy import create_engine, text

pytestmark = pytest.mark.integration


def _url():
    value = os.environ.get("TEST_DATABASE_URL")
    if not value:
        pytest.skip("TEST_DATABASE_URL is required")
    return value


def _policy(actor, code):
    return {
        "code": code,
        "version": "1.0.0",
        "selector": {
            "subjects": [actor], "roles": ["editor"],
            "business_domains": [str(uuid.uuid4())], "assets": [str(uuid.uuid4())],
            "purposes": ["quality_review"], "environments": ["test"],
            "actions": ["read"], "classifications": ["sensitive"],
            "expires_at": (datetime.now(UTC) + timedelta(days=1)).isoformat(),
        },
        "resource_rules": {
            "row_filter": {"op": "eq", "field": "region", "value": "CN-31"},
            "field_rules": [{"field": "code", "effect": "allow", "mask_ref": None}],
        },
    }


def test_trusted_delivery_http_success_is_committed_and_commit_failure_rolls_back(monkeypatch):
    """A 201 is observable by another connection; a failed commit leaves no row."""
    database_url = _url()
    actor = str(uuid.uuid4())
    prefix = f"WP06_HTTP_{actor[:8].upper()}"
    engine = create_engine(database_url, pool_pre_ping=True)
    try:
        with engine.begin() as connection:
            connection.execute(text("""INSERT INTO public.users(id,username,display_name,password_hash,status)
              VALUES(CAST(:id AS uuid),:name,:name,'test','active')"""), {"id": actor, "name": f"wp06-test-http-{actor[:8]}"})
        monkeypatch.setenv("DATABASE_URL", database_url)
        monkeypatch.setattr(
            "app.core.system.permissions.authenticate_request",
            lambda: {"id": actor, "sub": actor, "roles": ["editor"]},
        )
        from app import create_app, db

        app = create_app()
        app.config.update(TESTING=True)
        client = app.test_client()
        response = client.post("/api/system/trusted-delivery/policy-versions", json=_policy(actor, f"{prefix}_OK"))
        assert response.status_code == 201
        assert response.headers["Cache-Control"] == "no-store"
        with engine.connect() as observer:
            assert observer.execute(text("SELECT count(*) FROM public.trusted_delivery_policy_versions WHERE code=:code"), {"code": f"{prefix}_OK"}).scalar_one() == 1

        def fail_commit():
            raise RuntimeError("synthetic commit failure")

        monkeypatch.setattr(db.session, "commit", fail_commit)
        rejected = client.post("/api/system/trusted-delivery/policy-versions", json=_policy(actor, f"{prefix}_ROLLBACK"))
        assert rejected.status_code == 400
        with engine.connect() as observer:
            assert observer.execute(text("SELECT count(*) FROM public.trusted_delivery_policy_versions WHERE code=:code"), {"code": f"{prefix}_ROLLBACK"}).scalar_one() == 0
    finally:
        with engine.begin() as connection:
            connection.execute(text("DELETE FROM public.trusted_delivery_policy_transitions WHERE policy_uid IN (SELECT uid FROM public.trusted_delivery_policy_versions WHERE code LIKE :prefix)"), {"prefix": f"{prefix}%"})
            connection.execute(text("DELETE FROM public.trusted_delivery_policy_versions WHERE code LIKE :prefix"), {"prefix": f"{prefix}%"})
            connection.execute(text("ALTER TABLE public.trusted_delivery_evidence DISABLE TRIGGER USER"))
            connection.execute(text("DELETE FROM public.trusted_delivery_evidence WHERE actor_uid=CAST(:id AS uuid)"), {"id": actor})
            connection.execute(text("ALTER TABLE public.trusted_delivery_evidence ENABLE TRIGGER USER"))
            connection.execute(text("DELETE FROM public.users WHERE id=CAST(:id AS uuid)"), {"id": actor})
        engine.dispose()