132 lines
5.4 KiB
Python
132 lines
5.4 KiB
Python
"""Real PostgreSQL qualification for approval revocation versus serving promotion."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib
|
|
import threading
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from concurrent.futures import TimeoutError as FutureTimeoutError
|
|
|
|
import pytest
|
|
from sqlalchemy import event, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
pytest_plugins = ("tests.test_managed_identities_postgres",)
|
|
|
|
from modelforge_api.domain.lifecycle_contracts import ApprovalDecision
|
|
from modelforge_api.persistence.models import (
|
|
CapabilityContract,
|
|
LifecycleApprovalRequest,
|
|
LifecycleEvent,
|
|
LifecycleOperation,
|
|
)
|
|
from modelforge_api.persistence.serving_repository import ServingRepository
|
|
from modelforge_api.services.lifecycle import LifecycleError, LifecycleService
|
|
from tests.test_lifecycle_serving import approved_plan, build_pair
|
|
from tests.test_managed_identities_postgres import IdentityDatabase
|
|
|
|
|
|
def test_revoked_approval_fences_inflight_serving_promotion(
|
|
identity_database: IdentityDatabase,
|
|
) -> None:
|
|
"""A commit waiting on the approval row must see the revocation, not switch the route."""
|
|
importlib.import_module("alembic.command").upgrade(identity_database.migration(), "head")
|
|
engine = identity_database.runtime
|
|
with Session(engine) as session:
|
|
pair = build_pair(session)
|
|
_, _, current, candidate, selection, _ = pair
|
|
_, plan, operation, _ = approved_plan(pair)
|
|
approval_id = plan.approval_request_id
|
|
operation_id = operation.id
|
|
contract_id = current.capability_contract_id
|
|
current_id, candidate_id = current.id, candidate.id
|
|
original_generation = selection.expected_route_generation
|
|
baseline_events = set(session.scalars(select(LifecycleEvent.id)))
|
|
|
|
revoker_holds_lock = threading.Event()
|
|
release_revoker = threading.Event()
|
|
promoter_requested_lock = threading.Event()
|
|
|
|
def revoke() -> None:
|
|
with Session(engine) as session:
|
|
original_commit = session.commit
|
|
|
|
def commit_after_gate() -> None:
|
|
revoker_holds_lock.set()
|
|
if not release_revoker.wait(timeout=20):
|
|
raise RuntimeError("timed out holding the approval revocation")
|
|
original_commit()
|
|
|
|
session.commit = commit_after_gate # type: ignore[method-assign]
|
|
LifecycleService(session).decide_approval(
|
|
approval_id,
|
|
ApprovalDecision(actor="qualification", reason="Revoke before serving promotion"),
|
|
"revoke",
|
|
)
|
|
|
|
def promote() -> str:
|
|
with Session(engine) as session:
|
|
try:
|
|
LifecycleService(session).commit_operation(
|
|
operation_id,
|
|
ApprovalDecision(actor="qualification", reason="Attempt reviewed route change"),
|
|
)
|
|
except LifecycleError as error:
|
|
session.rollback()
|
|
return error.code
|
|
return "COMMITTED"
|
|
|
|
@event.listens_for(engine, "before_cursor_execute")
|
|
def observe_approval_lock(
|
|
_connection: object,
|
|
_cursor: object,
|
|
statement: str,
|
|
_parameters: object,
|
|
_context: object,
|
|
_executemany: bool,
|
|
) -> None:
|
|
if (
|
|
threading.current_thread().name.startswith("promotion-race-promoter")
|
|
and "lifecycle_approval_requests" in statement.lower()
|
|
and "for update" in statement.lower()
|
|
):
|
|
promoter_requested_lock.set()
|
|
|
|
try:
|
|
with ThreadPoolExecutor(
|
|
max_workers=1, thread_name_prefix="promotion-race-revoker"
|
|
) as first:
|
|
revoked = first.submit(revoke)
|
|
if not revoker_holds_lock.wait(timeout=10):
|
|
revoked.result(timeout=1)
|
|
pytest.fail("revoker did not reach the held commit")
|
|
with ThreadPoolExecutor(
|
|
max_workers=1, thread_name_prefix="promotion-race-promoter"
|
|
) as second:
|
|
promoted = second.submit(promote)
|
|
if not promoter_requested_lock.wait(timeout=10):
|
|
promoted.result(timeout=1)
|
|
pytest.fail("promoter did not request the approval lock")
|
|
with pytest.raises(FutureTimeoutError):
|
|
promoted.result(timeout=0.25)
|
|
release_revoker.set()
|
|
revoked.result(timeout=30)
|
|
assert promoted.result(timeout=30) == "APPROVAL_STALE"
|
|
finally:
|
|
release_revoker.set()
|
|
event.remove(engine, "before_cursor_execute", observe_approval_lock)
|
|
|
|
with Session(engine) as session:
|
|
approval = session.get(LifecycleApprovalRequest, approval_id)
|
|
operation = session.get(LifecycleOperation, operation_id)
|
|
contract = session.get(CapabilityContract, contract_id)
|
|
assert approval is not None and approval.status == "REVOKED"
|
|
assert operation is not None and operation.stage == "VERIFYING"
|
|
assert contract is not None and contract.serving_generation == original_generation
|
|
route = ServingRepository(session).stable_deployment(contract_id)
|
|
assert route is not None and route.id == current_id and route.id != candidate_id
|
|
events = list(
|
|
session.scalars(select(LifecycleEvent).where(LifecycleEvent.id.not_in(baseline_events)))
|
|
)
|
|
assert [item.event_type for item in events] == ["APPROVAL_REVOKED"]
|