Function test_control_knowledge

tests/server/test_agent_server_e2e.py:626–659

knowledge control returns stats and toggles enable/disable.

(tmp_path)

@pytest.mark.asyncio
async def test_control_knowledge(tmp_path):
    """knowledge control returns stats and toggles enable/disable."""
    from src.tool_system.registry import ToolRegistry

    with contextlib.ExitStack() as stack:
        for p in _patches(_TextProvider, ToolRegistry([])):
            stack.enter_context(p)
        spawn = make_spawn_agent(AgentServerConfig(permission_mode="default"))
        handle = await spawn("ds_test", str(tmp_path), None)
        gen = handle.messages_from_agent()
        try:
            init = await asyncio.wait_for(gen.__anext__(), timeout=5)
            assert init["subtype"] == "init"

            async def _reply_for(rid, req):
                await handle.send_to_agent({"type": "control_request", "request_id": rid, "request": req})
                for _ in range(12):
                    msg = await asyncio.wait_for(gen.__anext__(), timeout=5)
                    if msg.get("type") == "control_response" and msg["response"].get("request_id") == rid:
                        return msg["response"]["response"]
                raise AssertionError(f"no reply for {rid}")

            st = await _reply_for("k1", {"subtype": "knowledge", "action": "status"})
            assert st["ok"] is True and "total" in st["stats"] and st["enabled"] is True

            off = await _reply_for("k2", {"subtype": "knowledge", "action": "disable"})
            assert off["enabled"] is False

            lst = await _reply_for("k3", {"subtype": "knowledge", "action": "list"})
            assert lst["ok"] is True and isinstance(lst["entities"], list)
        finally:
            await handle.shutdown()
            with contextlib.suppress(Exception):
                await gen.aclose()


@pytest.mark.asyncio