apunkt/trustgraph
1
0
Fork 0
mirror of https://github.com/trustgraph-ai/trustgraph.git synced 2026-07-24 20:51:02 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

Fixing tests

Browse source
This commit is contained in:
Cyber MacGeddon 2025-09-03 16:34:28 +01:00
parent 8ee976a69b
commit 6bc95e773f
1 changed files with 21 additions and 39 deletions
Download patch file Download diff file Expand all files Collapse all files

60
tests/integration/test_tool_group_integration.py
View file

@ -14,7 +14,6 @@ from unittest.mock import Mock, AsyncMock, patch
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', 'trustgraph-base')) sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', 'trustgraph-base'))
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', 'trustgraph-flow')) sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', 'trustgraph-flow'))
from trustgraph.schema import AgentRequest, AgentResponse, AgentStep
from trustgraph.agent.tool_filter import filter_tools_by_group_and_state, get_next_state, validate_tool_config from trustgraph.agent.tool_filter import filter_tools_by_group_and_state, get_next_state, validate_tool_config
@ -238,48 +237,31 @@ class TestCompleteWorkflow:
final_state = get_next_state(analysis_tool, 'analysis') final_state = get_next_state(analysis_tool, 'analysis')
assert final_state == 'results' # Transition to results state assert final_state == 'results' # Transition to results state
@pytest.mark.asyncio def test_multi_tenant_scenario(self, sample_tools):
async def test_multi_tenant_scenario(self, agent_processor, sample_tools):
"""Test different users with different permissions.""" """Test different users with different permissions."""
agent_processor.agent.tools = sample_tools # User A: Read-only permissions in undefined state
user_a_tools = filter_tools_by_group_and_state(
from trustgraph.agent.react.types import Final sample_tools,
['read-only'],
# User A: Read-only permissions 'undefined'
user_a_final = Final(final="Read-only operations completed")
agent_processor.agent.react = AsyncMock(return_value=user_a_final)
request_a = AgentRequest(
question="What information is available about X?",
state="undefined",
group=["read-only"],
history=[]
) )
responses_a = [] # Should only have access to read-only tools in undefined state
await agent_processor.agent_request(request_a, assert 'knowledge_query' in user_a_tools # read-only + available in undefined
lambda r: responses_a.append(r), assert 'text_completion' in user_a_tools # read-only + available in all states
lambda x: None, {}) assert 'graph_update' not in user_a_tools # write permissions required
assert 'complex_analysis' not in user_a_tools # advanced permissions required
# User B: Admin permissions # User B: Admin permissions in analysis state
user_b_final = Final(final="Administrative tasks completed") user_b_tools = filter_tools_by_group_and_state(
agent_processor.agent.react = AsyncMock(return_value=user_b_final) sample_tools,
['write', 'admin'],
request_b = AgentRequest( 'analysis'
question="Update the knowledge base",
state="analysis",
group=["write", "admin"],
history=[]
) )
responses_b = [] # Should have access to admin tools available in analysis state
await agent_processor.agent_request(request_b, assert 'graph_update' in user_b_tools # admin + available in analysis
lambda r: responses_b.append(r), assert 'complex_analysis' not in user_b_tools # wrong group (needs advanced/compute)
lambda x: None, {}) assert 'knowledge_query' not in user_b_tools # not available in analysis state
assert 'text_completion' not in user_b_tools # wrong group (no admin)
# Verify both users got appropriate responses
assert len(responses_a) == 1
assert len(responses_b) == 1
assert "Read-only" in responses_a[0].answer
assert "Administrative" in responses_b[0].answer