File size: 3,854 Bytes
6d20eab
 
 
36d5132
 
 
 
 
 
 
 
 
 
 
6d20eab
 
 
 
 
 
 
 
 
 
 
 
 
 
 
36d5132
 
6d20eab
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
36d5132
 
6d20eab
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
36d5132
 
6d20eab
 
 
 
 
 
 
 
 
 
 
 
36d5132
 
6d20eab
36d5132
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
"""
Tests for governance endpoints: /api/v1/intents/evaluate
"""
import pytest
from app.database.models_intents import TenantDB


@pytest.fixture(autouse=True)
def seed_tenant(db_session):
    """Ensure the tenant 'test-tenant' exists before each test."""
    tenant = db_session.query(TenantDB).filter_by(id="test-tenant").first()
    if not tenant:
        db_session.add(TenantDB(id="test-tenant", name="Test Tenant"))
        db_session.commit()


def test_evaluate_provision_intent(client):
    payload = {
        "intent_type": "provision_resource",
        "environment": "prod",
        "resource_type": "database",
        "region": "eastus",
        "size": "Standard",
        "estimated_cost": 1200,
        "policy_violations": [],
        "requester": "alice",
        "provenance": {},
        "configuration": {}
    }
    response = client.post("/api/v1/intents/evaluate", json=payload,
                           headers={"X-Tenant-ID": "test-tenant"})
    assert response.status_code == 200, response.text
    data = response.json()
    assert "risk_score" in data


def test_evaluate_grant_access(client):
    payload = {
        "intent_type": "grant_access",
        "environment": "dev",
        "principal": "bob",
        "permission_level": "read",
        "resource_scope": "/subscriptions/123",
        "estimated_cost": None,
        "policy_violations": [],
        "requester": "alice",
        "provenance": {},
        "justification": "test"
    }
    response = client.post("/api/v1/intents/evaluate", json=payload,
                           headers={"X-Tenant-ID": "test-tenant"})
    assert response.status_code == 200, response.text
    data = response.json()
    assert "risk_score" in data


def test_evaluate_deploy_config(client):
    payload = {
        "intent_type": "deploy_config",
        "environment": "staging",
        "service_name": "payments-api",
        "change_scope": "canary",
        "deployment_target": "staging",
        "estimated_cost": 20,
        "policy_violations": [],
        "requester": "alice",
        "provenance": {},
        "configuration": {}
    }
    response = client.post("/api/v1/intents/evaluate", json=payload,
                           headers={"X-Tenant-ID": "test-tenant"})
    assert response.status_code == 200, response.text
    data = response.json()
    assert "risk_score" in data


def test_invalid_intent_type(client):
    payload = {
        "intent_type": "UnknownIntent",
        "environment": "prod",
        "requester": "alice",
        "provenance": {}
    }
    response = client.post("/api/v1/intents/evaluate", json=payload,
                           headers={"X-Tenant-ID": "test-tenant"})
    assert response.status_code == 422


def test_evaluate_with_criticality(client):
    """v4.3.2: criticality is accepted and a context_hash is generated."""
    payload = {
        "intent_type": "provision_resource",
        "environment": "prod",
        "resource_type": "database",
        "region": "eastus",
        "size": "Standard",
        "estimated_cost": 1200,
        "policy_violations": [],
        "requester": "alice",
        "provenance": {},
        "configuration": {},
        "criticality": 0.85
    }
    response = client.post("/api/v1/intents/evaluate", json=payload,
                           headers={"X-Tenant-ID": "test-tenant"})
    assert response.status_code == 200, response.text
    data = response.json()
    assert "risk_score" in data
    # The healing_intent dict should contain the new fields.
    healing = data.get("healing_intent", {})
    # criticality is passed through
    assert healing.get("criticality") == 0.85
    # context_hash is computed by the governance loop (a 64‑char hex string)
    ctx_hash = healing.get("context_hash")
    assert isinstance(ctx_hash, str) and len(ctx_hash) == 64