continuum-testing

name	continuum-testing
description	Write tests for Continuum agents — mock LLM and memory clients via the DI Container, use fakeredis for sessions, snapshot agent responses, and run pytest-asyncio. Invoke when the user asks "test my agent", "mock the LLM", "fakeredis", "container injection", "pytest", or wants their CI to validate agent behavior without burning real API tokens.

Continuum Testing Skill

Continuum's test-friendly seams:

1. Container(ContainerConfig(auto_initialize=False)) — opt out of auto-init and inject mocks.
2. Container.set_*_client(...) accepts the real client, a custom subclass, or any object satisfying the ILLMClient / IMemoryClient / ISessionClient protocols.
3. fakeredis (a dev dependency) drops in for the Redis session provider.
4. reset_container() and reset_global_memory() / reset_global_session() cleanly nuke globals between tests.

---

Imports

import pytest
from orchestrator.agent import AgentRunner, BaseAgent
from orchestrator.core.container import Container, ContainerConfig, reset_container
from orchestrator.llm.types import LLMResponse, StreamChunk, Usage
from orchestrator.protocols import ILLMClient

---

1. Mock LLM that satisfies ILLMClient

class MockLLM:
    is_enabled = True

    def __init__(self, replies: list[str]):
        self._replies = list(replies)

    async def chat(self, messages, **kw):
        text = self._replies.pop(0) if self._replies else "mock reply"
        return LLMResponse(model="mock", content=text, role="assistant",
                           usage=Usage(prompt_tokens=10, completion_tokens=5,
                                       total_tokens=15))

    async def chat_stream(self, messages, **kw):
        text = self._replies.pop(0) if self._replies else "mock reply"
        for ch in text:
            yield StreamChunk(model="mock", content=ch)
        yield StreamChunk(model="mock", is_finished=True)

    def count_tokens(self, messages, model=None) -> int:
        return sum(len(m.get("content", "")) // 4 for m in messages)

Wire it into the container:

container = Container(ContainerConfig(auto_initialize=False))
container.set_llm_client(MockLLM(["Hello, world!"]))
container.set_memory_client(None)
container.set_session_client(None)

runner = AgentRunner(container=container)
agent = BaseAgent(name="t", instructions="...")
resp = await runner.run(agent, "anything")
assert resp.content == "Hello, world!"

---

2. pytest fixtures

import pytest
from orchestrator.core.container import reset_container

@pytest.fixture(autouse=True)
def clean_globals():
    reset_container()
    try:
        from orchestrator.memory.client import reset_global_memory
        reset_global_memory()
    except ImportError:
        pass
    try:
        from orchestrator.session.client import reset_global_session
        reset_global_session()
    except ImportError:
        pass
    yield

@pytest.fixture
def mock_runner():
    container = Container(ContainerConfig(auto_initialize=False))
    container.set_llm_client(MockLLM(["mocked"]))
    container.set_memory_client(None)
    container.set_session_client(None)
    return AgentRunner(container=container)

@pytest.mark.asyncio
async def test_agent_replies(mock_runner):
    agent = BaseAgent(name="echo", instructions="...")
    resp = await mock_runner.run(agent, "ping")
    assert resp.content == "mocked"

pytest-asyncio is in the [dev] extra. With asyncio_mode = "auto" in pyproject.toml, you don't need the @pytest.mark.asyncio decorator.

---

3. fakeredis for session tests

import pytest, fakeredis.aioredis
from orchestrator.session import SessionClient, SessionConfig
from orchestrator.session.providers.redis import RedisSessionProvider

@pytest.fixture
async def session_client(monkeypatch):
    fake = fakeredis.aioredis.FakeRedis()
    cfg = SessionConfig(provider="redis", enabled=True,
                        redis_host="ignored", redis_port=0)
    provider = RedisSessionProvider(cfg, auto_initialize=False)
    provider._redis = fake                       # bypass real Redis connect
    provider._initialized = True
    return SessionClient(session_config=cfg, provider=provider, auto_initialize=False)

Now await session_client.add_message(sid, ChatMessage(...)) round-trips through fakeredis without a Docker container.

---

4. Snapshot-style tests against structured output

from pydantic import BaseModel

class Result(BaseModel):
    sentiment: str
    confidence: float

async def test_sentiment_schema(mock_runner):
    mock_runner._container.set_llm_client(
        MockLLM(['{"sentiment":"positive","confidence":0.95}'])
    )
    agent = BaseAgent(name="s", instructions="...", output_schema=Result)
    resp = await mock_runner.run(agent, "I love it!")
    assert isinstance(resp.structured_output, Result)
    assert resp.structured_output.sentiment == "positive"

---

5. Testing tool calls

If your MockLLM needs to emit tool calls, populate tool_calls on the LLMResponse:

from orchestrator.llm.types import LLMResponse, ToolCall, FunctionCall

class ToolCallingMock:
    is_enabled = True
    def __init__(self):
        self._step = 0
    async def chat(self, messages, **kw):
        self._step += 1
        if self._step == 1:
            return LLMResponse(
                model="mock", content=None, role="assistant",
                tool_calls=[ToolCall(id="c1", function=FunctionCall(
                    name="get_weather", arguments='{"city": "Paris"}'))])
        return LLMResponse(model="mock", content="It's sunny in Paris.",
                           role="assistant")
    async def chat_stream(self, *a, **kw):
        yield StreamChunk(model="mock", is_finished=True)
    def count_tokens(self, *a, **kw): return 0

Couple this with a fake ToolExecutor to test full tool-loop behaviour without real MCP servers.

---

6. Asserting on streaming events

from orchestrator.agent.types import EventType

async def test_streaming_emits_deltas(mock_runner):
    agent = BaseAgent(name="s", instructions="...")
    types = [ev.type async for ev in mock_runner.run_stream(agent, "hi")]
    assert EventType.RUN_START in types
    assert EventType.CONTENT_DELTA in types
    assert EventType.RUN_END in types

---

7. End-to-end with a live LLM but no infra

For "happy path" tests against a real LLM but without Redis/Qdrant:

container = Container(ContainerConfig(
    enable_memory=False, enable_session=False, enable_langfuse=False,
))
agent = BaseAgent(
    name="e2e",
    instructions="...",
    memory_config=AgentMemoryConfig(search_memories=False, store_memories=False),
)
runner = AgentRunner(container=container)

Combine with pytest -m integration markers to gate them behind a key.

---

Don't

• Don't await runner.run_stream(...) — it's an async generator; use async for.
• Don't mock BaseAgent itself — it's a dataclass; mock the LLM client underneath instead.
• Don't reuse Container across tests without reset_container() — the singleton survives between tests and leaks state.
• Don't try to test prompt-injection scanning by passing raw HTML — the scanners run on user input, not arbitrary strings; pass via runner.run(agent, payload).
• Don't real-Redis your tests in CI without isolation — use fakeredis or a per-test Redis db (SESSION_REDIS_DB).