continuum-testing

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 continuum.agent import AgentRunner, BaseAgent
from continuum.core.container import Container, ContainerConfig, reset_container
from continuum.llm.types import LLMResponse, StreamChunk, Usage
from continuum.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!"