Claude Slack

""" Comprehensive integration tests for agent management. Tests agent lifecycle, discovery, permissions, and DM policies. """ import pytest import pytest_asyncio from typing import Dict, List, Optional class TestAgentLifecycle: """Test agent registration, updates, and deletion.""" @pytest.mark.asyncio async def test_agent_registration_all_scopes(self, test_db): """Test registering agents in different scopes.""" # Global agent await test_db.register_agent( name="global_agent", project_id=None, description="Global agent", dm_policy="open", discoverable="public" ) # Project agent await test_db.register_project("test_proj", "/path", "Test Project") await test_db.register_agent( name="project_agent", project_id="test_proj", description="Project agent", dm_policy="restricted", discoverable="project" ) # Verify agents exist global_agent = await test_db.get_agent("global_agent", None) assert global_agent is not None assert global_agent['project_id'] is None assert global_agent['dm_policy'] == 'open' project_agent = await test_db.get_agent("project_agent", "test_proj") assert project_agent is not None assert project_agent['project_id'] == "test_proj" assert project_agent['discoverable'] == 'project' @pytest.mark.asyncio async def test_agent_unique_per_project(self, test_db): """Test that same agent name can exist in different projects.""" # Create two projects await test_db.register_project("proj_a", "/a", "Project A") await test_db.register_project("proj_b", "/b", "Project B") # Register same name in different projects await test_db.register_agent("shared_name", "proj_a", "Agent in A") await test_db.register_agent("shared_name", "proj_b", "Agent in B") await test_db.register_agent("shared_name", None, "Global agent") # All three should exist agent_a = await test_db.get_agent("shared_name", "proj_a") agent_b = await test_db.get_agent("shared_name", "proj_b") agent_global = await test_db.get_agent("shared_name", None) assert agent_a['description'] == "Agent in A" assert agent_b['description'] == "Agent in B" assert agent_global['description'] == "Global agent" @pytest.mark.asyncio async def test_agent_update(self, populated_db): """Test updating agent properties.""" # Update alice's properties await populated_db.update_agent( agent_name="alice", agent_project_id="proj_test1", description="Updated Alice", dm_policy="restricted", discoverable="private" ) # Verify updates agent = await populated_db.get_agent("alice", "proj_test1") assert agent['description'] == "Updated Alice" assert agent['dm_policy'] == "restricted" assert agent['discoverable'] == "private" class TestAgentDiscovery: """Test agent discovery and visibility rules.""" @pytest.mark.asyncio async def test_discoverability_matrix(self, test_db): """Test all discoverability settings.""" await test_db.register_project("disc_proj", "/disc", "Discovery Test") # Create agents with different discoverability agents = [ ("public_agent", None, "public"), ("project_agent", "disc_proj", "project"), ("private_agent", "disc_proj", "private"), ] for name, proj_id, discoverable in agents: await test_db.register_agent( name=name, project_id=proj_id, description=f"{discoverable} agent", discoverable=discoverable ) # Create a viewer agent to test discovery await test_db.register_agent("test_viewer", None, "Test Viewer") # Get discoverable agents from global context global_discoverable = await test_db.get_discoverable_agents( agent_name="test_viewer", agent_project_id=None ) names = [a['name'] for a in global_discoverable] assert "public_agent" in names # Public is always visible assert "project_agent" in names # Global agents can see project agents assert "private_agent" not in names # Private never visible # Get discoverable from same project # First create a viewer agent in the project await test_db.register_agent("disc_viewer", "disc_proj", "Viewer") project_discoverable = await test_db.get_discoverable_agents( agent_name="disc_viewer", agent_project_id="disc_proj" ) proj_names = [a['name'] for a in project_discoverable] assert "public_agent" in proj_names # Public visible assert "project_agent" in proj_names # Same project visible assert "private_agent" not in proj_names # Private not discoverable @pytest.mark.asyncio async def test_linked_project_discovery(self, test_db): """Test discovery across linked projects.""" # Create projects await test_db.register_project("link_a", "/a", "Link A") await test_db.register_project("link_b", "/b", "Link B") await test_db.register_project("link_c", "/c", "Link C") # Link A and B await test_db.add_project_link("link_a", "link_b", "bidirectional") # Create agents await test_db.register_agent( "agent_a", "link_a", "Agent A", discoverable="project" ) await test_db.register_agent( "agent_b", "link_b", "Agent B", discoverable="project" ) await test_db.register_agent( "agent_c", "link_c", "Agent C", discoverable="project" ) # Create viewer agents in each project await test_db.register_agent("viewer_a", "link_a", "Viewer A") await test_db.register_agent("viewer_c", "link_c", "Viewer C") # From A's perspective (linked to B) from_a = await test_db.get_discoverable_agents( agent_name="viewer_a", agent_project_id="link_a" ) from_a_names = [a['name'] for a in from_a] assert "agent_a" in from_a_names # Own agent assert "agent_b" in from_a_names # Linked project agent assert "agent_c" not in from_a_names # Not linked # From C's perspective (not linked) from_c = await test_db.get_discoverable_agents( agent_name="viewer_c", agent_project_id="link_c" ) from_c_names = [a['name'] for a in from_c] assert "agent_a" not in from_c_names # Not linked assert "agent_b" not in from_c_names # Not linked assert "agent_c" in from_c_names # Own agent @pytest.mark.asyncio async def test_discovery_with_filters(self, populated_db): """Test agent discovery with various filters.""" # Get agents by scope all_agents = await populated_db.get_agents_by_scope(scope='all') assert len(all_agents) >= 3 # alice, bob, charlie # Get agents from specific project proj_agents = await populated_db.get_agents_by_scope( scope='project', project_id="proj_test1" ) proj_names = [a['name'] for a in proj_agents] assert "alice" in proj_names assert "bob" not in proj_names # Different project # Get global agents only global_agents = await populated_db.get_agents_by_scope(scope='global') global_names = [a['name'] for a in global_agents] assert "charlie" in global_names # Global agent assert "alice" not in global_names # Project agent class TestDMPolicies: """Test direct message policies and permissions.""" @pytest.mark.asyncio async def test_dm_policy_enforcement(self, test_db): """Test DM policy enforcement.""" # Create agents with different policies await test_db.register_agent( "open_dm", None, "Open DM", dm_policy="open" ) await test_db.register_agent( "restricted_dm", None, "Restricted DM", dm_policy="restricted" ) await test_db.register_agent( "closed_dm", None, "Closed DM", dm_policy="closed" ) await test_db.register_agent( "sender", None, "Sender", dm_policy="open" ) # Test open -> anyone can DM can_dm_open = await test_db.check_dm_permission( "sender", None, "open_dm", None ) assert can_dm_open is True # Test closed -> no one can DM (without explicit allow) can_dm_closed = await test_db.check_dm_permission( "sender", None, "closed_dm", None ) assert can_dm_closed is False # Test restricted -> needs explicit permission can_dm_restricted = await test_db.check_dm_permission( "sender", None, "restricted_dm", None ) assert can_dm_restricted is False @pytest.mark.asyncio async def test_dm_explicit_permissions(self, test_db): """Test explicit DM allow/block lists.""" await test_db.register_agent( "controller", None, "Controller", dm_policy="restricted" ) await test_db.register_agent("friend", None, "Friend") await test_db.register_agent("enemy", None, "Enemy") # Add explicit allow await test_db.set_dm_permission( "controller", None, "friend", None, permission="allow", reason="Friend" ) # Add explicit block await test_db.set_dm_permission( "controller", None, "enemy", None, permission="block", reason="Enemy" ) # Friend can DM (explicit allow) can_friend = await test_db.check_dm_permission( "friend", None, "controller", None ) assert can_friend is True # Enemy cannot DM (explicit block) can_enemy = await test_db.check_dm_permission( "enemy", None, "controller", None ) assert can_enemy is False @pytest.mark.asyncio async def test_dm_block_overrides_policy(self, test_db): """Test that explicit blocks override open policies.""" await test_db.register_agent( "open_agent", None, "Open", dm_policy="open" ) await test_db.register_agent("blocked_sender", None, "Blocked") # Initially can DM (open policy) can_before = await test_db.check_dm_permission( "blocked_sender", None, "open_agent", None ) assert can_before is True # Add block await test_db.set_dm_permission( "open_agent", None, "blocked_sender", None, permission="block", reason="Blocked" ) # Now cannot DM (block overrides open) can_after = await test_db.check_dm_permission( "blocked_sender", None, "open_agent", None ) assert can_after is False @pytest.mark.asyncio async def test_dm_permissions_bidirectional(self, test_db): """Test that DM permissions work both ways.""" await test_db.register_agent("agent_x", None, "X") await test_db.register_agent("agent_y", None, "Y") # X blocks Y await test_db.set_dm_permission( "agent_x", None, "agent_y", None, permission="block", reason="Test" ) # Y cannot DM X can_y_to_x = await test_db.check_dm_permission( "agent_y", None, "agent_x", None ) assert can_y_to_x is False # X also cannot DM Y (blocks are bidirectional) can_x_to_y = await test_db.check_dm_permission( "agent_x", None, "agent_y", None ) assert can_x_to_y is False class TestAgentChannelMembership: """Test agent channel membership and permissions.""" @pytest.mark.asyncio async def test_agent_default_channels(self, test_db): """Test agents are added to default channels on registration.""" # Create default channels await test_db.create_channel( name="welcome", access_type="open", scope="global", is_default=True ) await test_db.register_project("def_proj", "/def", "Default Test") await test_db.create_channel( name="team", access_type="open", scope="project", project_id="def_proj", is_default=True ) # Register new agent in project await test_db.register_agent( "new_agent", "def_proj", "New Agent" ) # Should be in both default channels global_member = await test_db.is_channel_member( "global:welcome", "new_agent", "def_proj" ) project_member = await test_db.is_channel_member( "def_proj:team", "new_agent", "def_proj" ) # Note: Default channel provisioning happens in ChannelManager.apply_default_channels # which needs to be called explicitly after agent registration # This test verifies the infrastructure is in place @pytest.mark.asyncio async def test_agent_channel_permissions(self, populated_db): """Test agent permissions within channels.""" # Add alice to channel with specific permissions await populated_db.add_channel_member( channel_id="global:general", agent_name="alice", agent_project_id="proj_test1", can_send=True, can_invite=True, can_leave=True ) # Add bob with restricted permissions await populated_db.add_channel_member( channel_id="global:general", agent_name="bob", agent_project_id="proj_test2", can_send=False, can_invite=False, can_leave=True ) # Get members and check permissions members = await populated_db.get_channel_members("global:general") alice_member = next((m for m in members if m['agent_name'] == 'alice'), None) bob_member = next((m for m in members if m['agent_name'] == 'bob'), None) assert alice_member['can_send'] is True assert alice_member['can_invite'] is True assert bob_member['can_send'] is False assert bob_member['can_invite'] is False @pytest.mark.asyncio async def test_agent_channels_list(self, populated_db): """Test listing channels for an agent.""" # Add alice to multiple channels await populated_db.add_channel_member( "global:general", "alice", "proj_test1" ) await populated_db.add_channel_member( "proj_test1:dev", "alice", "proj_test1" ) await populated_db.add_channel_member( "proj_test1:private", "alice", "proj_test1" ) # Get alice's channels channels = await populated_db.get_agent_channels( agent_name="alice", agent_project_id="proj_test1" ) channel_ids = [c['id'] for c in channels] assert "global:general" in channel_ids assert "proj_test1:dev" in channel_ids assert "proj_test1:private" in channel_ids # Bob should have different channels bob_channels = await populated_db.get_agent_channels( agent_name="bob", agent_project_id="proj_test2" ) bob_ids = [c['id'] for c in bob_channels] # Bob shouldn't be in alice's project channels assert "proj_test1:dev" not in bob_ids assert "proj_test1:private" not in bob_ids class TestAgentProjectScope: """Test agent behavior across project boundaries.""" @pytest.mark.asyncio async def test_global_agent_access(self, populated_db): """Test that global agents have broader access.""" # Charlie is a global agent charlie = await populated_db.get_agent("charlie", None) assert charlie['project_id'] is None # Global agents can join project channels await populated_db.add_channel_member( "proj_test1:dev", "charlie", None ) is_member = await populated_db.is_channel_member( "proj_test1:dev", "charlie", None ) assert is_member is True @pytest.mark.asyncio async def test_agent_project_isolation(self, populated_db): """Test that project agents are isolated by default.""" # Alice (proj_test1) and Bob (proj_test2) are isolated alice = await populated_db.get_agent("alice", "proj_test1") bob = await populated_db.get_agent("bob", "proj_test2") assert alice['project_id'] == "proj_test1" assert bob['project_id'] == "proj_test2" # They can still interact in global channels await populated_db.add_channel_member( "global:general", "alice", "proj_test1" ) await populated_db.add_channel_member( "global:general", "bob", "proj_test2" ) members = await populated_db.get_channel_members("global:general") member_names = [(m['agent_name'], m['agent_project_id']) for m in members] assert ("alice", "proj_test1") in member_names assert ("bob", "proj_test2") in member_names @pytest.mark.asyncio async def test_agent_cross_project_with_linking(self, test_db): """Test agent interaction across linked projects.""" # Create and link projects await test_db.register_project("proj_x", "/x", "Project X") await test_db.register_project("proj_y", "/y", "Project Y") await test_db.add_project_link("proj_x", "proj_y", "bidirectional") # Create agents await test_db.register_agent("agent_x", "proj_x", "Agent X") await test_db.register_agent("agent_y", "proj_y", "Agent Y") # Create open channel in proj_x channel_id = await test_db.create_channel( name="shared", access_type="open", scope="project", project_id="proj_x" ) # Agent from linked project can join await test_db.add_channel_member( channel_id, "agent_y", "proj_y" ) is_member = await test_db.is_channel_member( channel_id, "agent_y", "proj_y" ) assert is_member is True