74 lines
3.1 KiB
Python
74 lines
3.1 KiB
Python
"""Deterministic storage fault injection for the authenticated embedding gateway."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import redis
|
|
from sqlalchemy import create_engine, select
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from modelforge_api.domain.serving import EmbeddingInvokeRequest, ServiceClientCreate
|
|
from modelforge_api.persistence.models import Base, GatewayRequest, ServingGpuLease, ServingJob
|
|
from modelforge_api.services.serving import ServingError
|
|
from tests.test_serving import InlineWorkerServingService, approve_and_promote, setup_m5
|
|
|
|
|
|
def test_redis_payload_write_outage_fails_closed_and_releases_gateway_lease() -> None:
|
|
engine = create_engine(
|
|
"sqlite+pysqlite:///:memory:",
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
)
|
|
Base.metadata.create_all(engine)
|
|
with Session(engine) as session:
|
|
base, candidate, _node = setup_m5(session)
|
|
approve_and_promote(base, candidate.id)
|
|
service = InlineWorkerServingService(
|
|
session, base.settings, base.manifests, base.payload_store
|
|
)
|
|
created = service.create_client(
|
|
ServiceClientCreate(
|
|
name="storage-fault-client",
|
|
allowed_capabilities=["rag.embedding@1"],
|
|
requests_per_minute=10,
|
|
)
|
|
)
|
|
client = service.authenticate(f"Bearer {created.credential}", "rag.embedding@1")
|
|
payload_store = service.payload_store
|
|
original_put = payload_store.put
|
|
|
|
def unavailable(*_args: object, **_kwargs: object) -> str:
|
|
raise redis.ConnectionError("private redis host and credential")
|
|
|
|
payload_store.put = unavailable
|
|
try:
|
|
try:
|
|
service.invoke(EmbeddingInvokeRequest(input="sensitive test input"), client)
|
|
except ServingError as error:
|
|
assert error.status_code == 503
|
|
assert error.code == "PAYLOAD_STORE_UNAVAILABLE"
|
|
assert "private" not in str(error)
|
|
else:
|
|
raise AssertionError("Redis outage must prevent dispatch")
|
|
finally:
|
|
payload_store.put = original_put
|
|
|
|
request = session.scalars(select(GatewayRequest)).one()
|
|
lease = session.scalars(select(ServingGpuLease)).one()
|
|
assert request.status == "failed"
|
|
assert request.failure_code == "PAYLOAD_STORE_UNAVAILABLE"
|
|
assert request.finished_at is not None
|
|
assert request.decision_evidence["input_persisted"] is False
|
|
assert lease.state == "failed"
|
|
assert lease.failure_code == "PAYLOAD_STORE_UNAVAILABLE"
|
|
assert lease.released_at is not None
|
|
assert (
|
|
session.scalars(select(ServingJob).where(ServingJob.operation == "invoke")).all() == []
|
|
)
|
|
assert payload_store.values == {}
|
|
|
|
# Recovery must use normal serving, without a stale lease or hidden fallback.
|
|
recovered = service.invoke(EmbeddingInvokeRequest(input="after recovery"), client)
|
|
assert recovered.dimension == 1024
|
|
assert session.scalars(select(GatewayRequest)).all()[-1].status == "completed"
|