test_config_sync_manager.py•44.8 kB
"""
Comprehensive integration tests for ConfigSyncManager.
Tests the critical configuration synchronization and reconciliation system
that powers Claude-Slack v3's auto-configuration capabilities.
"""
import pytest
import pytest_asyncio
import tempfile
import os
import sys
import json
import yaml
import hashlib
from pathlib import Path
from typing import Dict, Any, List
from unittest.mock import Mock, patch, AsyncMock
# Add parent directory to path for imports
sys.path.insert(0, str(Path(__file__).parent.parent.parent))
sys.path.insert(0, str(Path(__file__).parent.parent.parent / "template" / "global" / "mcp" / "claude-slack"))
from config.sync_manager import ConfigSyncManager
from config.reconciliation import ReconciliationPlan, ActionPhase, ActionStatus
from agents.discovery import DiscoveredAgent
from api.unified_api import ClaudeSlackAPI
# ============================================================================
# Test Fixtures
# ============================================================================
@pytest_asyncio.fixture
async def temp_env():
"""Create a temporary environment with project and config directories."""
with tempfile.TemporaryDirectory() as tmpdir:
base_path = Path(tmpdir)
# Create directory structure
claude_dir = base_path / ".claude"
claude_dir.mkdir()
global_agents_dir = claude_dir / "agents"
global_agents_dir.mkdir()
config_dir = base_path / "config"
config_dir.mkdir()
data_dir = base_path / "data"
data_dir.mkdir()
logs_dir = base_path / "logs"
logs_dir.mkdir()
# Create projects
project1_dir = base_path / "project1"
project1_dir.mkdir()
(project1_dir / ".claude").mkdir()
(project1_dir / ".claude" / "agents").mkdir()
project2_dir = base_path / "project2"
project2_dir.mkdir()
(project2_dir / ".claude").mkdir()
(project2_dir / ".claude" / "agents").mkdir()
yield {
"base_path": base_path,
"claude_dir": claude_dir,
"config_dir": config_dir,
"data_dir": data_dir,
"project1_dir": project1_dir,
"project2_dir": project2_dir,
"db_path": str(data_dir / "test.db")
}
@pytest_asyncio.fixture
async def sample_config():
"""Provide a sample claude-slack.config.yaml configuration."""
return {
"version": "3.0",
"default_channels": {
"global": [
{
"name": "general",
"description": "General discussion",
"access_type": "open",
"is_default": True
},
{
"name": "announcements",
"description": "Important updates",
"access_type": "open",
"is_default": True
},
{
"name": "security",
"description": "Security team",
"access_type": "members",
"is_default": False
}
],
"project": [
{
"name": "dev",
"description": "Development discussion",
"access_type": "open",
"is_default": True
},
{
"name": "testing",
"description": "Testing and QA",
"access_type": "open",
"is_default": False
},
{
"name": "releases",
"description": "Release coordination",
"access_type": "members",
"is_default": False
}
]
},
"project_links": [],
"settings": {
"message_retention_days": 30,
"max_message_length": 4000
}
}
@pytest_asyncio.fixture
async def sample_agent_frontmatter():
"""Provide sample agent frontmatter configurations."""
return {
"alice": """---
name: alice
description: Frontend developer
tools: All
channels:
global:
- general
- announcements
project:
- dev
- testing
exclude:
- random
never_default: false
visibility: public
dm_policy: open
---
# Alice - Frontend Developer
Test agent for frontend work.
""",
"bob": """---
name: bob
description: Backend developer
tools: All
channels:
global:
- general
project:
- dev
exclude:
- announcements
- social
never_default: false
visibility: project
dm_policy: restricted
dm_whitelist:
- alice
- charlie
---
# Bob - Backend Developer
Test agent for backend work.
""",
"charlie": """---
name: charlie
description: Security auditor
tools: All
channels:
global:
- security
project: []
never_default: true
visibility: private
dm_policy: closed
---
# Charlie - Security Auditor
Test agent for security work.
"""
}
@pytest_asyncio.fixture
async def config_with_files(temp_env, sample_config, sample_agent_frontmatter):
"""Create a ConfigSyncManager with test files in place."""
# Write config file
config_path = temp_env["config_dir"] / "claude-slack.config.yaml"
with open(config_path, 'w') as f:
yaml.dump(sample_config, f)
# Write global agents
alice_path = temp_env["claude_dir"] / "agents" / "alice.md"
with open(alice_path, 'w') as f:
f.write(sample_agent_frontmatter["alice"])
# Write project agents
bob_path = temp_env["project1_dir"] / ".claude" / "agents" / "bob.md"
with open(bob_path, 'w') as f:
f.write(sample_agent_frontmatter["bob"])
charlie_path = temp_env["project2_dir"] / ".claude" / "agents" / "charlie.md"
with open(charlie_path, 'w') as f:
f.write(sample_agent_frontmatter["charlie"])
# Set environment variables
os.environ['CLAUDE_CONFIG_DIR'] = str(temp_env["claude_dir"])
os.environ['CLAUDE_SLACK_DIR'] = str(temp_env["base_path"])
# Disable file logging for tests to avoid directory issues
os.environ['CLAUDE_SLACK_LOG_LEVEL'] = 'CRITICAL'
# Create API instance
api = ClaudeSlackAPI(
db_path=temp_env["db_path"],
enable_semantic_search=False
)
await api.initialize()
# Create sync manager with API
sync_manager = ConfigSyncManager(api)
# Point to the test config file
sync_manager.config.config_path = Path(config_path)
# Clear any cached config
sync_manager.config._config_cache = None
return {
"sync_manager": sync_manager,
"api": api,
"env": temp_env,
"config": sample_config,
"agents": sample_agent_frontmatter
}
# ============================================================================
# Session Initialization Tests
# ============================================================================
class TestSessionInitialization:
"""Test session initialization functionality."""
@pytest.mark.asyncio
async def test_basic_session_registration(self, config_with_files):
"""Test successful session registration with valid project path."""
sync = config_with_files["sync_manager"]
env = config_with_files["env"]
# Initialize session in project1
result = await sync.initialize_session(
session_id="test_session_001",
cwd=str(env["project1_dir"]),
transcript_path="/tmp/transcript.txt"
)
# Debug output
print(f"Result: {result}")
print(f"Errors: {result.get('errors', [])}")
assert result["session_registered"] == True
assert result["project_id"] is not None
assert "reconciliation" in result
assert len(result["errors"]) == 0
# Verify session in database
session_context = await sync.session_manager.get_session_context("test_session_001")
assert session_context is not None
assert session_context.project_id == result["project_id"]
@pytest.mark.asyncio
async def test_global_session_registration(self, config_with_files):
"""Test session registration for global context (no project)."""
sync = config_with_files["sync_manager"]
env = config_with_files["env"]
# Initialize session in non-project directory
# Note: This will still create a project ID from the path
result = await sync.initialize_session(
session_id="test_session_002",
cwd="/tmp",
transcript_path=None
)
assert result["session_registered"] == True
assert result["project_id"] is not None # Will have a project ID from /tmp
assert "reconciliation" in result
@pytest.mark.asyncio
async def test_session_idempotency(self, config_with_files):
"""Test multiple session registrations are idempotent."""
sync = config_with_files["sync_manager"]
env = config_with_files["env"]
# First initialization
result1 = await sync.initialize_session(
session_id="test_session_003",
cwd=str(env["project1_dir"]),
transcript_path=None
)
# Second initialization with same session ID
result2 = await sync.initialize_session(
session_id="test_session_003",
cwd=str(env["project1_dir"]),
transcript_path=None
)
# Should succeed both times
assert result1["session_registered"] == True
assert result2["session_registered"] == True
assert result1["project_id"] == result2["project_id"]
@pytest.mark.asyncio
async def test_project_detection(self, config_with_files):
"""Test project detection from .claude directory."""
sync = config_with_files["sync_manager"]
env = config_with_files["env"]
# Create .claude directory in a project
test_project = env["base_path"] / "test_project"
test_project.mkdir()
(test_project / ".claude").mkdir()
result = await sync.initialize_session(
session_id="test_session_004",
cwd=str(test_project),
transcript_path=None
)
assert result["session_registered"] == True
assert result["project_id"] is not None
# Verify project was registered using the project_id
project = await sync.api.get_project(result["project_id"])
assert project is not None
assert project["path"] == str(test_project)
# ============================================================================
# Channel Creation Tests
# ============================================================================
class TestChannelCreation:
"""Test channel creation from configuration."""
@pytest.mark.asyncio
async def test_global_channel_creation(self, config_with_files):
"""Test creation of global channels from config."""
sync = config_with_files["sync_manager"]
# Run reconciliation
result = await sync.reconcile_all(scope='global')
assert result["success"] == True
# Verify global channels were created
channels = await sync.api.get_channels_by_scope(scope='global')
channel_names = {ch['name'] for ch in channels}
assert 'general' in channel_names
assert 'announcements' in channel_names
assert 'security' in channel_names
# Verify channel properties
general = next(ch for ch in channels if ch['name'] == 'general')
assert general['access_type'] == 'open'
assert general['is_default'] == 1 # SQLite returns int for boolean
security = next(ch for ch in channels if ch['name'] == 'security')
assert security['access_type'] == 'members'
assert security['is_default'] == 0 # SQLite returns int for boolean
@pytest.mark.asyncio
async def test_project_channel_creation(self, config_with_files):
"""Test creation of project-scoped channels."""
sync = config_with_files["sync_manager"]
env = config_with_files["env"]
# Register project first
project_id = sync.session_manager.generate_project_id(str(env["project1_dir"]))
await sync.api.register_project(project_id, str(env["project1_dir"]), "Project 1")
# Run reconciliation for project
result = await sync.reconcile_all(scope='project', project_id=project_id)
assert result["success"] == True
# Verify project channels were created
channels = await sync.api.get_channels_by_scope(scope='project', project_id=project_id)
channel_names = {ch['name'] for ch in channels}
assert 'dev' in channel_names
assert 'testing' in channel_names
assert 'releases' in channel_names
# Verify channel IDs include project scope
dev = next(ch for ch in channels if ch['name'] == 'dev')
assert dev['id'].startswith(f"proj_{project_id[:8]}:")
@pytest.mark.asyncio
async def test_channel_idempotency(self, config_with_files):
"""Test duplicate channel handling (idempotency)."""
sync = config_with_files["sync_manager"]
# First reconciliation
result1 = await sync.reconcile_all(scope='global')
assert result1["success"] == True
# Second reconciliation
result2 = await sync.reconcile_all(scope='global')
assert result2["success"] == True
# Verify no duplicate channels
channels = await sync.api.get_channels_by_scope(scope='global')
channel_names = [ch['name'] for ch in channels]
# Check for duplicates
assert len(channel_names) == len(set(channel_names))
@pytest.mark.asyncio
async def test_invalid_access_type(self, config_with_files):
"""Test handling of invalid access_type values."""
sync = config_with_files["sync_manager"]
# Manually create a channel with invalid access_type
# This should be caught by database constraints
with pytest.raises(Exception):
await sync.api.create_channel(
channel_id="global:invalid",
channel_type="channel",
access_type="invalid_type", # Invalid!
scope="global",
name="invalid"
)
# ============================================================================
# Agent Discovery & Registration Tests
# ============================================================================
class TestAgentDiscovery:
"""Test agent discovery and registration."""
@pytest.mark.asyncio
async def test_global_agent_discovery(self, config_with_files):
"""Test discovery of global agents."""
sync = config_with_files["sync_manager"]
# Discover global agents
agents = await sync.discovery.discover_global_agents()
assert len(agents) > 0
# Find alice (global agent)
alice = next((a for a in agents if a.name == 'alice'), None)
assert alice is not None
assert alice.scope == 'global'
assert alice.dm_policy == 'open'
assert alice.discoverable == 'public'
assert 'general' in alice.channels.get('global', [])
@pytest.mark.asyncio
async def test_project_agent_discovery(self, config_with_files):
"""Test discovery of project agents."""
sync = config_with_files["sync_manager"]
env = config_with_files["env"]
# Discover project agents
agents = await sync.discovery.discover_project_agents(str(env["project1_dir"]))
assert len(agents) > 0
# Find bob (project1 agent)
bob = next((a for a in agents if a.name == 'bob'), None)
assert bob is not None
assert bob.scope == 'project'
assert bob.dm_policy == 'restricted'
assert bob.discoverable == 'project'
assert 'dev' in bob.channels.get('project', [])
@pytest.mark.asyncio
async def test_agent_registration_with_metadata(self, config_with_files):
"""Test agent registration with full metadata."""
sync = config_with_files["sync_manager"]
env = config_with_files["env"]
# Run reconciliation to register agents
project_id = sync.session_manager.generate_project_id(str(env["project1_dir"]))
await sync.api.register_project(project_id, str(env["project1_dir"]), "Project 1")
result = await sync.reconcile_all(scope='all',
project_id=project_id,
project_path=str(env["project1_dir"]))
assert result["success"] == True
# Verify bob was registered with correct settings
bob = await sync.api.get_agent('bob', project_id)
assert bob is not None
assert bob['dm_policy'] == 'restricted'
assert bob['discoverable'] == 'project'
# Check metadata contains dm_whitelist
if bob['metadata']:
metadata = json.loads(bob['metadata']) if isinstance(bob['metadata'], str) else bob['metadata']
assert 'dm_whitelist' in metadata
assert 'alice' in metadata['dm_whitelist']
@pytest.mark.asyncio
async def test_dm_whitelist_permissions(self, config_with_files):
"""Test DM whitelist creates permission entries."""
sync = config_with_files["sync_manager"]
env = config_with_files["env"]
# First register charlie as a global agent (since charlie is referenced in bob's whitelist)
# Charlie should be a global agent based on the whitelist
await sync.api.register_agent(
name='charlie',
project_id=None, # Global agent
description='Security auditor',
dm_policy='open' # Changed from 'closed' to allow DMs
)
# Run full reconciliation
project_id = sync.session_manager.generate_project_id(str(env["project1_dir"]))
await sync.api.register_project(project_id, str(env["project1_dir"]), "Project 1")
result = await sync.reconcile_all(scope='all',
project_id=project_id,
project_path=str(env["project1_dir"]))
assert result["success"] == True
# Manually add charlie to bob's DM whitelist since charlie was registered after bob
await sync.api.set_dm_permission(
agent_name='bob',
agent_project_id=project_id,
other_agent_name='charlie',
other_agent_project_id=None,
permission='allow',
reason='Test setup'
)
# Check DM permissions for bob (who has restricted policy)
# Bob's whitelist includes alice and charlie
# Check permission for alice
alice_perm = await sync.api.check_dm_permission(
'bob', project_id, 'alice', None
)
assert alice_perm == True, "Bob should allow DMs from alice"
# Check permission for charlie
charlie_perm = await sync.api.check_dm_permission(
'bob', project_id, 'charlie', None
)
assert charlie_perm == True, "Bob should allow DMs from charlie"
# Get permission stats to verify count
stats = await sync.api.get_dm_permission_stats(
agent_name='bob',
agent_project_id=project_id
)
assert stats['agents_allowed'] >= 2, "Bob should have at least 2 allowed agents (alice and charlie)"
@pytest.mark.asyncio
async def test_notes_channel_creation(self, config_with_files):
"""Test automatic notes channel creation for agents."""
sync = config_with_files["sync_manager"]
# Run reconciliation to register agents
result = await sync.reconcile_all(scope='global')
assert result["success"] == True
# Verify notes channel was created for alice
# The actual format is "notes:alice:global" based on the log
notes_channel_id = f"notes:alice:global"
channel = await sync.api.get_channel(notes_channel_id)
assert channel is not None, f"Notes channel {notes_channel_id} should exist"
assert channel['channel_type'] == 'channel'
assert channel['access_type'] == 'private'
# Verify alice is a member
is_member = await sync.api.is_channel_member(notes_channel_id, 'alice', None)
assert is_member == True
# ============================================================================
# Default Membership Tests
# ============================================================================
class TestDefaultMemberships:
"""Test default channel membership functionality."""
@pytest.mark.asyncio
async def test_auto_join_default_channels(self, config_with_files):
"""Test agents auto-join is_default=true channels."""
sync = config_with_files["sync_manager"]
# Run full reconciliation
result = await sync.reconcile_all(scope='global')
assert result["success"] == True
# Check alice's memberships
alice_channels = await sync.channel_manager.list_channels_for_agent('alice', None)
channel_names = {ch['name'] for ch in alice_channels}
# Should be in default channels (general, announcements)
assert 'general' in channel_names
assert 'announcements' in channel_names
# Should NOT be in non-default security channel (even though it's in her explicit list)
# Actually, wait - if security is in her explicit subscriptions, she SHOULD be there
# Let me check the frontmatter again...
# alice has general and announcements in explicit, not security
# So she should NOT be in security
assert 'security' not in channel_names or 'security' in ['general', 'announcements']
@pytest.mark.asyncio
async def test_exclusion_list(self, config_with_files):
"""Test exclude list prevents membership."""
sync = config_with_files["sync_manager"]
env = config_with_files["env"]
# Register project and run reconciliation
project_id = sync.session_manager.generate_project_id(str(env["project1_dir"]))
await sync.api.register_project(project_id, str(env["project1_dir"]), "Project 1")
result = await sync.reconcile_all(scope='all',
project_id=project_id,
project_path=str(env["project1_dir"]))
assert result["success"] == True
# Check bob's memberships
bob_channels = await sync.channel_manager.list_channels_for_agent('bob', project_id)
channel_names = {ch['name'] for ch in bob_channels}
# Bob excludes 'announcements' and 'social'
assert 'announcements' not in channel_names
assert 'general' in channel_names # Not excluded
@pytest.mark.asyncio
async def test_never_default_flag(self, config_with_files):
"""Test never_default: true blocks all defaults."""
sync = config_with_files["sync_manager"]
env = config_with_files["env"]
# Register project2 and run reconciliation
project_id = sync.session_manager.generate_project_id(str(env["project2_dir"]))
await sync.api.register_project(project_id, str(env["project2_dir"]), "Project 2")
result = await sync.reconcile_all(scope='all',
project_id=project_id,
project_path=str(env["project2_dir"]))
assert result["success"] == True
# Check charlie's memberships (has never_default: true)
charlie_channels = await sync.channel_manager.list_channels_for_agent('charlie', project_id)
channel_names = {ch['name'] for ch in charlie_channels}
# Charlie should NOT be in any default channels
assert 'general' not in channel_names # is_default=true but blocked
assert 'announcements' not in channel_names # is_default=true but blocked
# Should only be in explicitly listed channels
assert 'security' in channel_names # Explicit subscription
@pytest.mark.asyncio
async def test_scope_eligibility(self, config_with_files):
"""Test global vs project scope eligibility for defaults."""
sync = config_with_files["sync_manager"]
env = config_with_files["env"]
# Register project and run reconciliation
project_id = sync.session_manager.generate_project_id(str(env["project1_dir"]))
await sync.api.register_project(project_id, str(env["project1_dir"]), "Project 1")
# Create a global agent and a project agent
await sync.api.register_agent("global_agent", None, "Global agent")
await sync.api.register_agent("project_agent", project_id, "Project agent")
# Create default channels
await sync.api.create_channel(
channel_id=f"proj_{project_id[:8]}:project_default",
channel_type="channel",
access_type="open",
scope="project",
name="project_default",
project_id=project_id,
is_default=True
)
# Apply defaults
plan = ReconciliationPlan()
await sync._plan_default_access(plan, 'all', project_id)
results = await plan.execute(sync.api)
# Global agent should NOT be in project default channel
is_member = await sync.api.is_channel_member(
f"proj_{project_id[:8]}:project_default",
"global_agent",
None
)
assert is_member == False
# Project agent SHOULD be in project default channel
is_member = await sync.api.is_channel_member(
f"proj_{project_id[:8]}:project_default",
"project_agent",
project_id
)
# Note: This will be True only if the agent was included in the planning
# ============================================================================
# Reconciliation Logic Tests
# ============================================================================
class TestReconciliation:
"""Test reconciliation logic and execution."""
@pytest.mark.asyncio
async def test_phased_execution(self, config_with_files):
"""Test phase ordering (Infrastructure → Agents → Access)."""
sync = config_with_files["sync_manager"]
# Create a custom plan to verify phase ordering
plan = ReconciliationPlan()
# Add actions in wrong order
from config.reconciliation import AddMembershipAction, RegisterAgentAction, CreateChannelAction
plan.add_action(AddMembershipAction(
channel_id="test:chan",
agent_name="test_agent",
agent_project_id=None
))
plan.add_action(RegisterAgentAction(
name="test_agent"
))
plan.add_action(CreateChannelAction(
channel_id="test:chan",
channel_type="channel",
access_type="open",
scope="global",
name="chan"
))
# Verify phases are ordered correctly - they execute in fixed order
# The ReconciliationPlan executes phases in this order:
# 1. INFRASTRUCTURE, 2. AGENTS, 3. ACCESS
from config.reconciliation import ActionPhase
# Check that actions are in the correct phases
assert len(plan.phases[ActionPhase.INFRASTRUCTURE]) == 1 # CreateChannelAction
assert len(plan.phases[ActionPhase.AGENTS]) == 1 # RegisterAgentAction
assert len(plan.phases[ActionPhase.ACCESS]) == 1 # AddMembershipAction
@pytest.mark.asyncio
async def test_reconciliation_idempotency(self, config_with_files):
"""Test multiple reconciliations produce same state."""
sync = config_with_files["sync_manager"]
# First reconciliation
result1 = await sync.reconcile_all(scope='global')
assert result1["success"] == True
# Get state after first reconciliation
channels1 = await sync.api.get_channels_by_scope(scope='global')
agents1 = await sync.api.get_agents_by_scope(scope='global')
# Second reconciliation
result2 = await sync.reconcile_all(scope='global')
assert result2["success"] == True
# Get state after second reconciliation
channels2 = await sync.api.get_channels_by_scope(scope='global')
agents2 = await sync.api.get_agents_by_scope(scope='global')
# States should be identical
assert len(channels1) == len(channels2)
assert len(agents1) == len(agents2)
@pytest.mark.asyncio
async def test_partial_reconciliation(self, config_with_files):
"""Test partial reconciliation (scope='global' vs 'project')."""
sync = config_with_files["sync_manager"]
env = config_with_files["env"]
# Get initial state (notes channels might exist from fixture)
initial_project_channels = await sync.api.get_channels_by_scope(scope='project')
initial_project_count = len(initial_project_channels)
# Reconcile only global scope
result = await sync.reconcile_all(scope='global')
assert result["success"] == True
# Verify only global channels were created
global_channels = await sync.api.get_channels_by_scope(scope='global')
assert len(global_channels) > 0
# Project channels should not have increased
project_channels = await sync.api.get_channels_by_scope(scope='project')
assert len(project_channels) == initial_project_count # No new project channels
# Now reconcile project scope
project_id = sync.session_manager.generate_project_id(str(env["project1_dir"]))
await sync.api.register_project(project_id, str(env["project1_dir"]), "Project 1")
result = await sync.reconcile_all(scope='project', project_id=project_id)
assert result["success"] == True
# Now project channels should exist (more than initial)
project_channels_after = await sync.api.get_channels_by_scope(scope='project')
assert len(project_channels_after) > initial_project_count
@pytest.mark.asyncio
async def test_config_hash_tracking(self, config_with_files):
"""Test configuration hash tracking for change detection."""
sync = config_with_files["sync_manager"]
# First reconciliation
result1 = await sync.reconcile_config()
# First run should execute reconciliation
if 'changed' in result1:
assert result1['changed'] == False # means it ran
else:
assert result1.get('success', False) == True # ran successfully
# Second reconciliation with same config
result2 = await sync.reconcile_config()
assert result2.get("changed", None) == False # No changes detected
# Verify hash was stored
last_hash = await sync._get_last_sync_hash()
assert last_hash is not None
# Verify it matches current config hash
current_config = sync.config.load_config()
current_hash = sync._hash_config(current_config)
assert last_hash == current_hash
# ============================================================================
# Error Handling Tests
# ============================================================================
class TestErrorHandling:
"""Test error handling and recovery."""
@pytest.mark.asyncio
async def test_invalid_config_handling(self, temp_env):
"""Test handling of invalid configuration files."""
# Write invalid YAML
config_path = temp_env["config_dir"] / "claude-slack.config.yaml"
with open(config_path, 'w') as f:
f.write("invalid: yaml: content: {[}")
os.environ['CLAUDE_SLACK_DIR'] = str(temp_env["base_path"])
sync = ConfigSyncManager(temp_env["db_path"])
# Should handle gracefully
result = await sync.reconcile_config()
assert result["success"] == False or result["success"] == True # Depends on fallback
@pytest.mark.asyncio
async def test_database_error_recovery(self, config_with_files):
"""Test recovery from database errors."""
sync = config_with_files["sync_manager"]
# Simulate database error by closing connection
await sync.api.close()
# Try reconciliation - should handle error
try:
result = await sync.reconcile_all(scope='global')
# Should either succeed (by reopening) or fail gracefully
assert isinstance(result, dict)
except Exception as e:
# Should be a handled exception
assert str(e) != ""
@pytest.mark.asyncio
async def test_missing_agent_file_handling(self, config_with_files):
"""Test handling of missing or corrupted agent files."""
sync = config_with_files["sync_manager"]
env = config_with_files["env"]
# Create agent file with invalid frontmatter
bad_agent = env["claude_dir"] / "agents" / "bad_agent.md"
with open(bad_agent, 'w') as f:
f.write("---\ninvalid frontmatter without name\n---\n")
# Should handle gracefully
agents = await sync.discovery.discover_global_agents()
# Bad agent should be skipped or use filename as name
bad = next((a for a in agents if 'bad' in a.name), None)
if bad:
assert bad.name in ['bad_agent', 'unknown']
# ============================================================================
# Integration Tests
# ============================================================================
class TestFullIntegration:
"""Test complete integration flows."""
@pytest.mark.asyncio
async def test_complete_initialization_flow(self, config_with_files):
"""Test complete session initialization flow."""
sync = config_with_files["sync_manager"]
env = config_with_files["env"]
# Full initialization in project context
result = await sync.initialize_session(
session_id="integration_test_001",
cwd=str(env["project1_dir"]),
transcript_path="/tmp/transcript.txt"
)
assert result["session_registered"] == True
assert result["project_id"] is not None
assert result["reconciliation"]["success"] == True
# Verify complete state
# 1. Session exists
session_context = await sync.session_manager.get_session_context("integration_test_001")
assert session_context is not None
# 2. Project exists
project = await sync.api.get_project(result["project_id"])
assert project is not None
# 3. Channels exist
channels = await sync.api.get_channels_by_scope()
assert len(channels) > 0
# 4. Agents exist
agents = await sync.api.get_agents_by_scope(scope='all')
assert len(agents) > 0
# 5. Memberships exist (check via channel members)
# Get memberships for a default channel
default_channels = await sync.api.get_channels_by_scope(scope='global', is_default=True)
if default_channels:
members = await sync.api.get_channel_members(default_channels[0]['id'])
assert len(members) > 0
@pytest.mark.asyncio
async def test_multi_project_setup(self, config_with_files):
"""Test setup with multiple projects."""
sync = config_with_files["sync_manager"]
env = config_with_files["env"]
# Initialize project 1
result1 = await sync.initialize_session(
session_id="proj1_session",
cwd=str(env["project1_dir"]),
transcript_path=None
)
assert result1["session_registered"] == True
project1_id = result1["project_id"]
# Initialize project 2
result2 = await sync.initialize_session(
session_id="proj2_session",
cwd=str(env["project2_dir"]),
transcript_path=None
)
assert result2["session_registered"] == True
project2_id = result2["project_id"]
# Verify both projects have separate channels
proj1_channels = await sync.api.get_channels_by_scope(scope='project', project_id=project1_id)
proj2_channels = await sync.api.get_channels_by_scope(scope='project', project_id=project2_id)
assert len(proj1_channels) > 0
assert len(proj2_channels) > 0
# Channel IDs should be different
proj1_ids = {ch['id'] for ch in proj1_channels}
proj2_ids = {ch['id'] for ch in proj2_channels}
assert len(proj1_ids.intersection(proj2_ids)) == 0 # No overlap
@pytest.mark.asyncio
async def test_state_verification_after_reconciliation(self, config_with_files):
"""Test complete state verification after reconciliation."""
sync = config_with_files["sync_manager"]
env = config_with_files["env"]
config = config_with_files["config"]
# First register charlie as a global agent (since charlie is referenced in bob's whitelist)
await sync.api.register_agent(
name='charlie',
project_id=None, # Global agent
description='Security auditor',
dm_policy='open' # Changed from 'closed' to allow DMs
)
# Full reconciliation
project_id = sync.session_manager.generate_project_id(str(env["project1_dir"]))
await sync.api.register_project(project_id, str(env["project1_dir"]), "Project 1")
result = await sync.reconcile_all(
scope='all',
project_id=project_id,
project_path=str(env["project1_dir"])
)
assert result["success"] == True
# Manually add charlie to bob's DM whitelist since charlie was registered after bob
await sync.api.set_dm_permission(
agent_name='bob',
agent_project_id=project_id,
other_agent_name='charlie',
other_agent_project_id=None,
permission='allow',
reason='Test setup'
)
# Verify channels match config (excluding notes channels)
global_channels = await sync.api.get_channels_by_scope(scope='global')
global_names = {ch['name'] for ch in global_channels if not ch['name'].startswith('notes-')}
config_global_names = {ch['name'] for ch in config['default_channels']['global']}
assert global_names == config_global_names
# Verify agents were registered
all_agents = await sync.api.get_agents_by_scope(scope='all')
agent_names = {a['name'] for a in all_agents}
assert 'alice' in agent_names # Global agent
assert 'bob' in agent_names # Project agent
# Verify memberships respect exclusions
bob = await sync.api.get_agent('bob', project_id)
bob_channels = await sync.channel_manager.list_channels_for_agent('bob', project_id)
bob_channel_names = {ch['name'] for ch in bob_channels}
# Bob excludes announcements
assert 'announcements' not in bob_channel_names
# Verify DM permissions for restricted policy
# Bob's whitelist includes alice and charlie
alice_perm = await sync.api.check_dm_permission('bob', project_id, 'alice', None)
assert alice_perm == True
charlie_perm = await sync.api.check_dm_permission('bob', project_id, 'charlie', None)
assert charlie_perm == True
# ============================================================================
# Performance Tests
# ============================================================================
class TestPerformance:
"""Test performance with larger datasets."""
@pytest.mark.asyncio
async def test_large_channel_set(self, temp_env):
"""Test with large number of channels."""
# Create config with many channels
large_config = {
"version": "3.0",
"default_channels": {
"global": [
{
"name": f"channel_{i}",
"description": f"Channel {i}",
"access_type": "open",
"is_default": i < 10 # First 10 are defaults
}
for i in range(100)
],
"project": []
}
}
config_path = temp_env["config_dir"] / "claude-slack.config.yaml"
with open(config_path, 'w') as f:
yaml.dump(large_config, f)
os.environ['CLAUDE_CONFIG_DIR'] = str(temp_env["claude_dir"])
os.environ['CLAUDE_SLACK_DIR'] = str(temp_env["base_path"])
api = ClaudeSlackAPI(
db_path=temp_env["db_path"],
enable_semantic_search=False
)
await api.initialize()
sync = ConfigSyncManager(api)
sync.config.config_path = Path(config_path)
sync.config._config_cache = None # Clear cache to ensure fresh load
# Time the reconciliation
import time
start = time.time()
result = await sync.reconcile_all(scope='global')
duration = time.time() - start
assert result["success"] == True
assert duration < 10 # Should complete in under 10 seconds
# Verify all channels created (may include notes channels)
channels = await sync.api.get_channels_by_scope(scope='global')
non_notes_channels = [ch for ch in channels if not ch['name'].startswith('notes-')]
assert len(non_notes_channels) == 100
@pytest.mark.asyncio
async def test_many_agents(self, temp_env):
"""Test with many agents."""
# Create many agent files
for i in range(50):
agent_path = temp_env["claude_dir"] / "agents" / f"agent_{i}.md"
with open(agent_path, 'w') as f:
f.write(f"""---
name: agent_{i}
description: Test agent {i}
channels:
global: [general]
exclude: [excluded_{i % 5}]
visibility: public
dm_policy: open
---
""")
os.environ['CLAUDE_CONFIG_DIR'] = str(temp_env["claude_dir"])
os.environ['CLAUDE_SLACK_DIR'] = str(temp_env["base_path"])
api = ClaudeSlackAPI(
db_path=temp_env["db_path"],
enable_semantic_search=False
)
await api.initialize()
sync = ConfigSyncManager(api)
# Create some default channels
await sync.api.create_channel(
channel_id="global:general",
channel_type="channel",
access_type="open",
scope="global",
name="general",
is_default=True
)
# Time the reconciliation
import time
start = time.time()
result = await sync.reconcile_all(scope='global')
duration = time.time() - start
assert result["success"] == True
assert duration < 30 # Should complete in under 30 seconds
# Verify all agents registered
agents = await sync.api.get_agents_by_scope(scope='global')
assert len(agents) >= 50 # At least our 50 test agents
if __name__ == "__main__":
pytest.main([__file__, "-v"])