64 lines
1.7 KiB
Python
64 lines
1.7 KiB
Python
from contextlib import asynccontextmanager
|
|
from types import SimpleNamespace
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
|
|
from src.models.approval import ApprovalStatus
|
|
from src.services import approval_db as approval_db_module
|
|
from src.services.approval_db import ApprovalDBService
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pending_handoff_preserves_approved_status_and_records_truth(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
record = SimpleNamespace(
|
|
status=ApprovalStatus.APPROVED,
|
|
extra_metadata={"existing": "kept"},
|
|
)
|
|
|
|
class FakeResult:
|
|
def scalar_one_or_none(self):
|
|
return record
|
|
|
|
class FakeDB:
|
|
async def execute(self, _query):
|
|
return FakeResult()
|
|
|
|
async def flush(self):
|
|
return None
|
|
|
|
@asynccontextmanager
|
|
async def fake_db_context():
|
|
yield FakeDB()
|
|
|
|
monkeypatch.setattr(
|
|
approval_db_module,
|
|
"get_db_context",
|
|
fake_db_context,
|
|
)
|
|
|
|
recorded = await ApprovalDBService().record_controlled_handoff_pending(
|
|
uuid4(),
|
|
execution_kind="host_ansible_check_mode_queued",
|
|
receipt={
|
|
"automation_run_id": "run-host-188",
|
|
"single_writer_executor": "awoooi-ansible-executor-broker",
|
|
},
|
|
)
|
|
|
|
assert recorded is True
|
|
assert record.status == ApprovalStatus.APPROVED
|
|
assert record.extra_metadata == {
|
|
"existing": "kept",
|
|
"execution_kind": "host_ansible_check_mode_queued",
|
|
"execution_state": "queued",
|
|
"repair_attempted": False,
|
|
"repair_executed": False,
|
|
"repair_verified": False,
|
|
"incident_closure_allowed": False,
|
|
"automation_run_id": "run-host-188",
|
|
"single_writer_executor": "awoooi-ansible-executor-broker",
|
|
}
|