423 lines
23 KiB
Python
423 lines
23 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine, select, update
|
|
from sqlalchemy.orm import Session
|
|
|
|
from modelforge_api.domain.evaluation import (
|
|
EvaluationCaseCreate,
|
|
EvaluationRevisionCreate,
|
|
EvaluationSuiteCreate,
|
|
)
|
|
from modelforge_api.domain.lifecycle_contracts import (
|
|
ApprovalDecision,
|
|
ApprovalRequestCreate,
|
|
LifecycleEvidenceBundle,
|
|
LifecycleSubjectCreate,
|
|
PlanExecutionCreate,
|
|
PromotionPlanCreate,
|
|
ServingTransitionSelection,
|
|
)
|
|
from modelforge_api.domain.selected_model_comparison import ComparisonCreate
|
|
from modelforge_api.domain.serving import EmbeddingInvokeRequest, ServiceClientCreate
|
|
from modelforge_api.persistence.models import (
|
|
ArtifactSet,
|
|
Base,
|
|
CapabilityContract,
|
|
CapabilityDeployment,
|
|
CapabilityExperimentRoute,
|
|
CapabilityResourceEnvelope,
|
|
DeploymentCandidate,
|
|
EmbeddingSpace,
|
|
GatewayRequest,
|
|
LifecycleApprovalRequest,
|
|
LifecycleEvent,
|
|
LifecycleOperation,
|
|
LifecycleRetentionRecord,
|
|
LifecycleSubject,
|
|
Project,
|
|
ProjectBinding,
|
|
ProjectFitEvidence,
|
|
ResidencyAllocation,
|
|
RuntimeProfile,
|
|
SelectedModelComparison,
|
|
ServiceClient,
|
|
)
|
|
from modelforge_api.services.audit import AuditContext
|
|
from modelforge_api.services.lifecycle import LifecycleError, LifecycleService
|
|
from modelforge_api.services.lifecycle_serving import LifecycleServing, digest
|
|
from modelforge_api.services.selected_model_comparison import SelectedModelComparisonService
|
|
from modelforge_api.services.serving import ServingError
|
|
from tests.test_selected_model_comparison import _claimed
|
|
from tests.test_serving import InlineWorkerServingService, approve_and_promote, setup_m5
|
|
|
|
|
|
def _clone(model, row, **changes):
|
|
return model(**{column.name: getattr(row, column.name) for column in model.__table__.columns
|
|
if column.name not in {"id", "created_at", "updated_at"}} | changes)
|
|
|
|
|
|
@pytest.fixture
|
|
def serving_pair():
|
|
engine = create_engine("sqlite://")
|
|
Base.metadata.create_all(engine)
|
|
with Session(engine) as session:
|
|
yield build_pair(session)
|
|
engine.dispose()
|
|
|
|
|
|
def build_pair(session):
|
|
serving, candidate_ref, _ = setup_m5(session)
|
|
production_approval, deployed = approve_and_promote(serving, candidate_ref.id)
|
|
current = session.get(CapabilityDeployment, deployed.id)
|
|
candidate = _clone(CapabilityDeployment, current, config_fingerprint="b" * 64,
|
|
channel="experiment", status="candidate", production=False, production_approval_id=None,
|
|
promoted_at=None)
|
|
session.add(candidate)
|
|
session.flush()
|
|
envelope = session.scalar(select(CapabilityResourceEnvelope).where(
|
|
CapabilityResourceEnvelope.capability_deployment_id == current.id))
|
|
next_envelope = _clone(CapabilityResourceEnvelope, envelope, capability_deployment_id=candidate.id)
|
|
session.add(next_envelope)
|
|
residency = session.scalar(select(ResidencyAllocation).where(
|
|
ResidencyAllocation.capability_deployment_id == current.id))
|
|
session.add(_clone(ResidencyAllocation, residency, capability_deployment_id=candidate.id))
|
|
project = Project(key="selected-owner", name="Owner", description="test")
|
|
session.add(project)
|
|
session.flush()
|
|
binding = ProjectBinding(project_id=project.id, capability_contract_id=current.capability_contract_id,
|
|
channel="stable", priority="production")
|
|
session.add(binding)
|
|
session.flush()
|
|
client = ServiceClient(name="selected-consumer", allowed_capabilities=["rag.embedding@1"],
|
|
project_binding_id=binding.id)
|
|
route = CapabilityExperimentRoute(route_key="selected-test", capability_contract_id=current.capability_contract_id,
|
|
capability_deployment_id=candidate.id, purpose="Bounded comparison", evidence={})
|
|
session.add_all((client, route))
|
|
session.commit()
|
|
from modelforge_api.services.evaluation import EvaluationService
|
|
|
|
relevant = uuid4()
|
|
suite = EvaluationService(session).create_suite(EvaluationSuiteCreate(project_id=project.id,
|
|
key="selected", name="Selected", description="One case", revision=EvaluationRevisionCreate(
|
|
revision="v1", dataset_revision="selected-v1", retrieval_settings={"top_k": 10}, cases=[
|
|
EvaluationCaseCreate(case_key="one", query="Exact unchanged query", relevant_chunk_ids=[relevant],
|
|
critical=True, review_status="reviewed", label_provenance={"source": "review", "corpus_revision": "a" * 64})])))
|
|
comparisons = SelectedModelComparisonService(session, AuditContext("operator", "test"))
|
|
request = ComparisonCreate(project_binding_id=binding.id, consumer_client_id=client.id,
|
|
suite_revision_id=suite.latest_revision_id, current_deployment_id=current.id,
|
|
candidate_deployment_id=candidate.id, candidate_route_id=route.id, corpus_sha256="a" * 64,
|
|
retrieval_config_sha256=digest({"top_k": 10}), current_index_ref="current-index",
|
|
candidate_index_ref="isolated-test-index", idempotency_key="selected-one")
|
|
claim, result, _ = _claimed((session, comparisons, request, client, relevant))
|
|
completed = comparisons.result(claim.comparison.id, result, client)
|
|
contract = session.get(CapabilityContract, current.capability_contract_id)
|
|
selection = ServingTransitionSelection(current_deployment_id=current.id, candidate_deployment_id=candidate.id,
|
|
production_approval_id=production_approval.id, expected_route_generation=contract.serving_generation,
|
|
comparison_ids={binding.id: completed.id})
|
|
artifact = session.get(ArtifactSet, candidate.artifact_set_id)
|
|
profile = session.get(RuntimeProfile, candidate.runtime_profile_id)
|
|
source = session.get(DeploymentCandidate, candidate.deployment_candidate_id)
|
|
evidence = LifecycleEvidenceBundle(serving_transition=selection,
|
|
artifact_set_id=artifact.id, model_revision_id=artifact.revision_id,
|
|
runtime_profile_id=profile.id, runtime_probe_id=source.runtime_probe_id,
|
|
runtime_image_digest=profile.runtime_image_digest, capability_deployment_id=candidate.id,
|
|
production_execution_approval_id=production_approval.id, resource_envelope_id=next_envelope.id,
|
|
embedding_space_id=candidate.embedding_space_id, rollback_target_ref=str(current.id),
|
|
evaluation_run_ids=[completed.baseline_run_id, completed.candidate_run_id],
|
|
integrity="VERIFIED", security="APPROVED", license="APPROVED", runtime="PROVEN",
|
|
evaluation="PROMOTION_ELIGIBLE", production_validation="SATISFIED",
|
|
scheduler_readiness="READY", evidence_revisions={"scheduler": "fixture"})
|
|
return session, serving, current, candidate, selection, evidence
|
|
|
|
|
|
def approved_plan(pair):
|
|
session, _, current, candidate, _, evidence = pair
|
|
service = LifecycleService(session)
|
|
subject = service.create_subject(LifecycleSubjectCreate(target_type="CAPABILITY_DEPLOYMENT",
|
|
target_ref=str(candidate.id), environment="PRODUCTION", state="PROMOTION_ELIGIBLE"))
|
|
approval = service.request_approval(ApprovalRequestCreate(policy_key="capability-production",
|
|
subject_id=subject.id, target_type=subject.target_type, target_ref=subject.target_ref,
|
|
environment="PRODUCTION", requested_transition="STABLE", evidence=evidence,
|
|
requested_by="owner", reason="Review exact pair and all affected projects"))
|
|
assert approval.status == "PENDING", approval.blockers
|
|
service.decide_approval(approval.id, ApprovalDecision(actor="owner", reason="Evidence reviewed for every project"), "approve")
|
|
plan = service.create_plan(PromotionPlanCreate(approval_request_id=approval.id, subject_id=subject.id,
|
|
desired_state="STABLE", migration_class="transparent", candidate_deployment_id=candidate.id,
|
|
rollback_target_ref=str(current.id), created_by="owner"))
|
|
plan = service.approve_plan(plan.id, ApprovalDecision(actor="owner", reason="Exact immutable transition approved"))
|
|
operation = service.execute_plan(plan.id, PlanExecutionCreate(executor="owner", idempotency_key="execute-one",
|
|
expected_subject_version=subject.version))
|
|
assert operation.stage == "VERIFYING"
|
|
return service, plan, operation, subject
|
|
|
|
|
|
def test_commit_and_rollback_switch_real_route_and_retain_exact_assets(serving_pair):
|
|
session, serving, current, candidate, selection, _ = serving_pair
|
|
before = (LifecycleServing.route_snapshot(current), LifecycleServing.route_snapshot(candidate))
|
|
service, plan, operation, subject = approved_plan(serving_pair)
|
|
assert serving.repo.stable_deployment(current.capability_contract_id).id == current.id
|
|
result = service.commit_operation(operation.id, ApprovalDecision(actor="owner", reason="Explicit permanent replacement"))
|
|
assert result.stage == "COMMITTED"
|
|
assert service.commit_operation(operation.id, ApprovalDecision(actor="owner", reason="Idempotent committed result retry")).stage == "COMMITTED"
|
|
assert serving.repo.stable_deployment(current.capability_contract_id).id == candidate.id
|
|
assert current.status == "draining" and candidate.production is True
|
|
assert session.get(LifecycleSubject, subject.id).state == "STABLE"
|
|
assert session.get(CapabilityContract, current.capability_contract_id).serving_generation == selection.expected_route_generation + 1
|
|
retained = list(session.scalars(select(LifecycleRetentionRecord)))
|
|
assert any(item.target_type == "ARTIFACT_SET" and item.target_ref == str(current.artifact_set_id) for item in retained)
|
|
assert plan.project_consumers == [session.get(ProjectBinding, next(iter(selection.comparison_ids))).project_id.__str__()]
|
|
rolled = service.rollback_operation(operation.id, ApprovalDecision(actor="owner", reason="Restore exact previous serving route"))
|
|
assert rolled.stage == "ROLLED_BACK"
|
|
assert serving.repo.stable_deployment(current.capability_contract_id).id == current.id
|
|
assert (LifecycleServing.route_snapshot(current), LifecycleServing.route_snapshot(candidate)) == before
|
|
assert session.get(CapabilityContract, current.capability_contract_id).serving_generation == selection.expected_route_generation + 2
|
|
assert service.rollback_operation(operation.id, ApprovalDecision(actor="owner", reason="Idempotent rollback repeat")).stage == "ROLLED_BACK"
|
|
|
|
|
|
def test_commit_and_rollback_route_live_gateway_requests_to_exact_deployments(serving_pair):
|
|
session, serving, current, candidate, _, _ = serving_pair
|
|
gateway = InlineWorkerServingService(
|
|
session,
|
|
serving.settings,
|
|
serving.manifests,
|
|
serving.payload_store,
|
|
)
|
|
created = gateway.create_client(
|
|
ServiceClientCreate(
|
|
name="lifecycle-route-proof",
|
|
allowed_capabilities=["rag.embedding@1"],
|
|
)
|
|
)
|
|
service, _, operation, _ = approved_plan(serving_pair)
|
|
|
|
service.commit_operation(
|
|
operation.id,
|
|
ApprovalDecision(actor="owner", reason="Route proof commits the reviewed replacement"),
|
|
)
|
|
candidate_client = gateway.authenticate(
|
|
f"Bearer {created.credential}", "rag.embedding@1"
|
|
)
|
|
candidate_response = gateway.invoke(
|
|
EmbeddingInvokeRequest(input="request after replacement"), candidate_client
|
|
)
|
|
candidate_request = session.scalar(
|
|
select(GatewayRequest).where(GatewayRequest.request_id == candidate_response.request_id)
|
|
)
|
|
assert candidate_response.embedding_space_id == candidate.embedding_space_id
|
|
assert candidate_request is not None
|
|
assert candidate_request.capability_deployment_id == candidate.id
|
|
|
|
service.rollback_operation(
|
|
operation.id,
|
|
ApprovalDecision(actor="owner", reason="Route proof restores the exact prior deployment"),
|
|
)
|
|
current_client = gateway.authenticate(f"Bearer {created.credential}", "rag.embedding@1")
|
|
current_response = gateway.invoke(
|
|
EmbeddingInvokeRequest(input="request after rollback"), current_client
|
|
)
|
|
current_request = session.scalar(
|
|
select(GatewayRequest).where(GatewayRequest.request_id == current_response.request_id)
|
|
)
|
|
assert current_response.embedding_space_id == current.embedding_space_id
|
|
assert current_request is not None
|
|
assert current_request.capability_deployment_id == current.id
|
|
|
|
|
|
@pytest.mark.parametrize("mutation", ["generation", "bindings", "revoked", "comparison", "envelope"])
|
|
def test_changed_authority_or_route_blocks_commit(serving_pair, mutation):
|
|
session, serving, current, candidate, selection, _ = serving_pair
|
|
service, _, operation, _ = approved_plan(serving_pair)
|
|
if mutation == "generation":
|
|
session.get(CapabilityContract, current.capability_contract_id).serving_generation += 1
|
|
elif mutation == "bindings":
|
|
project = Project(key="second", name="Second", description="test")
|
|
session.add(project)
|
|
session.flush()
|
|
session.add(ProjectBinding(project_id=project.id, capability_contract_id=current.capability_contract_id,
|
|
channel="stable", priority="production"))
|
|
elif mutation == "revoked":
|
|
serving.repo.approval(selection.production_approval_id).revoked_at = datetime.now(UTC)
|
|
elif mutation == "comparison":
|
|
session.get(SelectedModelComparison, next(iter(selection.comparison_ids.values()))).status = "failed"
|
|
else:
|
|
session.scalar(select(CapabilityResourceEnvelope).where(
|
|
CapabilityResourceEnvelope.capability_deployment_id == candidate.id)).stale = True
|
|
session.commit()
|
|
with pytest.raises((LifecycleError, ServingError)):
|
|
service.commit_operation(operation.id, ApprovalDecision(actor="owner", reason="Must reject changed authority"))
|
|
session.rollback()
|
|
assert serving.repo.stable_deployment(current.capability_contract_id).id == current.id
|
|
assert session.get(LifecycleOperation, operation.id).stage == "VERIFYING"
|
|
|
|
|
|
def test_equal_dimension_different_space_requires_consumer_cutover(serving_pair):
|
|
session, _, current, candidate, selection, _ = serving_pair
|
|
old_space = session.get(EmbeddingSpace, current.embedding_space_id)
|
|
space = _clone(EmbeddingSpace, old_space, identity_digest="d" * 64)
|
|
session.add(space)
|
|
session.flush()
|
|
candidate.embedding_space_id = space.id
|
|
session.commit()
|
|
with pytest.raises(LifecycleError, match="consumer reindex"):
|
|
LifecycleServing(session).prepare(selection)
|
|
|
|
|
|
@pytest.mark.parametrize("mutation", ["generation", "binding", "config"])
|
|
def test_rollback_preserves_later_route_or_configuration(serving_pair, mutation):
|
|
session, serving, current, candidate, _, _ = serving_pair
|
|
service, _, operation, _ = approved_plan(serving_pair)
|
|
service.commit_operation(operation.id, ApprovalDecision(actor="owner", reason="Explicit serving replacement"))
|
|
if mutation == "generation":
|
|
# Even an A->B->A route cycle must fence the old operation.
|
|
session.get(CapabilityContract, current.capability_contract_id).serving_generation += 2
|
|
elif mutation == "binding":
|
|
session.scalar(select(ProjectBinding).where(ProjectBinding.capability_contract_id == current.capability_contract_id)).priority = "background"
|
|
else:
|
|
candidate.max_queue_depth += 1
|
|
session.commit()
|
|
with pytest.raises(LifecycleError, match="preserved|changed"):
|
|
service.rollback_operation(operation.id, ApprovalDecision(actor="owner", reason="Old operation cannot overwrite later state"))
|
|
session.rollback()
|
|
assert serving.repo.stable_deployment(current.capability_contract_id).id == candidate.id
|
|
assert session.get(LifecycleOperation, operation.id).stage == "COMMITTED"
|
|
|
|
|
|
def test_subject_conflict_rolls_back_both_route_and_generation(serving_pair):
|
|
session, serving, current, _, selection, _ = serving_pair
|
|
service, _, operation, subject = approved_plan(serving_pair)
|
|
session.execute(update(LifecycleSubject).where(LifecycleSubject.id == subject.id).values(version=100))
|
|
session.commit()
|
|
with pytest.raises(LifecycleError, match="subject changed"):
|
|
service.commit_operation(operation.id, ApprovalDecision(actor="owner", reason="Reject concurrent subject transition"))
|
|
assert serving.repo.stable_deployment(current.capability_contract_id).id == current.id
|
|
assert session.get(CapabilityContract, current.capability_contract_id).serving_generation == selection.expected_route_generation
|
|
|
|
|
|
def test_revoked_lifecycle_approval_cannot_commit(serving_pair):
|
|
session, serving, current, _, _, _ = serving_pair
|
|
service, plan, operation, _ = approved_plan(serving_pair)
|
|
session.get(LifecycleApprovalRequest, plan.approval_request_id).status = "REVOKED"
|
|
session.commit()
|
|
with pytest.raises(LifecycleError, match="no longer current"):
|
|
service.commit_operation(operation.id, ApprovalDecision(actor="owner", reason="Reject revoked lifecycle approval"))
|
|
assert serving.repo.stable_deployment(current.capability_contract_id).id == current.id
|
|
|
|
|
|
@pytest.mark.parametrize("mutation", ["revoked", "quarantined", "offline"])
|
|
def test_rollback_cannot_reactivate_invalid_prior_production_authority(serving_pair, mutation):
|
|
session, serving, current, candidate, _, _ = serving_pair
|
|
service, _, operation, _ = approved_plan(serving_pair)
|
|
service.commit_operation(operation.id, ApprovalDecision(actor="owner", reason="Explicit serving replacement"))
|
|
if mutation == "revoked":
|
|
serving.repo.approval(current.production_approval_id).revoked_at = datetime.now(UTC)
|
|
elif mutation == "quarantined":
|
|
serving.repo.set_artifacts(current.artifact_set_id)[0].quarantined = True
|
|
else:
|
|
serving.repo.node(current.compute_node_id).liveness_state = "offline"
|
|
session.commit()
|
|
with pytest.raises((LifecycleError, ServingError)):
|
|
service.rollback_operation(operation.id, ApprovalDecision(actor="owner", reason="Refuse a revoked rollback target"))
|
|
session.rollback()
|
|
assert serving.repo.stable_deployment(current.capability_contract_id).id == candidate.id
|
|
|
|
|
|
@pytest.mark.parametrize("validation", ["DEFERRED_EXTERNAL_VALIDATION", "REQUIRED", "NOT_REQUIRED"])
|
|
def test_actual_serving_approval_requires_satisfied_validation(serving_pair, validation):
|
|
serving_pair[-1].production_validation = validation
|
|
with pytest.raises(LifecycleError, match="satisfied production validation"):
|
|
approved_plan(serving_pair)
|
|
|
|
|
|
def test_deferred_historical_approval_cannot_commit_actual_route(serving_pair):
|
|
session, serving, current, _, _, _ = serving_pair
|
|
service, plan, operation, _ = approved_plan(serving_pair)
|
|
approval = session.get(LifecycleApprovalRequest, plan.approval_request_id)
|
|
# Simulate a previously accepted capability-only snapshot; its declared deferred
|
|
# validation must not gain operational authority under the new serving adapter.
|
|
approval.evidence_snapshot = {**approval.evidence_snapshot, "declared": {
|
|
**approval.evidence_snapshot["declared"], "production_validation": "DEFERRED_EXTERNAL_VALIDATION"}}
|
|
session.commit()
|
|
with pytest.raises(LifecycleError, match="no longer current"):
|
|
service.commit_operation(operation.id, ApprovalDecision(actor="owner", reason="Reject explicitly deferred validation"))
|
|
session.rollback()
|
|
assert serving.repo.stable_deployment(current.capability_contract_id).id == current.id
|
|
|
|
|
|
def _fit(pair, *, deployment_id=None, **changes):
|
|
session, _, _, candidate, selection, _ = pair
|
|
values = {"project_binding_id": next(iter(selection.comparison_ids)),
|
|
"capability_deployment_id": deployment_id or candidate.id, "environment": "production",
|
|
"recommendation": "PROMOTION_ELIGIBLE", "evidence_class": "OWNER_PHOTO",
|
|
"engineering_integration": "PASS", "production_validation": "SATISFIED", "production_action": "NONE",
|
|
"case_count": 1, "metric_values": {}, "critical_errors": 0, "blockers": [], "evidence_digest": uuid4().hex * 2,
|
|
"evidence": {"source": "offline test fixture"}, "created_at": datetime.now(UTC)}
|
|
row = ProjectFitEvidence(**(values | changes))
|
|
session.add(row)
|
|
session.commit()
|
|
return row
|
|
|
|
|
|
@pytest.mark.parametrize("changes", [
|
|
{"recommendation": "KEEP_LAB", "production_validation": "DEFERRED_EXTERNAL_VALIDATION"},
|
|
{"recommendation": "BLOCKED", "production_validation": "REQUIRED"},
|
|
{"recommendation": "REQUIRES_MORE_EVIDENCE"},
|
|
{"engineering_integration": "BLOCKED"},
|
|
{"blockers": ["EXTERNAL_VALIDATION_REQUIRED"]},
|
|
{"critical_errors": 1},
|
|
])
|
|
def test_later_negative_exact_project_fit_blocks_commit(serving_pair, changes):
|
|
session, serving, current, _, _, _ = serving_pair
|
|
service, _, operation, _ = approved_plan(serving_pair)
|
|
_fit(serving_pair, **changes)
|
|
with pytest.raises(LifecycleError, match="no longer current"):
|
|
service.commit_operation(operation.id, ApprovalDecision(actor="owner", reason="Respect later project-specific blocker"))
|
|
session.rollback()
|
|
assert serving.repo.stable_deployment(current.capability_contract_id).id == current.id
|
|
|
|
|
|
def test_project_fit_is_exactly_scoped_and_bound_into_plan(serving_pair):
|
|
_, _, current, candidate, selection, _ = serving_pair
|
|
_fit(serving_pair, deployment_id=current.id, recommendation="BLOCKED")
|
|
fit = _fit(serving_pair)
|
|
service, plan, operation, _ = approved_plan(serving_pair)
|
|
contract = plan.affected_identities["serving_transition"]
|
|
assert contract["production_validation"] == "SATISFIED"
|
|
assert contract["comparisons"][str(next(iter(selection.comparison_ids)))]["project_fit"]["id"] == str(fit.id)
|
|
service.commit_operation(operation.id, ApprovalDecision(actor="owner", reason="Exact candidate project validation reviewed"))
|
|
assert candidate.status == "stable"
|
|
|
|
|
|
def test_newer_positive_fit_still_requires_reviewed_plan_refresh(serving_pair):
|
|
session, serving, current, _, _, _ = serving_pair
|
|
_fit(serving_pair)
|
|
service, _, operation, _ = approved_plan(serving_pair)
|
|
_fit(serving_pair)
|
|
with pytest.raises(LifecycleError, match="no longer current"):
|
|
service.commit_operation(operation.id, ApprovalDecision(actor="owner", reason="New evidence needs a new review"))
|
|
session.rollback()
|
|
assert serving.repo.stable_deployment(current.capability_contract_id).id == current.id
|
|
|
|
|
|
@pytest.mark.parametrize("read_before_revoke", [False, True])
|
|
def test_committed_serving_approval_remains_readable_and_revocable(serving_pair, read_before_revoke):
|
|
session, serving, current, candidate, _, _ = serving_pair
|
|
service, plan, operation, _ = approved_plan(serving_pair)
|
|
service.commit_operation(operation.id, ApprovalDecision(actor="owner", reason="Explicit permanent replacement"))
|
|
if read_before_revoke:
|
|
historical = service.approval(plan.approval_request_id)
|
|
assert historical.status == "STALE"
|
|
assert "SERVING_ROUTE_CHANGED" in historical.blockers
|
|
revoked = service.decide_approval(plan.approval_request_id,
|
|
ApprovalDecision(actor="owner", reason="Withdraw obsolete replacement authority"), "revoke")
|
|
assert revoked.status == "REVOKED"
|
|
assert service.approval(plan.approval_request_id).status == "REVOKED"
|
|
events = list(session.scalars(select(LifecycleEvent).where(
|
|
LifecycleEvent.event_type == "APPROVAL_STALE")))
|
|
assert len(events) == 1
|
|
assert "SERVING_ROUTE_CHANGED" in str(events[0].details)
|
|
assert serving.repo.stable_deployment(current.capability_contract_id).id == candidate.id
|