Spaces:
Configuration error
Configuration error
File size: 4,777 Bytes
6d20eab de5a0fa 6d20eab de5a0fa 6d20eab de5a0fa 6d20eab de5a0fa 6d20eab de5a0fa 6d20eab de5a0fa 6d20eab de5a0fa 6d20eab de5a0fa 6d20eab de5a0fa 6d20eab de5a0fa 6d20eab de5a0fa 6d20eab | 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 | import pytest
import datetime
from unittest.mock import MagicMock
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.database.base import Base
from app.database.models_intents import IntentDB, TenantDB
from app.services.outcome_service import record_outcome, OutcomeConflictError
from agentic_reliability_framework.core.governance.intents import (
ProvisionResourceIntent,
ResourceType,
)
@pytest.fixture
def db_session():
engine = create_engine("sqlite:///:memory:", future=True)
TestingSessionLocal = sessionmaker(bind=engine, future=True)
Base.metadata.create_all(bind=engine)
sess = TestingSessionLocal()
# Ensure a tenant exists for foreign key constraints
if not sess.query(TenantDB).filter_by(id="test-tenant").first():
sess.add(TenantDB(id="test-tenant", name="Test Tenant"))
sess.commit()
yield sess
sess.close()
@pytest.fixture
def mock_risk_engine():
engine = MagicMock()
engine.update_outcome = MagicMock()
return engine
def test_record_outcome_creates_row_and_updates_engine(
db_session, mock_risk_engine):
oss_intent = ProvisionResourceIntent(
resource_type=ResourceType.VM,
region="eastus",
size="Standard",
environment="dev",
requester="test_user"
)
oss_payload = oss_intent.model_dump(mode='json')
intent = IntentDB(
deterministic_id="intent_abc",
tenant_id="test-tenant", # <-- required
intent_type="ProvisionResourceIntent",
payload={},
oss_payload=oss_payload,
created_at=datetime.datetime.utcnow()
)
db_session.add(intent)
db_session.commit()
db_session.refresh(intent)
outcome = record_outcome(
db=db_session,
tenant_id="test-tenant",
deterministic_id="intent_abc",
success=True,
recorded_by="tester",
notes="works",
risk_engine=mock_risk_engine,
idempotency_key="key123"
)
assert outcome.success is True
assert outcome.recorded_by == "tester"
assert outcome.idempotency_key == "key123"
mock_risk_engine.update_outcome.assert_called_once()
# Idempotent call with same key should return existing outcome and not
# call engine again
outcome2 = record_outcome(
db=db_session,
tenant_id="test-tenant",
deterministic_id="intent_abc",
success=True,
recorded_by="tester",
notes="again",
risk_engine=mock_risk_engine,
idempotency_key="key123"
)
assert outcome2.id == outcome.id
mock_risk_engine.update_outcome.assert_called_once() # still once
def test_conflict_different_result(db_session, mock_risk_engine):
intent = IntentDB(
deterministic_id="intent_def",
tenant_id="test-tenant", # <-- required
intent_type="ProvisionResourceIntent",
payload={},
created_at=datetime.datetime.utcnow()
)
db_session.add(intent)
db_session.commit()
record_outcome(
db_session,
"test-tenant",
"intent_def",
True,
None,
None,
mock_risk_engine)
with pytest.raises(OutcomeConflictError):
record_outcome(
db_session,
"test-tenant",
"intent_def",
False,
None,
None,
mock_risk_engine)
def test_nonexistent_intent(db_session, mock_risk_engine):
with pytest.raises(ValueError):
record_outcome(
db_session,
"test-tenant",
"missing",
True,
None,
None,
mock_risk_engine)
def test_record_outcome_reconstruction_failure_does_not_update_engine(
db_session, mock_risk_engine):
# Create an intent with invalid oss_payload (missing required fields)
intent = IntentDB(
deterministic_id="intent_bad",
tenant_id="test-tenant", # <-- required
intent_type="ProvisionResourceIntent",
payload={},
oss_payload={"intent_type": "provision_resource"}, # missing fields
created_at=datetime.datetime.utcnow()
)
db_session.add(intent)
db_session.commit()
# This should NOT call risk_engine.update_outcome (no dummy fallback)
outcome = record_outcome(
db=db_session,
tenant_id="test-tenant",
deterministic_id="intent_bad",
success=True,
recorded_by="tester",
notes="fallback test",
risk_engine=mock_risk_engine
)
assert outcome.success is True
# The engine should NOT be updated because reconstruction failed
mock_risk_engine.update_outcome.assert_not_called()
|