Spaces:
Running
Running
File size: 4,562 Bytes
b1e14ec | 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 | from unittest.mock import MagicMock
import pytest
from adapters.inference.capability_registry import CapabilityRegistry
from adapters.inference.fallback_adapter import FallbackInferenceAdapter
from pydantic import BaseModel
class User(BaseModel):
name: str
age: int
def test_fallback_generate_structured_success():
mock_adapter1 = MagicMock()
mock_adapter1.generate_structured.side_effect = Exception("Failed")
mock_adapter2 = MagicMock()
mock_user = User(name="Test User", age=25)
mock_adapter2.generate_structured.return_value = mock_user
fallback = FallbackInferenceAdapter(adapters=[mock_adapter1, mock_adapter2])
result = fallback.generate_structured(
prompt="Extract user info", response_model=User
)
assert result == mock_user
mock_adapter1.generate_structured.assert_called_once()
mock_adapter2.generate_structured.assert_called_once()
def test_fallback_generate_structured_all_failed():
mock_adapter1 = MagicMock()
mock_adapter1.generate_structured.side_effect = Exception("Failed 1")
mock_adapter2 = MagicMock()
mock_adapter2.generate_structured.side_effect = Exception("Failed 2")
fallback = FallbackInferenceAdapter(adapters=[mock_adapter1, mock_adapter2])
with pytest.raises(Exception) as excinfo:
fallback.generate_structured(prompt="Extract user info", response_model=User)
assert "Tous les adaptateurs ont échoué" in str(excinfo.value)
from core.ports.inference_port import InferencePort # noqa: E402
class MockCapableAdapter(InferencePort):
def estimate_depth(self, image_data: bytes) -> bytes:
return b"depth_map"
def health_check(self) -> dict:
return {"status": "online"}
def generate(self, *args, **kwargs):
raise NotImplementedError
def stream_generate(self, *args, **kwargs):
raise NotImplementedError
def get_text_embedding(self, *args, **kwargs):
raise NotImplementedError
class MockGenericAdapter(InferencePort):
def health_check(self) -> dict:
return {"status": "online"}
def generate(self, *args, **kwargs):
raise NotImplementedError
def stream_generate(self, *args, **kwargs):
raise NotImplementedError
def get_text_embedding(self, *args, **kwargs):
raise NotImplementedError
def test_fallback_introspection_capability_mapping():
adapter1 = MockGenericAdapter()
adapter2 = MockCapableAdapter()
fallback = FallbackInferenceAdapter(adapters=[adapter1, adapter2])
# Introspection checks
assert CapabilityRegistry.is_method_overridden(adapter2, "estimate_depth") is True
assert CapabilityRegistry.is_method_overridden(adapter1, "estimate_depth") is False
# Capability cache verification
capable = fallback._capabilities.for_method("estimate_depth")
assert adapter2 in capable
assert adapter1 not in capable
def test_fallback_call_routes_directly_to_capable_adapters():
adapter1 = MockGenericAdapter()
# Mocking standard Port methods that generic adapter should not execute
adapter1.estimate_depth = MagicMock(side_effect=Exception("Should not be called"))
adapter2 = MockCapableAdapter()
adapter2.estimate_depth = MagicMock(return_value=b"correct_depth")
fallback = FallbackInferenceAdapter(adapters=[adapter1, adapter2])
result = fallback.estimate_depth(b"sample_image")
assert result == b"correct_depth"
adapter1.estimate_depth.assert_not_called()
adapter2.estimate_depth.assert_called_once_with(b"sample_image")
from unittest.mock import patch # noqa: E402
from core.ports.inference_port import InferenceNotImplementedError # noqa: E402
def test_not_implemented_exception_is_silent_and_does_not_log_error():
mock_obs = MagicMock()
adapter1 = MockCapableAdapter()
adapter1.estimate_depth = MagicMock(
side_effect=InferenceNotImplementedError("Dynamic override disabled")
)
adapter2 = MockCapableAdapter()
adapter2.estimate_depth = MagicMock(return_value=b"correct_depth_2")
fallback = FallbackInferenceAdapter(
adapters=[adapter1, adapter2], obs_service=mock_obs
)
with patch("adapters.inference.fallback_adapter.logger.error") as mock_log_err:
result = fallback.estimate_depth(b"sample_image")
assert result == b"correct_depth_2"
# Verify NO error log was written and NO observability error was recorded for the NotImplemented event
mock_log_err.assert_not_called()
mock_obs.log_error.assert_not_called()
|