Claude Slack

"""
Comprehensive tests for the v3 channel permissions model.
Tests the unified membership model, scope restrictions, and cross-project access.
"""

import pytest
from typing import Dict, List, Optional


class TestChannelPermissions:
    """Test suite for channel access and permissions."""
    
    # =========================================================================
    # Self-Join Tests (join_channel)
    # =========================================================================
    
    @pytest.mark.asyncio
    async def test_join_open_global_channel(self, api, populated_db):
        """Any agent can join an open global channel."""
        # Alice from proj_test1 joins global channel
        success = await api.join_channel(
            agent_name="alice",
            agent_project_id="proj_test1",
            channel_id="global:general"
        )
        assert success is True
        
        # Verify membership
        is_member = await api.db.sqlite.is_channel_member(
            "global:general", "alice", "proj_test1"
        )
        assert is_member is True
    
    @pytest.mark.asyncio
    async def test_join_open_same_project_channel(self, api, populated_db):
        """Agent can join open channel in same project."""
        # Add alice to her project first
        await api.db.sqlite.register_agent("alice2", "proj_test1", "Another agent")
        
        # Alice2 joins project channel
        success = await api.join_channel(
            agent_name="alice2",
            agent_project_id="proj_test1",
            channel_id="proj_test1:dev"
        )
        assert success is True
    
    @pytest.mark.asyncio
    async def test_cannot_join_different_project_channel(self, api, populated_db):
        """Agent cannot self-join channel from different unlinked project."""
        # Bob (proj_test2) tries to join proj_test1 channel
        success = await api.join_channel(
            agent_name="bob",
            agent_project_id="proj_test2",
            channel_id="proj_test1:dev"
        )
        assert success is False
    
    @pytest.mark.asyncio
    async def test_join_linked_project_channel(self, api, populated_db):
        """Agent can join open channel from linked project."""
        # Link the projects
        await api.db.sqlite.add_project_link("proj_test1", "proj_test2")
        
        # Now Bob can join proj_test1 channel
        success = await api.join_channel(
            agent_name="bob",
            agent_project_id="proj_test2",
            channel_id="proj_test1:dev"
        )
        assert success is True
    
    @pytest.mark.asyncio
    async def test_global_agent_join_project_channel(self, api, populated_db):
        """Global agent can join any project's open channel."""
        # Charlie (global) joins project channel
        success = await api.join_channel(
            agent_name="charlie",
            agent_project_id=None,
            channel_id="proj_test1:dev"
        )
        assert success is True
    
    @pytest.mark.asyncio
    async def test_cannot_join_members_channel(self, api, populated_db):
        """No agent can self-join a members-only channel."""
        # Alice tries to join members channel in her own project
        success = await api.join_channel(
            agent_name="alice",
            agent_project_id="proj_test1",
            channel_id="proj_test1:private"
        )
        assert success is False
    
    @pytest.mark.asyncio
    async def test_cannot_join_private_channel(self, api, populated_db):
        """No agent can join a private channel (DMs)."""
        # Create a DM channel
        dm_id = await api.db.create_or_get_dm_channel(
            "alice", "proj_test1", "bob", "proj_test2"
        )
        
        # Charlie tries to join the DM
        success = await api.join_channel(
            agent_name="charlie",
            agent_project_id=None,
            channel_id=dm_id
        )
        assert success is False
    
    # =========================================================================
    # Invitation Tests (invite_to_channel)
    # =========================================================================
    
    @pytest.mark.asyncio
    async def test_cannot_invite_to_open_channel(self, api, populated_db):
        """Cannot invite to open channels - they allow self-service joining."""
        # Alice joins first
        await api.join_channel(
            "alice", "proj_test1", "global:general"
        )
        
        # Check that the channel is open (can't invite to open channels)
        channel = await api.db.sqlite.get_channel("global:general")
        assert channel['access_type'] == 'open'
        
        # In the new model, we don't have a direct invite method for open channels
        # Open channels allow self-join, so Bob should just join directly
        # This test validates the business rule that open channels don't need invites
        
        # Bob should just join directly
        success = await api.join_channel(
            "bob", "proj_test2", "global:general"
        )
        assert success is True
    
    @pytest.mark.asyncio
    async def test_invite_cross_project_to_members_channel(self, api, populated_db):
        """Can invite agents from different projects to members-only channels."""
        # Create a members-only channel using the API
        await api.create_channel(
            name="team-private",
            description="Private team channel",
            created_by="alice",
            created_by_project_id="proj_test1",
            scope="project",
            project_id="proj_test1",
            access_type="members"
        )
        channel_id = "proj_test1:team-private"
        
        # For members-only channels, we need to add members explicitly
        # Alice (creator) is already a member, now add Bob
        await api.db.sqlite.add_channel_member(
            channel_id,
            "bob",
            "proj_test2",
            invited_by="alice",
            source="invite",
            can_leave=True,
            can_send=True
        )
        # If no exception was raised, the operation succeeded
        
        # Bob can now access the members channel despite being from proj_test2
        is_member = await api.db.sqlite.is_channel_member(
            channel_id, "bob", "proj_test2"
        )
        assert is_member is True
    
    @pytest.mark.asyncio
    async def test_cannot_invite_to_private_channel(self, api, populated_db):
        """Cannot invite to private channels (DMs)."""
        # Create DM between Alice and Charlie
        dm_id = await api.db.create_or_get_dm_channel(
            "alice", "proj_test1", "charlie", None
        )
        
        # Alice tries to invite Bob to the DM
        with pytest.raises(ValueError, match="Cannot invite to private channels"):
            await api.invite_to_channel(
                channel_id=dm_id,
                invitee_name="bob",
                invitee_project_id="proj_test2",
                inviter_name="alice",
                inviter_project_id="proj_test1"
            )
    
    @pytest.mark.asyncio
    async def test_invite_to_members_channel(self, api, populated_db):
        """Members with invite permission can invite to members-only channels."""
        # Create a members channel with Alice as owner
        channel_id = await api.create_channel(
            name="team-only",
            scope="global",
            access_type="members",
            created_by="alice",
            created_by_project_id="proj_test1"
        )
        
        # Alice (creator with can_invite) invites Bob
        success = await api.invite_to_channel(
            channel_id=channel_id,
            invitee_name="bob",
            invitee_project_id="proj_test2",
            inviter_name="alice",
            inviter_project_id="proj_test1"
        )
        assert success is True
        
        # Bob is now a member
        is_member = await api.db.sqlite.is_channel_member(
            channel_id, "bob", "proj_test2"
        )
        assert is_member is True
    
    @pytest.mark.asyncio
    async def test_non_member_cannot_invite(self, api, populated_db):
        """Non-members cannot invite others to members-only channels."""
        # Use the existing members channel from populated_db
        # Bob (not a member of proj_test1:private) tries to invite Charlie
        with pytest.raises(ValueError, match="not a member"):
            await api.invite_to_channel(
                channel_id="proj_test1:private",
                invitee_name="charlie",
                invitee_project_id=None,
                inviter_name="bob",
                inviter_project_id="proj_test2"
            )
    
    # =========================================================================
    # Channel Discovery Tests (list_available_channels)
    # =========================================================================
    
    @pytest.mark.asyncio
    async def test_list_available_global_channels(self, api, populated_db):
        """All agents can see global channels."""
        channels = await api.list_channels(
            agent_name="bob",
            project_id="proj_test2",
            scope_filter="global"
        )
        
        channel_names = [c['name'] for c in channels]
        assert "general" in channel_names
        
        # Check joinability
        general = next(c for c in channels if c['name'] == 'general')
        assert general['can_join'] is True  # Open channel
        assert general['is_member'] is False  # Not yet joined
    
    @pytest.mark.asyncio
    async def test_list_available_project_channels(self, api, populated_db):
        """Agent sees channels from their project."""
        channels = await api.list_channels(
            agent_name="alice",
            project_id="proj_test1",
            scope_filter="project"
        )
        
        channel_names = [c['name'] for c in channels]
        assert "dev" in channel_names  # Open channel
        assert "private" in channel_names  # Members channel (visible but not joinable)
        
        # Check joinability
        dev = next(c for c in channels if c['name'] == 'dev')
        assert dev['can_join'] is True
        
        private = next(c for c in channels if c['name'] == 'private')
        assert private['can_join'] is False  # Members-only
    
    @pytest.mark.asyncio
    async def test_list_available_linked_project_channels(self, api, populated_db):
        """Agent sees channels from linked projects."""
        # Link projects
        await api.db.sqlite.add_project_link("proj_test1", "proj_test2")
        
        # Bob can now see proj_test1 channels
        channels = await api.list_channels(
            agent_name="bob",
            project_id="proj_test2",
            scope_filter="all"
        )
        
        # Should see proj_test1:dev
        proj1_channels = [c for c in channels if c.get('project_id') == 'proj_test1']
        assert len(proj1_channels) > 0
        
        dev = next((c for c in proj1_channels if c['name'] == 'dev'), None)
        assert dev is not None
        assert dev['can_join'] is True  # Can join open channel from linked project
    
    @pytest.mark.asyncio
    async def test_global_agent_sees_all_channels(self, api, populated_db):
        """Global agents can see all channels."""
        channels = await api.list_channels(
            agent_name="charlie",
            project_id=None,
            scope_filter="all"
        )
        
        # Should see both global and project channels
        assert any(c['scope'] == 'global' for c in channels)
        assert any(c['scope'] == 'project' for c in channels)
        
        # Can join open channels from any project
        proj_channels = [c for c in channels if c['scope'] == 'project' and c['access_type'] == 'open']
        for channel in proj_channels:
            assert channel['can_join'] is True
    
    # =========================================================================
    # Leave Channel Tests
    # =========================================================================
    
    @pytest.mark.asyncio
    async def test_leave_channel(self, api, populated_db):
        """Members can leave channels they joined."""
        # Alice joins and then leaves
        await api.join_channel("alice", "proj_test1", "global:general")
        
        success = await api.leave_channel(
            agent_name="alice",
            agent_project_id="proj_test1",
            channel_id="global:general"
        )
        assert success is True
        
        # No longer a member
        is_member = await api.db.sqlite.is_channel_member(
            "global:general", "alice", "proj_test1"
        )
        assert is_member is False
    
    @pytest.mark.asyncio
    async def test_cannot_leave_dm_channel(self, api, populated_db):
        """Cannot leave DM channels."""
        # Create DM
        dm_id = await api.db.create_or_get_dm_channel(
            "alice", "proj_test1", "bob", "proj_test2"
        )
        
        # Alice tries to leave
        success = await api.leave_channel(
            agent_name="alice",
            agent_project_id="proj_test1",
            channel_id=dm_id
        )
        assert success is False
    
    # =========================================================================
    # Default Channels Tests
    # =========================================================================
    
    @pytest.mark.asyncio
    async def test_apply_default_channels(self, api, populated_db):
        """New agents auto-join default channels."""
        # Create a default channel
        channel_id = await api.create_channel(
            name="welcome",
            scope="global",
            access_type="open",
            is_default=True,
            description="Welcome channel"
        )
        
        # Register new agent without auto-joining defaults
        await api.register_agent("diana", None, "New agent", auto_join_defaults=False)
        
        # Apply defaults
        count = await api.apply_default_channels(
            agent_name="diana",
            agent_project_id=None
        )
        assert count > 0
        
        # Diana is now in welcome channel
        is_member = await api.db.sqlite.is_channel_member(
            channel_id, "diana", None
        )
        assert is_member is True
    
    # =========================================================================
    # Message Permission Tests
    # =========================================================================
    
    @pytest.mark.asyncio
    async def test_send_message_requires_membership(self, api, populated_db):
        """Only members can send messages to channels."""
        # Alice is not in global:general
        with pytest.raises(ValueError, match="not a member"):
            await api.send_message(
                channel_id="global:general",
                sender_id="alice",
                sender_project_id="proj_test1",
                content="Hello"
            )
        
        # Add Alice to channel
        await api.db.sqlite.add_channel_member(
            "global:general",
            "alice",
            "proj_test1"
        )
        
        # Now she can send
        message_id = await api.send_message(
            channel_id="global:general",
            sender_id="alice",
            sender_project_id="proj_test1",
            content="Hello"
        )
        assert message_id > 0
    
    @pytest.mark.asyncio
    async def test_mention_validation(self, api, populated_db):
        """@mentions are validated against channel membership."""
        # Setup: Alice in channel, Bob not
        await api.db.sqlite.add_channel_member(
            "global:general",
            "alice",
            "proj_test1"
        )
        
        # Send a message with mentions to test validation
        msg_id = await api.send_message(
            channel_id="global:general",
            sender_id="alice",
            sender_project_id="proj_test1",
            content="Hey @alice @bob @nobody!",
            metadata={}
        )
        
        # Get the message to check mention validation in metadata
        messages = await api.get_agent_messages(
            agent_name="alice",
            agent_project_id="proj_test1",
            channel_id="global:general",
            limit=1
        )
        
        assert len(messages) == 1
        metadata = messages[0].get('metadata', {})
        
        if 'mentions' in metadata:
            mentions = metadata['mentions']
            # Check that mentions were validated
            valid_names = [m['name'] for m in mentions.get('valid', [])]
            invalid_names = [m['name'] for m in mentions.get('invalid', [])]
            
            assert 'alice' in valid_names  # Member
            assert 'bob' in invalid_names  # Not a member  
            assert 'nobody' in invalid_names  # Unknown agent


