File size: 4,174 Bytes
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 | 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
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()
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",
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,
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,
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",
intent_type="ProvisionResourceIntent",
payload={},
created_at=datetime.datetime.utcnow()
)
db_session.add(intent)
db_session.commit()
record_outcome(
db_session,
"intent_def",
True,
None,
None,
mock_risk_engine)
with pytest.raises(OutcomeConflictError):
record_outcome(
db_session,
"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,
"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",
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,
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()
|