Spaces:
Runtime error
Runtime error
File size: 11,041 Bytes
8aca429 | 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 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 | """
End‑to‑end integration tests for the ARF governance pipeline.
These tests exercise the full path from HTTP request to HealingIntent
response, validating that every layer – API, governance loop, policy
engine, risk engine, audit log, and optional skill/criticality features –
behaves correctly under realistic conditions.
v4.3.2: Covers basic evaluation, skill context, criticality, and audit
trace verification.
"""
import pytest
import time
from app.database.models_intents import TenantDB, DecisionAuditLogDB
@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()
class TestFullPipeline:
"""End‑to‑end tests for the /intents/evaluate endpoint."""
def test_basic_provision_evaluation(self, client):
"""A minimal valid request returns 200 and a well‑formed HealingIntent."""
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()
# Top‑level fields
assert "risk_score" in data
assert "explanation" in data
assert "deterministic_id" in data
assert "recommended_action" in data
assert isinstance(data["risk_score"], float)
assert 0.0 <= data["risk_score"] <= 1.0
# HealingIntent contract
healing = data.get("healing_intent", {})
assert healing.get("action") is not None
assert healing.get("component") is not None
assert "justification" in healing
assert "confidence" in healing
assert "version" in healing
assert healing["version"] == "2.6.0"
# v4.3.2: context_hash must be present (64 hex chars)
ctx_hash = healing.get("context_hash")
assert isinstance(ctx_hash, str) and len(ctx_hash) == 64, (
f"context_hash missing or invalid: {ctx_hash}"
)
def test_audit_log_written(self, client, db_session):
"""A successful evaluation writes a row to the decision audit log."""
# Count existing rows for this tenant
before = (
db_session.query(DecisionAuditLogDB)
.filter_by(tenant_id="test-tenant")
.count()
)
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
# The write_audit_log runs as a background task; give it a moment.
time.sleep(0.5)
after = (
db_session.query(DecisionAuditLogDB)
.filter_by(tenant_id="test-tenant")
.count()
)
assert after == before + 1, (
f"Expected one new audit log entry, but count went from "
f"{before} to {after}"
)
def test_skill_context_injection(self, client):
"""When skill_id is provided, the response includes skill posterior data."""
payload = {
"intent_type": "provision_resource",
"environment": "prod",
"resource_type": "database",
"region": "eastus",
"size": "Standard",
"estimated_cost": 1200,
"policy_violations": [],
"requester": "alice",
"provenance": {},
"configuration": {},
"skill_id": "pdf-skill",
}
response = client.post(
"/api/v1/intents/evaluate",
json=payload,
headers={"X-Tenant-ID": "test-tenant"},
)
assert response.status_code == 200
healing = response.json().get("healing_intent", {})
# Skill fields should be present in the HealingIntent
assert "skill_id" in healing
assert healing["skill_id"] == "pdf-skill"
# Because the skill registry is a singleton, the skill may or may not
# already exist. In either case, the fields are populated with either
# the posterior or the default prior.
assert "skill_alpha" in healing
assert "skill_beta" in healing
assert "skill_reliability_score" in healing
assert "skill_version" in healing
def test_criticality_parameter(self, client):
"""The criticality field is accepted and flows into the HealingIntent."""
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
healing = response.json().get("healing_intent", {})
assert healing.get("criticality") == 0.85, (
f"criticality should be 0.85, got {healing.get('criticality')}"
)
def test_policy_violation_denial(self, client):
"""An intent with a policy violation returns DENY."""
payload = {
"intent_type": "provision_resource",
"environment": "prod",
"resource_type": "database",
"region": "westus", # not in default allowed set
"size": "Standard",
"estimated_cost": 1200,
"policy_violations": ["Region 'westus' not allowed"], # pre‑computed
"requester": "alice",
"provenance": {},
"configuration": {}
}
response = client.post(
"/api/v1/intents/evaluate",
json=payload,
headers={"X-Tenant-ID": "test-tenant"},
)
assert response.status_code == 200
data = response.json()
assert data.get("recommended_action") == "deny", (
f"Expected action=deny, got {data.get('recommended_action')}"
)
class TestHealingPipeline:
"""End‑to‑end tests for the /healing/evaluate endpoint."""
def test_basic_healing_evaluation(self, client):
"""A reliability event triggers candidate healing actions."""
payload = {
"event": {
"component": "checkout-service",
"latency_p99": 600.0,
"error_rate": 0.25,
"service_mesh": "default",
"cpu_util": 0.85,
"memory_util": 0.90,
}
}
response = client.post(
"/api/v1/healing/evaluate",
json=payload,
headers={"X-Tenant-ID": "test-tenant"},
)
assert response.status_code == 200
data = response.json()
assert "selected_action" in data
assert data["selected_action"] != "NO_ACTION", (
"Expected at least one healing action to be triggered"
)
def test_healing_with_skill_context(self, client):
"""Skill context biases the healing decision utility."""
payload = {
"event": {
"component": "checkout-service",
"latency_p99": 600.0,
"error_rate": 0.25,
"service_mesh": "default",
"cpu_util": 0.85,
"memory_util": 0.90,
},
"skill_id": "pdf-skill",
"skill_version": 1,
}
response = client.post(
"/api/v1/healing/evaluate",
json=payload,
headers={"X-Tenant-ID": "test-tenant"},
)
assert response.status_code == 200
data = response.json()
# The response should echo the skill context back
assert data.get("skill_id") == "pdf-skill"
assert data.get("skill_version") == 1
assert "selected_action" in data
class TestOutcomeRecording:
"""End‑to‑end tests for the /intents/outcome endpoint."""
def test_record_outcome_updates_risk_engine(self, client, db_session):
"""Recording a successful outcome for a previously evaluated intent
updates the conjugate posterior and skill registry."""
# Step 1: evaluate an intent to create a record
payload = {
"intent_type": "provision_resource",
"environment": "prod",
"resource_type": "database",
"region": "eastus",
"size": "Standard",
"estimated_cost": 1200,
"policy_violations": [],
"requester": "alice",
"provenance": {},
"configuration": {},
}
eval_resp = client.post(
"/api/v1/intents/evaluate",
json=payload,
headers={"X-Tenant-ID": "test-tenant"},
)
assert eval_resp.status_code == 200
deterministic_id = eval_resp.json()["deterministic_id"]
# Step 2: record a successful outcome
outcome_payload = {
"deterministic_id": deterministic_id,
"success": True,
"recorded_by": "tester",
"notes": "integration test",
}
outcome_resp = client.post(
"/api/v1/intents/outcome",
json=outcome_payload,
headers={"X-Tenant-ID": "test-tenant"},
)
assert outcome_resp.status_code == 200
assert "outcome_id" in outcome_resp.json()
# Step 3: verify that the outcome row exists in the database
from app.database.models_intents import OutcomeDB
outcome = (
db_session.query(OutcomeDB)
.filter_by(idempotency_key=None) # we didn't send one
.order_by(OutcomeDB.id.desc())
.first()
)
assert outcome is not None
assert outcome.success is True
assert outcome.recorded_by == "tester"
|