Use when writing new agent tools, memory operations, or server features. Enforces Python TDD with pytest — write tests first, then implementation. 80%+ coverage required.
Write tests before code. Always. No exceptions.
VikarmaToolGatewayNexusBridgeKANMemoryserver/main.pyAs Tvaṣṭā, I want to [action],
so that [benefit for the user / system].
Example:
As Tvaṣṭā, I want to call the `translator` temple with a phrase and target language,
so that I can respond to users in their native language.
# server/tests/test_tool_gateway.py
@pytest.mark.asyncio
async def test_translator_temple_returns_translation(gw):
result = await gw.execute("temple", {
"temple": "translator",
"action": "translate",
"params": {"text": "hello", "target_lang": "es"}
})
assert "error" not in result
assert result["temple"] == "translator"
@pytest.mark.asyncio
async def test_translator_temple_handles_missing_text(gw):
result = await gw.execute("temple", {
"temple": "translator",
"action": "translate",
"params": {}
})
# Should return something, not crash
assert isinstance(result, dict)
python -m pytest server/tests/ -v -k "translator"
# FAILED — expected, we haven't implemented yet
Write the smallest change that makes tests pass. No extras.
python -m pytest server/tests/ -v -k "translator"
# PASSED
Improve without breaking tests:
python -m pytest server/tests/ --cov=server --cov-report=term-missing
# Target: 80%+ coverage
@pytest.fixture
def gw(tmp_path):
return VikarmaToolGateway(workspace=str(tmp_path))
@pytest.mark.asyncio
async def test_my_tool(gw):
result = await gw.execute("my_tool", {"param": "value"})
assert result["success"] is True
@pytest.mark.asyncio
async def test_agent_uses_temple(mem):
temple_call = '<tool>temple</tool><params>{"temple": "calculator", "action": "2+2", "params": {"expression": "2+2"}}</params>'
final = "The answer is 4."
agent = VikarmaAgent(AsyncMock(), VikarmaToolGateway(), mem)
agent._call_ai = AsyncMock(side_effect=[temple_call, final])
result = await agent.run("What is 2+2?")
assert result == final
def test_fact_survives_reload(tmp_path):
mem = KANMemory(storage_dir=str(tmp_path))
mem.remember_fact("language", "Python")
mem2 = KANMemory(storage_dir=str(tmp_path))
results = mem2.recall_fact("language")
assert any(f["value"] == "Python" for f in results)
@pytest.mark.asyncio
async def test_temple_fallback_no_crash():
bridge = NexusBridge()
result = await bridge.call_temple("calculator", "1+1", {"expression": "1+1"})
assert "error" not in result
assert result["result"] == 2
# Run with coverage
python -m pytest server/tests/ \
--cov=server \
--cov-report=term-missing \
--cov-fail-under=80
Minimum 80% coverage on:
server/agents/autonomous_agent.pyserver/agents/kan_memory.pyserver/tools/gateway.pyserver/nexus_bridge.pyfrom unittest.mock import patch, AsyncMock
@pytest.mark.asyncio
async def test_coingecko_temple(gw):
mock_response = {"bitcoin": {"usd": 50000, "usd_24h_change": 2.5}}
with patch("httpx.AsyncClient.get") as mock_get:
mock_get.return_value.json = AsyncMock(return_value=mock_response)
result = await gw.execute("temple", {
"temple": "coingecko",
"action": "price",
"params": {"coin": "bitcoin"}
})
assert result["temple"] == "coingecko"
agent._call_ai = AsyncMock(return_value="Final answer without tool tags")
result = await agent.run("task")
assert result == "Final answer without tool tags"
pytest server/tests/ -v)--cov-fail-under=80)