DataOps-platform/tests/security/test_trusted_delivery_controls_service.py

from __future__ import annotations

import uuid

import pytest

from app.core.system.trusted_delivery_controls import TrustedDeliveryControlError
from app.core.system.trusted_delivery_controls_service import (
    TrustedDeliveryControlsService,
)

ACTOR = "01900000-0000-7000-8000-000000068801"


class Repository:
    def __init__(self):
        self.profiles = set()
        self.records = []
        self.hold = False

    def users_available(self, values):
        return set(values)

    def profile_exists(self, profile_id, version):
        return (profile_id, version) in self.profiles

    def create_profile(self, record):
        self.profiles.add((record["profile_id"], record["version"]))
        return record

    def activate_profile(self, profile_id, version, idempotency_key, request_digest, actor_uid, approval_ref, approval_digest, operation):
        return {"profile_id": profile_id, "version": version, "status": "active", "operation": operation}

    def active_profile_summary(self, profile_id):
        return None

    def approve_capability(self, record):
        self.records.append(record)
        return {"status": "approved", "config_digest": record["config_digest"]}

    def record_release_gate(self, record):
        self.records.append(record)
        return {"uid": "01900000-0000-7000-8000-000000068802", **record}

    def create_evidence(self, record):
        self.records.append(record)
        return record

    def evidence_summary(self):
        return []

    def approve_destruction(self, record):
        if self.hold:
            raise PermissionError("destruction blocked by active legal hold")
        self.records.append(record)
        return {"status": "approved"}


def _digest(value: str) -> str:
    return value * 64


def test_controls_service_binds_catalog_capability_and_release_gate_to_safe_digests_only():
    repository = Repository()
    service = TrustedDeliveryControlsService(repository)
    profile = service.activate_profile({"profile_id": "finance", "version": "1.0", "approval_ref": "approval:profile", "approval_digest": _digest("a"), "idempotency_key": "profile-activate"}, actor_uid=ACTOR)
    assert profile["status"] == "active"
    approved = service.approve_capability({"provider": "kms", "capability": "kms-key-wrap", "version": "v1", "config_digest": _digest("b"), "approval_ref": "approval:capability-1", "approval_digest": _digest("c"), "idempotency_key": "capability-1"}, actor_uid=ACTOR)
    assert approved["status"] == "approved"
    gate = service.evaluate_release_gate({"artifact_digest": _digest("a"), "sbom_digest": _digest("b"), "license_policy_digest": _digest("c"), "vulnerability_scan_digest": _digest("d"), "approval_id": "release-approval-1", "approval_version": "v1", "scan_decision": "passed", "evidence_expires_at": "2099-01-01T00:00:00+00:00", "approval_ref": "approval:gate-1", "approval_digest": _digest("e"), "idempotency_key": "gate-1"}, actor_uid=ACTOR)
    assert gate["decision_code"] == "RELEASE_GATE_APPROVED"
    assert "raw" not in str(repository.records).lower()


def test_controls_service_rejects_provider_mismatch_and_legal_hold_destruction():
    repository = Repository()
    service = TrustedDeliveryControlsService(repository)
    with pytest.raises(TrustedDeliveryControlError, match="unsupported"):
        service.approve_capability({"provider": "kms", "capability": "dlp-export-check", "version": "v1", "config_digest": _digest("b"), "idempotency_key": "capability-2"}, actor_uid=ACTOR)
    repository.hold = True
    with pytest.raises(PermissionError, match="active legal hold"):
        service.approve_destruction({"asset_uid": str(uuid.uuid4()), "approval_refs": ["destroy:one", "destroy:two"], "evidence_digest": _digest("c"), "idempotency_key": "destroy-1"}, actor_uid=ACTOR)