class TestChannelCreation:
    """Test suite for channel creation scenarios."""
    
    @pytest.mark.asyncio
    async def test_create_global_channel(self, api, populated_db):
        """Create a global channel."""
        channel_id = await api.create_channel(
            name="test-global",
            scope="global",
            access_type="open",
            created_by="alice",
            created_by_project_id="proj_test1"
        )
        
        assert channel_id == "global:test-global"
        
        # Verify channel exists
        channel = await api.get_channel(channel_id)
        assert channel is not None
        assert channel['scope'] == 'global'
        assert channel['access_type'] == 'open'
    
    @pytest.mark.asyncio
    async def test_create_project_channel(self, api, populated_db):
        """Create a project-scoped channel."""
        channel_id = await api.create_channel(
            name="test-project",
            scope="project",
            project_id="proj_test1",
            access_type="members",
            created_by="alice",
            created_by_project_id="proj_test1"
        )
        
        assert channel_id.startswith("proj_")
        assert "test-project" in channel_id
        
        # Creator should be added as member for members channels
        members = await api.list_channel_members(channel_id)
        member_names = [(m['agent_name'], m.get('agent_project_id')) for m in members]
        assert ('alice', 'proj_test1') in member_names
    
    @pytest.mark.asyncio
    async def test_create_dm_channel(self, api, populated_db):
        """Create a DM channel between two agents."""
        dm_id = await api.db.create_or_get_dm_channel(
            agent1_name="alice",
            agent1_project_id="proj_test1",
            agent2_name="bob",
            agent2_project_id="proj_test2"
        )
        
        assert dm_id.startswith("dm:")
        
        # Both agents should be members
        members = await api.list_channel_members(dm_id)
        member_names = [(m['agent_name'], m.get('agent_project_id')) for m in members]
        assert ('alice', 'proj_test1') in member_names
        assert ('bob', 'proj_test2') in member_names
        
        # Channel should be private
        channel = await api.get_channel(dm_id)
        assert channel['access_type'] == 'private'
    
    @pytest.mark.asyncio
    async def test_dm_channel_consistent_id(self, api, populated_db):
        """DM channel ID is consistent regardless of agent order."""
        # Create DM with agents in one order
        dm_id1 = await api.db.create_or_get_dm_channel(
            "alice", "proj_test1", "bob", "proj_test2"
        )
        
        # Create with reversed order - should get same ID
        dm_id2 = await api.db.create_or_get_dm_channel(
            "bob", "proj_test2", "alice", "proj_test1"
        )
        
        assert dm_id1 == dm_id2