EQTY LabDocs

Testing

guardian_agent.testing: FakeProxy, a stand-in for the Viper proxy in your tests, and FakeRequest, a model request that got through one.

View as Markdown

Test your agents without a Viper proxy: FakeProxy stands in for one.

A FakeProxy serves the proxy’s local agent API and its model traffic over real HTTP on 127.0.0.1, so your agents, and the OpenAI and Anthropic clients bound to them, run exactly as they do in production. It checks what the proxy checks and refuses with the proxy’s error codes, then answers model requests with a short canned reply instead of calling a model.

Example:

python
from openai import OpenAI

import guardian_agent as ga
from guardian_agent.testing import FakeProxy

with FakeProxy(agents=["researcher"]) as proxy:
    researcher = ga.Guardian(proxy.url).agent("researcher", model=...)
    researcher.bind(OpenAI(api_key="test")).chat.completions.create(...)
    assert proxy.requests[-1].agent == "researcher"

It takes skill and memory files as the proxy does (Guardian.skill(), Guardian.instructions()), with the proxy’s checks, and refuses a definition naming files it doesn’t hold.

What it doesn’t do: sign, compute real content IDs (each definition gets a stable fake-… stand-in, and each skill’s files a stable bagaachrafake… one in a collection CID’s form), store prompts apart, or forward anything anywhere.

Classes:

  • FakeRequest – A model request that got through a FakeProxy.
  • FakeProxy – A stand-in for viper proxy, for tests.

testing.FakeRequest

python
FakeRequest(method: str, path: str, agent: str | None, definition_cid: str | None, subagent: str | None, headers: Mapping[str, str], body: bytes, json: Any)

A model request that got through a FakeProxy.

Attributes:

  • method (str) – The HTTP method, such as "POST".
  • path (str) – The URL path, such as "/v1/chat/completions", without the query.
  • agent (str | None) – The agent it named (X-Viper-Agent), or None.
  • definition_cid (str | None) – The definition it named (X-Viper-Definition).
  • subagent (str | None) – The sub-agent it named (X-Viper-Subagent), as a path such as "critic/fact-checker".
  • headers (Mapping[str, str]) – Its other headers, names lower-cased. The proxy removes the X-Viper-* headers before forwarding, so they aren’t here.
  • body (bytes) – The body as sent.
  • json (Any) – The body parsed as JSON, or None if it isn’t JSON.

testing.FakeProxy

python
FakeProxy(agents: Iterable[str] = (), *, strict: bool = False, features: Iterable[str] | None = None)

A stand-in for viper proxy, for tests.

Use it as a context manager, or call start and close:

python
@pytest.fixture
def proxy() -> Iterator[FakeProxy]:
    with FakeProxy(agents=["researcher"]) as proxy:
        yield proxy

It is safe to use from many threads and async tasks at once.

Parameters:

  • agents (Iterable[str]) – The names of the agents enrolled on it.
  • strict (bool) – Refuse model requests that name no agent, as viper proxy --require-agent-header does.
  • features (Iterable[str] | None) – What it says it does beyond the API version: every feature the Viper release the SDK needs has, unless given. Leave one out to see how your code meets an older proxy.

Raises:

  • TypeError – agents or features is a single string rather than a list.
  • ValueError – A name isn’t a valid agent name.

Functions:

  • start – Start listening on a free port on 127.0.0.1.
  • close – Stop listening, and drop every open connection.
  • restart – Forget every registered definition, as a restarted proxy does.
  • refuse – Refuse the next model requests with code, as the proxy would.
  • reply – Answer every model request from now on with body, as JSON.

Attributes:

  • url (str) – The base URL to give guardian_agent.Guardian.
  • requests (list[FakeRequest]) – Every model request that got through, oldest first.
  • definitions (dict[str, dict[str, Any]]) – Each agent’s latest registered definition, as the proxy holds it.

testing.FakeProxy.url

python
url: str

The base URL to give guardian_agent.Guardian.

Raises:

  • RuntimeError – The proxy isn’t running.

testing.FakeProxy.requests

python
requests: list[FakeRequest]

Every model request that got through, oldest first.

A copy: requests refused (by the proxy’s checks, or by refuse) never reach the model, and aren’t here.

testing.FakeProxy.definitions

python
definitions: dict[str, dict[str, Any]]

Each agent’s latest registered definition, as the proxy holds it.

A copy, keyed by agent name: JSON data with the schema’s field names, defaults filled and empty fields dropped, as the proxy stores it. Unlike the proxy, long prompts keep their text.

testing.FakeProxy.start

python
start() -> Self

Start listening on a free port on 127.0.0.1.

Started again after close, it listens on a new port and keeps what it recorded; restart is what forgets registrations.

Returns:

  • Self – The proxy itself.

Raises:

  • RuntimeError – It is already running.

testing.FakeProxy.close

python
close() -> None

Stop listening, and drop every open connection.

Clients that kept a connection open then find nothing there, as they would with a stopped proxy. Closing a proxy that isn’t running does nothing.

testing.FakeProxy.restart

python
restart() -> None

Forget every registered definition, as a restarted proxy does.

An agent’s next model request is refused with definition_unknown, and the SDK registers the definition again and resends the request, as it does with the real proxy.

testing.FakeProxy.refuse

python
refuse(code: str, *, times: int = 1, message: str | None = None, hint: str | None = None) -> None

Refuse the next model requests with code, as the proxy would.

Each refusal has the code’s status and the proxy’s error shape and mark, so your code meets it as it would meet the real thing. Model clients retry some refusals (signing_failed, internal_error, upstream_unreachable) on their own; refuse more times than they retry to see the error.

Example:

python
proxy.refuse("upstream_response_lost")
with pytest.raises(openai.APIStatusError) as caught:
    client.chat.completions.create(...)
assert isinstance(ga.proxy_error(caught.value), ga.UpstreamResponseLost)

Parameters:

  • code (str) – A proxy error code, such as "signing_failed".
  • times (int) – How many requests to refuse, one after another.
  • message (str | None) – The error’s message; a default says it was injected.
  • hint (str | None) – The error’s hint.

Raises:

  • ValueError – code isn’t one the proxy sends, or times is less than 1.

testing.FakeProxy.reply

python
reply(body: object | _Reply, *, status: int = 200) -> None

Answer every model request from now on with body, as JSON.

Replaces the canned replies, which are made for OpenAI’s Chat Completions and Responses APIs and Anthropic’s Messages API, streamed when the request asks for a stream. A reply set here is never streamed.

Example:

python
proxy.reply(
    lambda request: {...}  # a tool call first, then an answer
    if len(proxy.requests) == 1
    else {...}
)

Parameters:

  • body (object | _Reply) – JSON data, or a function that takes the FakeRequest and returns JSON data.
  • status (int) – The HTTP status to answer with.

Raises:

  • TypeError – body isn’t JSON data or a function.
  • ValueError – status isn’t an HTTP status from 200 to 599.

A function that raises, or returns something JSON can’t carry, is answered with a 500 that names the error.