@@ -475,3 +475,77 @@ async def fail_if_called(**kwargs): # pragma: no cover
475475 assert "resolve_workspace_parameter should not be called" not in error_msg
476476 # Should not get a local ASGI routing error
477477 assert "no project found" not in error_msg
478+
479+ @pytest .mark .asyncio
480+ async def test_factory_mode_skips_workspace_resolution (self , config_manager , monkeypatch ):
481+ """When a client factory is set (in-process cloud server), skip workspace resolution.
482+
483+ The cloud MCP server calls set_client_factory() so that get_client() routes
484+ requests through TenantASGITransport. In this mode, workspace and tenant context
485+ are already resolved by the transport layer. Attempting cloud workspace resolution
486+ would call the production control-plane API and fail with 401.
487+ """
488+ from contextlib import asynccontextmanager
489+
490+ from basic_memory .mcp import async_client
491+ from basic_memory .mcp .project_context import get_project_client
492+ from basic_memory .config import ProjectEntry , ProjectMode
493+
494+ config = config_manager .load_config ()
495+ config .projects ["cloud-proj" ] = ProjectEntry (
496+ path = str (config_manager .config_dir .parent / "cloud-proj" ),
497+ mode = ProjectMode .CLOUD ,
498+ )
499+ config_manager .save_config (config )
500+
501+ # Set up a factory (simulates what cloud MCP server does)
502+ @asynccontextmanager
503+ async def fake_factory ():
504+ from httpx import ASGITransport , AsyncClient
505+ from basic_memory .api .app import app as fastapi_app
506+
507+ async with AsyncClient (
508+ transport = ASGITransport (app = fastapi_app ),
509+ base_url = "http://test" ,
510+ ) as client :
511+ yield client
512+
513+ original_factory = async_client ._client_factory
514+ async_client .set_client_factory (fake_factory )
515+
516+ # Patch workspace resolution to fail if called — factory mode should skip it
517+ async def fail_if_called (** kwargs ): # pragma: no cover
518+ raise AssertionError (
519+ "resolve_workspace_parameter must not be called in factory mode"
520+ )
521+
522+ monkeypatch .setattr (
523+ "basic_memory.mcp.project_context.resolve_workspace_parameter" ,
524+ fail_if_called ,
525+ )
526+
527+ # Patch get_cloud_control_plane_client to fail if called
528+ @asynccontextmanager
529+ async def fail_control_plane (): # pragma: no cover
530+ raise AssertionError (
531+ "get_cloud_control_plane_client must not be called in factory mode"
532+ )
533+
534+ monkeypatch .setattr (
535+ "basic_memory.mcp.async_client.get_cloud_control_plane_client" ,
536+ fail_control_plane ,
537+ )
538+
539+ try :
540+ # Will fail at project validation (no real project in DB), but proves
541+ # workspace resolution and control-plane calls were skipped
542+ with pytest .raises (Exception ) as exc_info :
543+ async with get_project_client (project = "cloud-proj" ):
544+ pass
545+
546+ error_msg = str (exc_info .value ).lower ()
547+ assert "resolve_workspace_parameter must not be called" not in error_msg
548+ assert "get_cloud_control_plane_client must not be called" not in error_msg
549+ finally :
550+ # Restore original factory to avoid polluting other tests
551+ async_client ._client_factory = original_factory
0 commit comments