test_resolve_project_id_behavior.py•5.42 kB
#!/usr/bin/env python3
"""Behavioral test for resolve_project_id() 4-tier resolution chain.
This script demonstrates the expected behavior of each priority tier:
1. Explicit ID takes precedence
2. Session-based config resolution (requires Context)
3. workflow-mcp integration (requires settings)
4. Default workspace (None)
Note: This is a demonstration script showing expected behavior.
Full integration tests require running servers and async test framework.
"""
import asyncio
from typing import Any
class MockContext:
"""Mock FastMCP Context for testing."""
def __init__(self, session_id: str):
self.session_id = session_id
class MockSettings:
"""Mock Settings for testing."""
def __init__(self, workflow_mcp_url: str | None = None):
self.workflow_mcp_url = workflow_mcp_url
self.workflow_mcp_timeout = 1.0
self.workflow_mcp_cache_ttl = 60
async def demonstrate_priority_1():
"""Priority 1: Explicit ID takes precedence over all other methods."""
print("\n" + "="*80)
print("Priority 1: Explicit ID (highest priority)")
print("="*80)
print("\nBehavior:")
print(" - If explicit_id is provided, return immediately")
print(" - No other resolution methods are attempted")
print(" - Context and settings are ignored\n")
print("Test case:")
print(" explicit_id='explicit-project', ctx=MockContext(), settings=MockSettings()")
print("\nExpected result:")
print(" ✅ Returns 'explicit-project' (explicit_id takes precedence)")
print(" ✅ No session config lookup")
print(" ✅ No workflow-mcp query")
async def demonstrate_priority_2():
"""Priority 2: Session-based config resolution."""
print("\n" + "="*80)
print("Priority 2: Session-based config resolution")
print("="*80)
print("\nBehavior:")
print(" - Requires ctx (FastMCP Context) to be provided")
print(" - Gets session_id from ctx.session_id")
print(" - Looks up working_directory for that session")
print(" - Searches for .codebase-mcp/config.json (up to 20 levels)")
print(" - Checks cache (async LRU with mtime invalidation)")
print(" - Validates config and extracts project.name or project.id\n")
print("Test case:")
print(" explicit_id=None, ctx=MockContext('session-123'), settings=None")
print("\nExpected result:")
print(" ✅ Calls _resolve_project_context(ctx)")
print(" ✅ Returns project_id from config.json if found")
print(" ✅ Falls through to Priority 3 if no config found")
print(" ✅ Falls through to Priority 3 if ctx is None")
async def demonstrate_priority_3():
"""Priority 3: workflow-mcp integration."""
print("\n" + "="*80)
print("Priority 3: workflow-mcp integration")
print("="*80)
print("\nBehavior:")
print(" - Requires settings.workflow_mcp_url to be configured")
print(" - Queries external workflow-mcp server for active project")
print(" - Uses timeout protection (default 1000ms)")
print(" - Handles connection errors gracefully\n")
print("Test case:")
print(" explicit_id=None, ctx=None, settings=MockSettings('http://localhost:3000')")
print("\nExpected result:")
print(" ✅ Queries workflow-mcp server")
print(" ✅ Returns active project UUID if available")
print(" ✅ Falls through to Priority 4 if server unavailable")
print(" ✅ Falls through to Priority 4 if no active project")
async def demonstrate_priority_4():
"""Priority 4: Default workspace fallback."""
print("\n" + "="*80)
print("Priority 4: Default workspace (fallback)")
print("="*80)
print("\nBehavior:")
print(" - Final fallback when all other methods unavailable")
print(" - Returns None (default workspace)")
print(" - Maintains backward compatibility\n")
print("Test case:")
print(" explicit_id=None, ctx=None, settings=None")
print("\nExpected result:")
print(" ✅ Returns None (default workspace)")
print(" ✅ No resolution methods attempted")
async def main():
"""Demonstrate all 4 priority tiers."""
print("\n" + "="*80)
print("resolve_project_id() 4-Tier Resolution Chain Demonstration")
print("="*80)
print("\nThe function uses a priority-based fallback chain:")
print(" 1. Explicit ID (highest priority)")
print(" 2. Session-based config (.codebase-mcp/config.json)")
print(" 3. workflow-mcp integration (external server query)")
print(" 4. Default workspace (None - backward compatibility)")
await demonstrate_priority_1()
await demonstrate_priority_2()
await demonstrate_priority_3()
await demonstrate_priority_4()
print("\n" + "="*80)
print("Summary")
print("="*80)
print("\nResolution order guarantees:")
print(" ✅ Explicit ID ALWAYS takes precedence (security)")
print(" ✅ Session config checked before external queries (performance)")
print(" ✅ workflow-mcp provides shared state across sessions")
print(" ✅ Default workspace ensures backward compatibility")
print("\nError handling:")
print(" ✅ Each tier handles errors gracefully")
print(" ✅ Failures cascade to next priority tier")
print(" ✅ No exceptions propagate to callers")
print(" ✅ Default workspace always available (None)")
print()
if __name__ == "__main__":
asyncio.run(main())