File size: 7,355 Bytes
23d337e | 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 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 | """Failure-scenario tests — verify graceful degradation."""
from __future__ import annotations
import asyncio
import base64
import pytest
from config.settings import Settings
from api.container import build_container
from models.jobs import JobKind, JobRequest
from models.providers import ProviderCapability
from providers.base import BaseProvider, ProviderResult
class FailingProvider(BaseProvider):
"""A provider that always raises an exception."""
name = "failing_provider"
capability = ProviderCapability.DETECTION
def __init__(self, settings=None):
super().__init__(settings=settings)
def is_available(self) -> bool:
return True
def _run(self, pipeline_output):
raise RuntimeError("Intentional failure for testing")
class NotConfiguredProvider(BaseProvider):
"""A provider that reports itself as not available."""
name = "not_configured_provider"
capability = ProviderCapability.DETECTION
def is_available(self) -> bool:
return False
def _run(self, pipeline_output):
raise RuntimeError("Should never be called")
@pytest.fixture
def container_with_failing_provider(test_settings):
container = build_container(test_settings)
# Register a failing provider into the registry
container.registry._providers["failing_provider"] = FailingProvider(settings=test_settings)
container.registry._providers["not_configured_provider"] = NotConfiguredProvider(settings=test_settings)
return container
class TestGracefulDegradation:
def test_failing_provider_does_not_crash_job(self, container_with_failing_provider, sample_image_b64):
"""A failing provider should be captured as failed evidence, not crash the job."""
c = container_with_failing_provider
# Use the failing provider explicitly
req = JobRequest(
kind=JobKind.DETECTION,
image_base64=sample_image_b64,
providers=["failing_provider"],
)
result = asyncio.run(c.detection_service.detect(req))
assert result["success"] is True # The JOB succeeded (with failures in evidence)
report = result["report"]
assert "failing_provider" in report["metadata"]["providers_failed"]
assert len(report["evidence"]) >= 1
# Find the failing provider's evidence
failing_evidence = [e for e in report["evidence"] if e["provider"] == "failing_provider"]
assert len(failing_evidence) == 1
assert failing_evidence[0]["success"] is False
assert "Intentional failure" in failing_evidence[0]["error"]
def test_not_configured_provider_returns_not_configured(self, container_with_failing_provider, sample_image_b64):
"""A provider that's not available should return a NotConfigured result."""
c = container_with_failing_provider
req = JobRequest(
kind=JobKind.DETECTION,
image_base64=sample_image_b64,
providers=["not_configured_provider"],
)
result = asyncio.run(c.detection_service.detect(req))
assert result["success"] is True # JOB succeeds
report = result["report"]
# Should have evidence with the not_configured error
nc_evidence = [e for e in report["evidence"] if e["provider"] == "not_configured_provider"]
assert len(nc_evidence) == 1
assert nc_evidence[0]["success"] is False
assert "NotConfigured" in nc_evidence[0]["error_type"]
def test_mixed_success_and_failure_preserves_both(self, container_with_failing_provider, sample_image_b64):
"""If some providers succeed and others fail, both should be in evidence."""
c = container_with_failing_provider
req = JobRequest(
kind=JobKind.DETECTION,
image_base64=sample_image_b64,
# Both haar (works) and failing_provider (fails) should run
providers=["haar", "failing_provider"],
)
result = asyncio.run(c.detection_service.detect(req))
report = result["report"]
assert "haar" in report["metadata"]["providers_succeeded"]
assert "failing_provider" in report["metadata"]["providers_failed"]
assert len(report["evidence"]) == 2
def test_invalid_image_returns_validation_error(self, container_with_failing_provider):
"""An invalid image should return a structured validation error."""
c = container_with_failing_provider
req = JobRequest(
kind=JobKind.DETECTION,
image_base64="!!!invalid base64!!!",
)
result = asyncio.run(c.detection_service.detect(req))
assert result["success"] is False
assert "error" in result
assert result.get("error_type") == "ValidationError"
def test_missing_image_returns_validation_error(self, container_with_failing_provider):
"""No image input should return a validation error."""
c = container_with_failing_provider
req = JobRequest(kind=JobKind.DETECTION)
result = asyncio.run(c.detection_service.detect(req))
assert result["success"] is False
assert result.get("error_type") == "ValidationError"
class TestCircuitBreaker:
def test_circuit_opens_after_threshold_failures(self, test_settings, sample_image_b64):
"""After N consecutive failures, the circuit should open."""
# Lower threshold for test
test_settings.circuit_breaker_failure_threshold = 3
test_settings.circuit_breaker_recovery_seconds = 60
container = build_container(test_settings)
container.registry._providers["failing_provider"] = FailingProvider(settings=test_settings)
# Trigger failures
for _ in range(3):
req = JobRequest(
kind=JobKind.DETECTION,
image_base64=sample_image_b64,
providers=["failing_provider"],
)
asyncio.run(container.detection_service.detect(req))
# Circuit should now be open
assert container.health_monitor.is_available("failing_provider") is False
def test_circuit_blocks_subsequent_invocations(self, test_settings, sample_image_b64):
"""When circuit is open, the orchestrator should skip the provider."""
test_settings.circuit_breaker_failure_threshold = 2
test_settings.circuit_breaker_recovery_seconds = 60
container = build_container(test_settings)
container.registry._providers["failing_provider"] = FailingProvider(settings=test_settings)
# Trip the breaker
for _ in range(2):
req = JobRequest(
kind=JobKind.DETECTION,
image_base64=sample_image_b64,
providers=["failing_provider"],
)
asyncio.run(container.detection_service.detect(req))
# Now run again — provider should be skipped (no new evidence)
req = JobRequest(
kind=JobKind.DETECTION,
image_base64=sample_image_b64,
providers=["failing_provider"],
)
result = asyncio.run(container.detection_service.detect(req))
report = result["report"]
# When circuit is open, provider is skipped — so providers_invoked is empty
assert "failing_provider" not in report["metadata"]["providers_invoked"]
|