"""Integration tests for CompactedResult compaction in MCP server. Tests verify that large result sets are properly compacted to prevent context overflow. These tests cover: - CompactedResult structure and validation - Compaction threshold behavior (triggers at >20 results) - Preview count consistency (shows 5 items) - Total count accuracy - Both ready() and list() MCP tools Note: Tests focus on the compaction logic itself, which is independent of the FastMCP decorator. The actual tool behavior is tested via _apply_compaction(). """ from datetime import datetime, timezone import pytest from beads_mcp.models import CompactedResult, Issue, IssueMinimal @pytest.fixture def sample_issues(count=25): """Create a list of sample issues for testing compaction. Default creates 25 issues to exceed the COMPACTION_THRESHOLD (20). """ now = datetime(2024, 1, 1, 0, 0, 0, tzinfo=timezone.utc) issues = [] for i in range(count): issue = Issue( id=f"bd-{i:04d}", title=f"Issue {i}", description=f"Description {i}", status="open" if i % 2 == 0 else "in_progress", priority=i % 5, issue_type="bug" if i % 3 == 0 else "task", created_at=now, updated_at=now, ) issues.append(issue) return issues def _to_minimal_test(issue: Issue) -> IssueMinimal: """Convert Issue to IssueMinimal - copy of server's _to_minimal for testing.""" return IssueMinimal( id=issue.id, title=issue.title, status=issue.status, priority=issue.priority, issue_type=issue.issue_type, assignee=issue.assignee, labels=issue.labels, dependency_count=issue.dependency_count, dependent_count=issue.dependent_count, ) def _apply_compaction( minimal_issues: list[IssueMinimal], threshold: int = 20, preview_count: int = 5, hint_prefix: str = "", ) -> list[IssueMinimal] | CompactedResult: """Apply compaction logic - copy of server's compaction for testing. This function contains the core compaction logic that's used in the MCP tools. """ if len(minimal_issues) > threshold: return CompactedResult( compacted=True, total_count=len(minimal_issues), preview=minimal_issues[:preview_count], preview_count=preview_count, hint=f"{hint_prefix}Showing {preview_count} of {len(minimal_issues)} issues. Use show(issue_id) for full details.", ) return minimal_issues @pytest.fixture(autouse=True) def reset_connection_pool(): """Reset connection pool before and after each test.""" from beads_mcp import tools tools._connection_pool.clear() yield tools._connection_pool.clear() class TestCompactedResultStructure: """Test CompactedResult model structure and validation.""" def test_compacted_result_model_valid(self): """Test CompactedResult model with valid data.""" minimal_issues = [ IssueMinimal( id="bd-1", title="Test 1", status="open", priority=1, issue_type="bug", ), IssueMinimal( id="bd-2", title="Test 2", status="open", priority=2, issue_type="task", ), ] result = CompactedResult( compacted=True, total_count=25, preview=minimal_issues, preview_count=2, hint="Use show(issue_id) for full details", ) assert result.compacted is True assert result.total_count == 25 assert result.preview_count == 2 assert len(result.preview) == 2 assert result.preview[0].id == "bd-1" assert result.hint == "Use show(issue_id) for full details" def test_compacted_result_default_hint(self): """Test CompactedResult uses default hint message.""" result = CompactedResult( compacted=True, total_count=30, preview=[], preview_count=0, ) assert result.hint == "Use show(issue_id) for full issue details" class TestCompactionLogic: """Test core compaction logic.""" def test_below_threshold_returns_list(self): """Test with <20 results returns list (no compaction).""" issues = [ Issue( id=f"bd-{i:04d}", title=f"Issue {i}", description="", status="open", priority=1, issue_type="task", created_at=datetime.now(timezone.utc), updated_at=datetime.now(timezone.utc), ) for i in range(10) ] minimal = [_to_minimal_test(i) for i in issues] result = _apply_compaction(minimal) assert isinstance(result, list) assert len(result) == 10 assert all(isinstance(issue, IssueMinimal) for issue in result) def test_above_threshold_returns_compacted(self, sample_issues): """Test with >20 results returns CompactedResult.""" minimal = [_to_minimal_test(i) for i in sample_issues] result = _apply_compaction(minimal) assert isinstance(result, CompactedResult) assert result.compacted is True assert result.total_count == 25 assert result.preview_count == 5 assert len(result.preview) == 5 assert all(isinstance(issue, IssueMinimal) for issue in result.preview) def test_compaction_preserves_order(self, sample_issues): """Test compaction shows first N issues in order.""" minimal = [_to_minimal_test(i) for i in sample_issues] result = _apply_compaction(minimal) # First 5 issues should match original order preview = result.preview for i, issue in enumerate(preview): assert issue.id == f"bd-{i:04d}" def test_exactly_threshold_no_compaction(self): """Test with exactly 20 items (at threshold) - no compaction.""" issues = [ Issue( id=f"bd-{i:04d}", title=f"Issue {i}", description="", status="open", priority=i % 5, issue_type="task", created_at=datetime.now(timezone.utc), updated_at=datetime.now(timezone.utc), ) for i in range(20) ] minimal = [_to_minimal_test(i) for i in issues] result = _apply_compaction(minimal) # Should return list since len(issues) == 20, not > 20 assert isinstance(result, list) assert len(result) == 20 def test_one_over_threshold_compacts(self): """Test with 21 items (just over threshold).""" issues = [ Issue( id=f"bd-{i:04d}", title=f"Issue {i}", description="", status="open", priority=i % 5, issue_type="task", created_at=datetime.now(timezone.utc), updated_at=datetime.now(timezone.utc), ) for i in range(21) ] minimal = [_to_minimal_test(i) for i in issues] result = _apply_compaction(minimal) # Should compact assert isinstance(result, CompactedResult) assert result.compacted is True assert result.total_count == 21 assert result.preview_count == 5 def test_empty_result_set(self): """Test with empty results.""" result = _apply_compaction([]) assert isinstance(result, list) assert len(result) == 0 class TestCompactedResultHint: """Test hint field behavior in CompactedResult.""" def test_compacted_result_hint_present(self, sample_issues): """Test compacted result includes helpful hint.""" minimal = [_to_minimal_test(i) for i in sample_issues] result = _apply_compaction(minimal) assert result.hint is not None assert isinstance(result.hint, str) assert len(result.hint) > 0 # Hint should guide user on how to proceed assert "show" in result.hint.lower() def test_compacted_result_hint_with_custom_prefix(self, sample_issues): """Test hint can have custom prefix.""" minimal = [_to_minimal_test(i) for i in sample_issues] custom_hint = "Ready work: " result = _apply_compaction(minimal, hint_prefix=custom_hint) assert result.hint.startswith(custom_hint) class TestCompactedResultDataTypes: """Test type conversions and data format in CompactedResult.""" def test_preview_items_are_minimal_format(self, sample_issues): """Test that preview items use IssueMinimal format.""" minimal = [_to_minimal_test(i) for i in sample_issues] result = _apply_compaction(minimal) # Check preview items have IssueMinimal fields only for issue in result.preview: assert isinstance(issue, IssueMinimal) assert hasattr(issue, "id") assert hasattr(issue, "title") assert hasattr(issue, "status") assert hasattr(issue, "priority") def test_total_count_includes_all_results(self, sample_issues): """Test total_count reflects all matching issues, not just preview.""" minimal = [_to_minimal_test(i) for i in sample_issues] result = _apply_compaction(minimal) # total_count should be 25, preview_count should be 5 assert result.total_count > result.preview_count assert result.preview_count == 5 assert len(result.preview) == result.preview_count class TestCompactionConsistency: """Test consistency of compaction behavior across different scenarios.""" def test_multiple_calls_same_behavior(self, sample_issues): """Test that multiple calls with same data behave consistently.""" minimal = [_to_minimal_test(i) for i in sample_issues] result1 = _apply_compaction(minimal) result2 = _apply_compaction(minimal) # Should be consistent assert result1.total_count == result2.total_count assert result1.preview_count == result2.preview_count assert len(result1.preview) == len(result2.preview) def test_custom_thresholds(self, sample_issues): """Test compaction with custom threshold.""" minimal = [_to_minimal_test(i) for i in sample_issues] # With threshold=10, 25 items should compact result_low = _apply_compaction(minimal, threshold=10, preview_count=3) assert isinstance(result_low, CompactedResult) assert result_low.preview_count == 3 assert len(result_low.preview) == 3 # With threshold=50, 25 items should NOT compact result_high = _apply_compaction(minimal, threshold=50) assert isinstance(result_high, list) assert len(result_high) == 25 class TestConversionToMinimal: """Test conversion from full Issue to IssueMinimal.""" def test_issue_to_minimal_conversion(self): """Test converting Issue to IssueMinimal.""" issue = Issue( id="bd-123", title="Test Issue", description="Long description with lots of detail", design="Design notes", status="open", priority=1, issue_type="bug", assignee="alice", labels=["urgent", "backend"], dependency_count=2, dependent_count=1, created_at=datetime.now(timezone.utc), updated_at=datetime.now(timezone.utc), ) minimal = _to_minimal_test(issue) # Check minimal has only expected fields assert minimal.id == "bd-123" assert minimal.title == "Test Issue" assert minimal.status == "open" assert minimal.priority == 1 assert minimal.issue_type == "bug" assert minimal.assignee == "alice" assert minimal.labels == ["urgent", "backend"] assert minimal.dependency_count == 2 assert minimal.dependent_count == 1 # Check it doesn't have full Issue fields assert not hasattr(minimal, "description") or minimal.description is None assert not hasattr(minimal, "design") or minimal.design is None def test_minimal_is_much_smaller(self): """Verify IssueMinimal is significantly smaller than Issue (roughly 80% reduction).""" now = datetime.now(timezone.utc) issue = Issue( id="bd-123", title="Test Issue", description="A" * 1000, # Long description design="B" * 500, # Design notes acceptance_criteria="C" * 500, # Acceptance criteria notes="D" * 500, # Notes status="open", priority=1, issue_type="bug", assignee="alice", labels=["a", "b", "c"], dependency_count=5, dependent_count=3, created_at=now, updated_at=now, ) issue_size = len(str(issue)) minimal = _to_minimal_test(issue) minimal_size = len(str(minimal)) # Minimal should be significantly smaller ratio = minimal_size / issue_size assert ratio < 0.3, f"Minimal {ratio*100:.1f}% of full size (expected <30%)" class TestCompactionWithFilters: """Test compaction behavior with filtered results.""" def test_filtered_results_below_threshold(self, sample_issues): """Test filtered results that stay below threshold don't compact.""" # Take only first 8 issues filtered = sample_issues[:8] minimal = [_to_minimal_test(i) for i in filtered] result = _apply_compaction(minimal) assert isinstance(result, list) assert len(result) == 8 def test_mixed_filters_and_compaction(self): """Test that filters are applied before compaction decision.""" # Create 50 issues now = datetime.now(timezone.utc) issues = [ Issue( id=f"bd-{i:04d}", title=f"Issue {i}", description="", status="open" if i < 30 else "closed", priority=i % 5, issue_type="task", created_at=now, updated_at=now, ) for i in range(50) ] # Simulate filtering to only open issues (first 30) filtered = [i for i in issues if i.status == "open"] assert len(filtered) == 30 minimal = [_to_minimal_test(i) for i in filtered] result = _apply_compaction(minimal) # Should compact since 30 > 20 assert isinstance(result, CompactedResult) assert result.total_count == 30 assert result.preview_count == 5