110 lines
3.8 KiB
Python
110 lines
3.8 KiB
Python
"""Stationary nominal completion clients for isolated tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
from band_aider.ai_client.completion_client import CompletionClient, OpenAIConfig
|
|
from band_aider.ai_client.openai_compatible_client import OpenAICompatibleClient
|
|
|
|
|
|
class StaticOpenAICompatibleClient(OpenAICompatibleClient):
|
|
"""Return one fixed completion without opening an external connection."""
|
|
|
|
def __init__(self, response: str) -> None:
|
|
super().__init__(model="test-model", api_base="http://model.invalid")
|
|
self._response = response
|
|
|
|
def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
|
|
"""Return the configured response for the supplied prompt."""
|
|
|
|
del prompt, logger
|
|
return self._response
|
|
|
|
|
|
class ScriptedOpenAICompatibleClient(OpenAICompatibleClient):
|
|
"""Return stationary OpenAI-compatible responses while preserving prompt evidence."""
|
|
|
|
def __init__(self, responses: tuple[str, ...]) -> None:
|
|
"""Bind a finite explicit response script without creating a transport client."""
|
|
|
|
super().__init__(model="test-model", api_base="http://model.invalid")
|
|
self._responses = list(responses)
|
|
self.prompts: list[str] = []
|
|
|
|
def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
|
|
"""Record one prompt and return its next stationary scripted response."""
|
|
|
|
del logger
|
|
self.prompts.append(prompt)
|
|
if not self._responses:
|
|
raise AssertionError("The stationary OpenAI-compatible response script was exhausted.")
|
|
return self._responses.pop(0)
|
|
|
|
|
|
class StaticCompletionClient(CompletionClient):
|
|
"""Return one fixed completion through the direct-edit client contract."""
|
|
|
|
def __init__(self, response: str) -> None:
|
|
"""Store one stationary response and non-routable client configuration."""
|
|
|
|
self._response = response
|
|
self._configs = (
|
|
OpenAIConfig(model="test-model", api_base="http://model.invalid"),
|
|
)
|
|
|
|
@property
|
|
def inner(self) -> CompletionClient | None:
|
|
"""Return no wrapped client for this terminal fake response source."""
|
|
|
|
return None
|
|
|
|
def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
|
|
"""Return the configured response without opening an external connection."""
|
|
|
|
del prompt, logger
|
|
return self._response
|
|
|
|
@property
|
|
def openai_configs(self) -> tuple[OpenAIConfig, ...]:
|
|
"""Expose one stationary configuration for escalation metadata."""
|
|
|
|
return self._configs
|
|
|
|
|
|
class ScriptedCompletionClient(CompletionClient):
|
|
"""Record prompts and return a fixed finite sequence without network transport."""
|
|
|
|
def __init__(
|
|
self,
|
|
responses: tuple[str, ...],
|
|
configs: tuple[OpenAIConfig, ...] | None = None,
|
|
) -> None:
|
|
"""Store fixed responses and an explicit configuration chain."""
|
|
|
|
self._responses = list(responses)
|
|
self._configs = configs or (
|
|
OpenAIConfig(model="test-model", api_base="http://model.invalid"),
|
|
)
|
|
self.prompts: list[str] = []
|
|
|
|
@property
|
|
def inner(self) -> CompletionClient | None:
|
|
"""Return no inner client because this is a terminal stationary source."""
|
|
|
|
return None
|
|
|
|
def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
|
|
"""Record one prompt and return its next configured stationary response."""
|
|
|
|
del logger
|
|
self.prompts.append(prompt)
|
|
if not self._responses:
|
|
raise AssertionError("The stationary completion response sequence was exhausted.")
|
|
return self._responses.pop(0)
|
|
|
|
@property
|
|
def openai_configs(self) -> tuple[OpenAIConfig, ...]:
|
|
"""Expose the explicitly configured model chain."""
|
|
|
|
return self._configs
|