From 27227aa31f3441343332205739dd63f99056ca37 Mon Sep 17 00:00:00 2001 From: huanghuoguoguo <1051233107@qq.com> Date: Fri, 8 May 2026 13:55:28 +0800 Subject: [PATCH] feat(test): add fake provider factory Add tests/factories/provider.py with: - FakeProvider: deterministic fake LLM provider - Error simulation: timeout, auth, rate-limit, malformed - Request capture for assertions - fake_model: mock model with attached provider Co-Authored-By: Claude Opus 4.7 --- tests/factories/__init__.py | 21 ++++ tests/factories/provider.py | 224 ++++++++++++++++++++++++++++++++++++ 2 files changed, 245 insertions(+) create mode 100644 tests/factories/provider.py diff --git a/tests/factories/__init__.py b/tests/factories/__init__.py index a2963799..aa87c688 100644 --- a/tests/factories/__init__.py +++ b/tests/factories/__init__.py @@ -27,6 +27,17 @@ from tests.factories.message import ( command_query, mention_query, empty_query, + image_query, +) +from tests.factories.provider import ( + FakeProvider, + fake_provider, + fake_provider_pong, + fake_provider_timeout, + fake_provider_auth_error, + fake_provider_rate_limit, + fake_provider_malformed, + fake_model, ) __all__ = [ @@ -45,4 +56,14 @@ __all__ = [ "command_query", "mention_query", "empty_query", + "image_query", + # Provider + "FakeProvider", + "fake_provider", + "fake_provider_pong", + "fake_provider_timeout", + "fake_provider_auth_error", + "fake_provider_rate_limit", + "fake_provider_malformed", + "fake_model", ] \ No newline at end of file diff --git a/tests/factories/provider.py b/tests/factories/provider.py new file mode 100644 index 00000000..7895cd2c --- /dev/null +++ b/tests/factories/provider.py @@ -0,0 +1,224 @@ +""" +Fake provider factory for tests. + +Provides a deterministic fake provider that simulates LLM responses without real API calls. +""" + +from __future__ import annotations + +from unittest.mock import AsyncMock, Mock +import typing + +import langbot_plugin.api.entities.builtin.provider.message as provider_message + + +class FakeProvider: + """Deterministic fake provider for unit and integration tests. + + Simulates various provider behaviors: + - Normal text response + - Streaming response + - Timeout error + - Auth error + - Rate-limit error + - Malformed response + + Does not call real LLM vendors. + Does not require API keys. + """ + + PONG_RESPONSE = "LANGBOT_FAKE_PONG" + + def __init__( + self, + *, + default_response: str = "fake response", + streaming_chunks: list[str] = None, + raise_error: Exception = None, + captured_requests: list = None, + ): + self._default_response = default_response + self._streaming_chunks = streaming_chunks or ["fake ", "response"] + self._raise_error = raise_error + self._captured_requests = captured_requests if captured_requests is not None else [] + + def returns(self, text: str) -> "FakeProvider": + """Configure provider to return a specific text response.""" + self._default_response = text + self._streaming_chunks = [text] + return self + + def returns_streaming(self, chunks: list[str]) -> "FakeProvider": + """Configure provider to return streaming chunks.""" + self._streaming_chunks = chunks + self._default_response = "".join(chunks) + return self + + def raises(self, error: Exception) -> "FakeProvider": + """Configure provider to raise an error.""" + self._raise_error = error + return self + + def timeout(self) -> "FakeProvider": + """Configure provider to simulate timeout.""" + return self.raises(TimeoutError("Provider timeout")) + + def auth_error(self) -> "FakeProvider": + """Configure provider to simulate auth error.""" + return self.raises(Exception("Invalid API key")) + + def rate_limit(self) -> "FakeProvider": + """Configure provider to simulate rate limit.""" + return self.raises(Exception("Rate limit exceeded")) + + def malformed(self) -> "FakeProvider": + """Configure provider to simulate malformed response.""" + self._default_response = None + return self + + def get_captured_requests(self) -> list: + """Get all captured request arguments for assertions.""" + return self._captured_requests.copy() + + def clear_captured_requests(self): + """Clear captured requests.""" + self._captured_requests.clear() + + def _create_message(self, content: str) -> provider_message.Message: + """Create a provider message from text content.""" + return provider_message.Message( + role="assistant", + content=content, + ) + + def _create_chunk( + self, + content: str, + is_final: bool = False, + msg_sequence: int = 0, + ) -> provider_message.MessageChunk: + """Create a provider message chunk.""" + return provider_message.MessageChunk( + role="assistant", + content=content, + is_final=is_final, + msg_sequence=msg_sequence, + ) + + async def invoke_llm( + self, + query, + model, + messages: list, + funcs: list, + extra_args: dict, + remove_think: bool = False, + ) -> provider_message.Message: + """Simulate non-streaming LLM invocation.""" + # Capture request for assertions + self._captured_requests.append({ + "query_id": query.query_id if query else None, + "model": model.model_entity.name if model and hasattr(model, 'model_entity') else None, + "messages": messages, + "funcs": funcs, + "extra_args": extra_args, + }) + + # Simulate error if configured + if self._raise_error: + raise self._raise_error + + # Return response + if self._default_response is None: + # Malformed response + return provider_message.Message(role="assistant", content=None) + + return self._create_message(self._default_response) + + async def invoke_llm_stream( + self, + query, + model, + messages: list, + funcs: list, + extra_args: dict, + remove_think: bool = False, + ) -> typing.AsyncGenerator[provider_message.MessageChunk, None]: + """Simulate streaming LLM invocation.""" + # Capture request for assertions + self._captured_requests.append({ + "query_id": query.query_id if query else None, + "model": model.model_entity.name if model and hasattr(model, 'model_entity') else None, + "messages": messages, + "funcs": funcs, + "extra_args": extra_args, + "streaming": True, + }) + + # Simulate error if configured + if self._raise_error: + raise self._raise_error + + # Yield chunks + for i, chunk in enumerate(self._streaming_chunks): + is_final = (i == len(self._streaming_chunks) - 1) + yield self._create_chunk(chunk, is_final=is_final, msg_sequence=i) + + +def fake_provider( + default_response: str = "fake response", +) -> FakeProvider: + """Create a FakeProvider with optional default response.""" + return FakeProvider(default_response=default_response) + + +def fake_provider_pong() -> FakeProvider: + """Create a FakeProvider that returns the pong response.""" + return FakeProvider(default_response=FakeProvider.PONG_RESPONSE) + + +def fake_provider_timeout() -> FakeProvider: + """Create a FakeProvider that simulates timeout.""" + return FakeProvider().timeout() + + +def fake_provider_auth_error() -> FakeProvider: + """Create a FakeProvider that simulates auth error.""" + return FakeProvider().auth_error() + + +def fake_provider_rate_limit() -> FakeProvider: + """Create a FakeProvider that simulates rate limit.""" + return FakeProvider().rate_limit() + + +def fake_provider_malformed() -> FakeProvider: + """Create a FakeProvider that simulates malformed response.""" + return FakeProvider().malformed() + + +# ============== Mock Model Factory ============== + + +def fake_model( + *, + uuid: str = "test-model-uuid", + name: str = "test-model", + abilities: list[str] = None, + provider: FakeProvider = None, +) -> Mock: + """Create a mock model with a fake provider.""" + model = Mock() + model.model_entity = Mock() + model.model_entity.uuid = uuid + model.model_entity.name = name + model.model_entity.abilities = abilities or ["func_call", "vision"] + model.model_entity.extra_args = {} + + # Attach fake provider + if provider is None: + provider = FakeProvider() + + model.provider = provider + + return model \ No newline at end of file