|
| 1 | +""" |
| 2 | +AGiXT Python SDK Tests |
| 3 | +
|
| 4 | +These tests run against a live AGiXT server. |
| 5 | +Set the following environment variables: |
| 6 | +- AGIXT_URI: AGiXT server URI (default: http://localhost:7437) |
| 7 | +- AGIXT_API_KEY: API key for authentication (default: test-api-key) |
| 8 | +""" |
| 9 | + |
| 10 | +import os |
| 11 | +import uuid |
| 12 | +import pytest |
| 13 | +from agixtsdk import AGiXTSDK |
| 14 | + |
| 15 | + |
| 16 | +# Configuration from environment |
| 17 | +AGIXT_URI = os.getenv("AGIXT_URI", "http://localhost:7437") |
| 18 | +AGIXT_API_KEY = os.getenv("AGIXT_API_KEY", "test-api-key") |
| 19 | + |
| 20 | + |
| 21 | +@pytest.fixture(scope="module") |
| 22 | +def agixt(): |
| 23 | + """Create AGiXT SDK instance.""" |
| 24 | + return AGiXTSDK(base_uri=AGIXT_URI, api_key=AGIXT_API_KEY) |
| 25 | + |
| 26 | + |
| 27 | +@pytest.fixture(scope="module") |
| 28 | +def test_user(agixt): |
| 29 | + """Create a test user and return credentials.""" |
| 30 | + email = f"test_{uuid.uuid4().hex[:8]}@example.com" |
| 31 | + response = agixt.register_user(email=email, first_name="Test", last_name="User") |
| 32 | + return {"email": email, "response": response} |
| 33 | + |
| 34 | + |
| 35 | +@pytest.fixture(scope="module") |
| 36 | +def test_agent(agixt): |
| 37 | + """Create a test agent and clean up after tests.""" |
| 38 | + agent_name = f"TestAgent_{uuid.uuid4().hex[:8]}" |
| 39 | + agent = agixt.add_agent( |
| 40 | + agent_name=agent_name, |
| 41 | + settings={"provider": "default"}, |
| 42 | + ) |
| 43 | + yield {"name": agent_name, "data": agent} |
| 44 | + # Cleanup: delete agent after tests |
| 45 | + try: |
| 46 | + agent_id = agixt.get_agent_id_by_name(agent_name) |
| 47 | + if agent_id: |
| 48 | + agixt.delete_agent(agent_id) |
| 49 | + except Exception: |
| 50 | + pass |
| 51 | + |
| 52 | + |
| 53 | +@pytest.fixture(scope="module") |
| 54 | +def test_conversation(agixt, test_agent): |
| 55 | + """Create a test conversation.""" |
| 56 | + agent_id = agixt.get_agent_id_by_name(test_agent["name"]) |
| 57 | + conv_name = f"TestConversation_{uuid.uuid4().hex[:8]}" |
| 58 | + conv = agixt.new_conversation(agent_name=agent_id, conversation_name=conv_name) |
| 59 | + yield {"name": conv_name, "data": conv} |
| 60 | + # Cleanup |
| 61 | + try: |
| 62 | + conv_id = agixt.get_conversation_id_by_name(conv_name) |
| 63 | + if conv_id: |
| 64 | + agixt.delete_conversation(conv_id) |
| 65 | + except Exception: |
| 66 | + pass |
| 67 | + |
| 68 | + |
| 69 | +class TestConnection: |
| 70 | + """Test basic connectivity to AGiXT server.""" |
| 71 | + |
| 72 | + def test_server_reachable(self, agixt): |
| 73 | + """Test that the AGiXT server is reachable.""" |
| 74 | + # If we can get providers, server is up |
| 75 | + providers = agixt.get_providers() |
| 76 | + assert providers is not None |
| 77 | + |
| 78 | + |
| 79 | +class TestUserManagement: |
| 80 | + """Test user management endpoints.""" |
| 81 | + |
| 82 | + def test_register_user(self, agixt): |
| 83 | + """Test user registration.""" |
| 84 | + email = f"newuser_{uuid.uuid4().hex[:8]}@example.com" |
| 85 | + response = agixt.register_user(email=email, first_name="New", last_name="User") |
| 86 | + assert response is not None |
| 87 | + |
| 88 | + def test_user_exists(self, agixt, test_user): |
| 89 | + """Test checking if user exists.""" |
| 90 | + exists = agixt.user_exists(test_user["email"]) |
| 91 | + assert exists is True |
| 92 | + |
| 93 | + def test_get_user(self, agixt): |
| 94 | + """Test getting current user info.""" |
| 95 | + user = agixt.get_user() |
| 96 | + # Should return user data or None |
| 97 | + assert user is None or isinstance(user, dict) |
| 98 | + |
| 99 | + |
| 100 | +class TestAgentManagement: |
| 101 | + """Test agent management endpoints.""" |
| 102 | + |
| 103 | + def test_get_agents(self, agixt): |
| 104 | + """Test listing agents.""" |
| 105 | + agents = agixt.get_agents() |
| 106 | + assert isinstance(agents, (list, dict)) |
| 107 | + |
| 108 | + def test_add_agent(self, agixt): |
| 109 | + """Test creating an agent.""" |
| 110 | + agent_name = f"NewAgent_{uuid.uuid4().hex[:8]}" |
| 111 | + agent = agixt.add_agent(agent_name=agent_name, settings={"provider": "default"}) |
| 112 | + assert agent is not None |
| 113 | + # Cleanup |
| 114 | + try: |
| 115 | + agent_id = agixt.get_agent_id_by_name(agent_name) |
| 116 | + if agent_id: |
| 117 | + agixt.delete_agent(agent_id) |
| 118 | + except Exception: |
| 119 | + pass |
| 120 | + |
| 121 | + def test_get_agent_id_by_name(self, agixt, test_agent): |
| 122 | + """Test getting agent ID by name.""" |
| 123 | + agent_id = agixt.get_agent_id_by_name(test_agent["name"]) |
| 124 | + assert agent_id is not None |
| 125 | + |
| 126 | + def test_get_agent_config(self, agixt, test_agent): |
| 127 | + """Test getting agent configuration.""" |
| 128 | + agent_id = agixt.get_agent_id_by_name(test_agent["name"]) |
| 129 | + config = agixt.get_agent_config(agent_id) |
| 130 | + assert config is not None |
| 131 | + |
| 132 | + |
| 133 | +class TestConversationManagement: |
| 134 | + """Test conversation management endpoints.""" |
| 135 | + |
| 136 | + def test_get_conversations(self, agixt): |
| 137 | + """Test listing conversations.""" |
| 138 | + conversations = agixt.get_conversations() |
| 139 | + assert isinstance(conversations, (list, dict)) |
| 140 | + |
| 141 | + def test_new_conversation(self, agixt, test_agent): |
| 142 | + """Test creating a conversation.""" |
| 143 | + agent_id = agixt.get_agent_id_by_name(test_agent["name"]) |
| 144 | + conv_name = f"NewConv_{uuid.uuid4().hex[:8]}" |
| 145 | + conv = agixt.new_conversation(agent_name=agent_id, conversation_name=conv_name) |
| 146 | + assert conv is not None |
| 147 | + # Cleanup |
| 148 | + try: |
| 149 | + conv_id = agixt.get_conversation_id_by_name(conv_name) |
| 150 | + if conv_id: |
| 151 | + agixt.delete_conversation(conv_id) |
| 152 | + except Exception: |
| 153 | + pass |
| 154 | + |
| 155 | + def test_get_conversation(self, agixt, test_conversation): |
| 156 | + """Test getting conversation history.""" |
| 157 | + conv_id = agixt.get_conversation_id_by_name(test_conversation["name"]) |
| 158 | + history = agixt.get_conversation(conv_id) |
| 159 | + assert history is not None |
| 160 | + |
| 161 | + |
| 162 | +class TestProviders: |
| 163 | + """Test provider endpoints.""" |
| 164 | + |
| 165 | + def test_get_providers(self, agixt): |
| 166 | + """Test getting all providers.""" |
| 167 | + providers = agixt.get_providers() |
| 168 | + assert providers is not None |
| 169 | + |
| 170 | + def test_get_providers_by_service(self, agixt): |
| 171 | + """Test getting providers by service type.""" |
| 172 | + providers = agixt.get_providers_by_service("llm") |
| 173 | + assert providers is not None |
| 174 | + |
| 175 | + |
| 176 | +class TestChains: |
| 177 | + """Test chain management endpoints.""" |
| 178 | + |
| 179 | + def test_get_chains(self, agixt): |
| 180 | + """Test listing chains.""" |
| 181 | + chains = agixt.get_chains() |
| 182 | + assert isinstance(chains, (list, dict)) |
| 183 | + |
| 184 | + |
| 185 | +class TestPrompts: |
| 186 | + """Test prompt management endpoints.""" |
| 187 | + |
| 188 | + def test_get_prompts(self, agixt): |
| 189 | + """Test listing prompts.""" |
| 190 | + prompts = agixt.get_prompts() |
| 191 | + assert isinstance(prompts, (list, dict)) |
| 192 | + |
| 193 | + def test_get_all_prompts(self, agixt): |
| 194 | + """Test getting all prompts.""" |
| 195 | + prompts = agixt.get_all_prompts() |
| 196 | + assert prompts is not None |
| 197 | + |
| 198 | + |
| 199 | +class TestExtensions: |
| 200 | + """Test extension endpoints.""" |
| 201 | + |
| 202 | + def test_get_extensions(self, agixt): |
| 203 | + """Test getting extensions.""" |
| 204 | + extensions = agixt.get_extensions() |
| 205 | + assert extensions is not None |
| 206 | + |
| 207 | + |
| 208 | +if __name__ == "__main__": |
| 209 | + pytest.main([__file__, "-v"]) |
0 commit comments