mirror of
https://github.com/basicmachines-co/basic-memory
synced 2026-06-21 13:47:35 +00:00
phernandez
e846ae85d8
The previous backfill trigger relied on Alembic revision tracking, but alembic_version only stores the head revision — intermediate revisions (like the backfill trigger) are invisible after a multi-step upgrade or fresh DB creation. Three changes fix this: 1. Replace Alembic revision check with a simple "entities exist but embeddings are empty" check that works regardless of migration path 2. Generate embeddings during sync — after FTS indexing, batch-embed all synced entities at the end of the sync operation 3. Add background backfill at MCP startup for the upgrade path (entities already exist, no embeddings) without blocking server readiness Also adds clear startup logging for semantic embedding status so issues are easy to spot in the logs. 📋 Covers: fresh DB, upgrade from pre-embedding version, db reset, interrupted backfill Signed-off-by: Pedro Hernandez <pedro@basicmachines.co> Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> Signed-off-by: phernandez <paul@basicmachines.co>
576 lines
22 KiB
Python
576 lines
22 KiB
Python
"""Tests for project context utilities (no standard-library mock usage).
|
|
|
|
These functions are config/env driven, so we use the real ConfigManager-backed
|
|
test config file and pytest monkeypatch for environment variables.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
|
|
class _ContextState:
|
|
"""Minimal FastMCP context-state stub for unit tests."""
|
|
|
|
def __init__(self):
|
|
self._state: dict[str, object] = {}
|
|
|
|
async def get_state(self, key: str):
|
|
return self._state.get(key)
|
|
|
|
async def set_state(self, key: str, value: object, **kwargs) -> None:
|
|
self._state[key] = value
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_returns_none_when_no_default_and_no_project(config_manager, monkeypatch):
|
|
from basic_memory.mcp.project_context import resolve_project_parameter
|
|
|
|
cfg = config_manager.load_config()
|
|
cfg.default_project = None
|
|
config_manager.save_config(cfg)
|
|
|
|
monkeypatch.delenv("BASIC_MEMORY_MCP_PROJECT", raising=False)
|
|
|
|
# Prevent API fallback from returning a project via stale dependency overrides
|
|
async def _no_api_fallback():
|
|
return None
|
|
|
|
monkeypatch.setattr(
|
|
"basic_memory.mcp.project_context._resolve_default_project_from_api",
|
|
_no_api_fallback,
|
|
)
|
|
assert await resolve_project_parameter(project=None, allow_discovery=False) is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_allows_discovery_when_enabled(config_manager, monkeypatch):
|
|
from basic_memory.mcp.project_context import resolve_project_parameter
|
|
|
|
cfg = config_manager.load_config()
|
|
cfg.default_project = None
|
|
config_manager.save_config(cfg)
|
|
|
|
# Prevent API fallback from returning a project via stale dependency overrides
|
|
async def _no_api_fallback():
|
|
return None
|
|
|
|
monkeypatch.setattr(
|
|
"basic_memory.mcp.project_context._resolve_default_project_from_api",
|
|
_no_api_fallback,
|
|
)
|
|
assert await resolve_project_parameter(project=None, allow_discovery=True) is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_returns_project_when_specified(config_manager):
|
|
from basic_memory.mcp.project_context import resolve_project_parameter
|
|
|
|
cfg = config_manager.load_config()
|
|
config_manager.save_config(cfg)
|
|
|
|
assert await resolve_project_parameter(project="my-project") == "my-project"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_uses_env_var_priority(config_manager, monkeypatch):
|
|
from basic_memory.mcp.project_context import resolve_project_parameter
|
|
|
|
cfg = config_manager.load_config()
|
|
config_manager.save_config(cfg)
|
|
|
|
monkeypatch.setenv("BASIC_MEMORY_MCP_PROJECT", "env-project")
|
|
assert await resolve_project_parameter(project="explicit-project") == "env-project"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_uses_explicit_project_when_no_env(config_manager, monkeypatch):
|
|
from basic_memory.mcp.project_context import resolve_project_parameter
|
|
|
|
cfg = config_manager.load_config()
|
|
config_manager.save_config(cfg)
|
|
|
|
monkeypatch.delenv("BASIC_MEMORY_MCP_PROJECT", raising=False)
|
|
assert await resolve_project_parameter(project="explicit-project") == "explicit-project"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_uses_default_project(config_manager, config_home, monkeypatch):
|
|
from basic_memory.mcp.project_context import resolve_project_parameter
|
|
from basic_memory.config import ProjectEntry
|
|
|
|
cfg = config_manager.load_config()
|
|
(config_home / "default-project").mkdir(parents=True, exist_ok=True)
|
|
cfg.projects["default-project"] = ProjectEntry(path=str(config_home / "default-project"))
|
|
cfg.default_project = "default-project"
|
|
config_manager.save_config(cfg)
|
|
|
|
monkeypatch.delenv("BASIC_MEMORY_MCP_PROJECT", raising=False)
|
|
assert await resolve_project_parameter(project=None) == "default-project"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_returns_none_when_no_default(config_manager, monkeypatch):
|
|
from basic_memory.mcp.project_context import resolve_project_parameter
|
|
|
|
cfg = config_manager.load_config()
|
|
cfg.default_project = None
|
|
config_manager.save_config(cfg)
|
|
|
|
monkeypatch.delenv("BASIC_MEMORY_MCP_PROJECT", raising=False)
|
|
|
|
# Prevent API fallback from returning a project via stale dependency overrides
|
|
async def _no_api_fallback():
|
|
return None
|
|
|
|
monkeypatch.setattr(
|
|
"basic_memory.mcp.project_context._resolve_default_project_from_api",
|
|
_no_api_fallback,
|
|
)
|
|
assert await resolve_project_parameter(project=None) is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_env_constraint_overrides_default(config_manager, config_home, monkeypatch):
|
|
from basic_memory.mcp.project_context import resolve_project_parameter
|
|
from basic_memory.config import ProjectEntry
|
|
|
|
cfg = config_manager.load_config()
|
|
(config_home / "default-project").mkdir(parents=True, exist_ok=True)
|
|
cfg.projects["default-project"] = ProjectEntry(path=str(config_home / "default-project"))
|
|
cfg.default_project = "default-project"
|
|
config_manager.save_config(cfg)
|
|
|
|
monkeypatch.setenv("BASIC_MEMORY_MCP_PROJECT", "env-project")
|
|
assert await resolve_project_parameter(project=None) == "env-project"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_workspace_auto_selects_single_and_caches(monkeypatch):
|
|
from basic_memory.mcp.project_context import resolve_workspace_parameter
|
|
from basic_memory.schemas.cloud import WorkspaceInfo
|
|
|
|
context = _ContextState()
|
|
only_workspace = WorkspaceInfo(
|
|
tenant_id="11111111-1111-1111-1111-111111111111",
|
|
workspace_type="personal",
|
|
name="Personal",
|
|
role="owner",
|
|
)
|
|
|
|
async def fake_get_available_workspaces(context=None):
|
|
return [only_workspace]
|
|
|
|
monkeypatch.setattr(
|
|
"basic_memory.mcp.project_context.get_available_workspaces",
|
|
fake_get_available_workspaces,
|
|
)
|
|
|
|
resolved = await resolve_workspace_parameter(context=context)
|
|
assert resolved.tenant_id == only_workspace.tenant_id
|
|
assert await context.get_state("active_workspace") == only_workspace.model_dump()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_workspace_requires_user_choice_when_multiple(monkeypatch):
|
|
from basic_memory.mcp.project_context import resolve_workspace_parameter
|
|
from basic_memory.schemas.cloud import WorkspaceInfo
|
|
|
|
workspaces = [
|
|
WorkspaceInfo(
|
|
tenant_id="11111111-1111-1111-1111-111111111111",
|
|
workspace_type="personal",
|
|
name="Personal",
|
|
role="owner",
|
|
),
|
|
WorkspaceInfo(
|
|
tenant_id="22222222-2222-2222-2222-222222222222",
|
|
workspace_type="organization",
|
|
name="Team",
|
|
role="editor",
|
|
),
|
|
]
|
|
|
|
async def fake_get_available_workspaces(context=None):
|
|
return workspaces
|
|
|
|
monkeypatch.setattr(
|
|
"basic_memory.mcp.project_context.get_available_workspaces",
|
|
fake_get_available_workspaces,
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="Multiple workspaces are available"):
|
|
await resolve_workspace_parameter(context=_ContextState())
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_workspace_explicit_selection_by_tenant_id_or_name(monkeypatch):
|
|
from basic_memory.mcp.project_context import resolve_workspace_parameter
|
|
from basic_memory.schemas.cloud import WorkspaceInfo
|
|
|
|
team_workspace = WorkspaceInfo(
|
|
tenant_id="22222222-2222-2222-2222-222222222222",
|
|
workspace_type="organization",
|
|
name="Team",
|
|
role="editor",
|
|
)
|
|
workspaces = [
|
|
WorkspaceInfo(
|
|
tenant_id="11111111-1111-1111-1111-111111111111",
|
|
workspace_type="personal",
|
|
name="Personal",
|
|
role="owner",
|
|
),
|
|
team_workspace,
|
|
]
|
|
|
|
async def fake_get_available_workspaces(context=None):
|
|
return workspaces
|
|
|
|
monkeypatch.setattr(
|
|
"basic_memory.mcp.project_context.get_available_workspaces",
|
|
fake_get_available_workspaces,
|
|
)
|
|
|
|
resolved_by_id = await resolve_workspace_parameter(workspace=team_workspace.tenant_id)
|
|
assert resolved_by_id.tenant_id == team_workspace.tenant_id
|
|
|
|
resolved_by_name = await resolve_workspace_parameter(workspace="team")
|
|
assert resolved_by_name.tenant_id == team_workspace.tenant_id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_workspace_invalid_selection_lists_choices(monkeypatch):
|
|
from basic_memory.mcp.project_context import resolve_workspace_parameter
|
|
from basic_memory.schemas.cloud import WorkspaceInfo
|
|
|
|
workspaces = [
|
|
WorkspaceInfo(
|
|
tenant_id="11111111-1111-1111-1111-111111111111",
|
|
workspace_type="personal",
|
|
name="Personal",
|
|
role="owner",
|
|
)
|
|
]
|
|
|
|
async def fake_get_available_workspaces(context=None):
|
|
return workspaces
|
|
|
|
monkeypatch.setattr(
|
|
"basic_memory.mcp.project_context.get_available_workspaces",
|
|
fake_get_available_workspaces,
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="Workspace 'missing-workspace' was not found"):
|
|
await resolve_workspace_parameter(workspace="missing-workspace")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_workspace_uses_cached_workspace_without_fetch(monkeypatch):
|
|
from basic_memory.mcp.project_context import resolve_workspace_parameter
|
|
from basic_memory.schemas.cloud import WorkspaceInfo
|
|
|
|
cached_workspace = WorkspaceInfo(
|
|
tenant_id="11111111-1111-1111-1111-111111111111",
|
|
workspace_type="personal",
|
|
name="Personal",
|
|
role="owner",
|
|
)
|
|
context = _ContextState()
|
|
await context.set_state("active_workspace", cached_workspace.model_dump())
|
|
|
|
async def fail_if_called(context=None): # pragma: no cover
|
|
raise AssertionError("Workspace fetch should not run when cache is available")
|
|
|
|
monkeypatch.setattr(
|
|
"basic_memory.mcp.project_context.get_available_workspaces",
|
|
fail_if_called,
|
|
)
|
|
|
|
resolved = await resolve_workspace_parameter(context=context)
|
|
assert resolved.tenant_id == cached_workspace.tenant_id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_project_client_rejects_workspace_for_local_project(config_manager):
|
|
from basic_memory.mcp.project_context import get_project_client
|
|
from basic_memory.config import ProjectEntry
|
|
|
|
# Register "main" as a LOCAL project so get_project_mode returns LOCAL
|
|
config = config_manager.load_config()
|
|
(config_manager.config_dir.parent / "main").mkdir(parents=True, exist_ok=True)
|
|
config.projects["main"] = ProjectEntry(path=str(config_manager.config_dir.parent / "main"))
|
|
config_manager.save_config(config)
|
|
|
|
with pytest.raises(
|
|
ValueError, match="Workspace 'tenant-123' cannot be used with local project"
|
|
):
|
|
async with get_project_client(project="main", workspace="tenant-123"):
|
|
pass
|
|
|
|
|
|
class TestDetectProjectFromUrlPrefix:
|
|
"""Test detect_project_from_url_prefix for URL-based project detection."""
|
|
|
|
def test_detects_project_from_memory_url(self, config_manager):
|
|
from basic_memory.mcp.project_context import detect_project_from_url_prefix
|
|
|
|
config = config_manager.load_config()
|
|
# The config has "test-project" from the conftest fixture
|
|
result = detect_project_from_url_prefix("memory://test-project/some-note", config)
|
|
assert result == "test-project"
|
|
|
|
def test_detects_project_from_plain_path(self, config_manager):
|
|
from basic_memory.mcp.project_context import detect_project_from_url_prefix
|
|
|
|
config = config_manager.load_config()
|
|
result = detect_project_from_url_prefix("test-project/some-note", config)
|
|
assert result == "test-project"
|
|
|
|
def test_returns_none_for_unknown_prefix(self, config_manager):
|
|
from basic_memory.mcp.project_context import detect_project_from_url_prefix
|
|
|
|
config = config_manager.load_config()
|
|
result = detect_project_from_url_prefix("memory://unknown-project/note", config)
|
|
assert result is None
|
|
|
|
def test_returns_none_for_no_slash(self, config_manager):
|
|
from basic_memory.mcp.project_context import detect_project_from_url_prefix
|
|
|
|
config = config_manager.load_config()
|
|
result = detect_project_from_url_prefix("memory://single-segment", config)
|
|
assert result is None
|
|
|
|
def test_returns_none_for_wildcard_prefix(self, config_manager):
|
|
from basic_memory.mcp.project_context import detect_project_from_url_prefix
|
|
|
|
config = config_manager.load_config()
|
|
result = detect_project_from_url_prefix("memory://*/notes", config)
|
|
assert result is None
|
|
|
|
def test_matches_case_insensitive_via_permalink(self, config_manager):
|
|
from basic_memory.mcp.project_context import detect_project_from_url_prefix
|
|
from basic_memory.config import ProjectEntry
|
|
|
|
config = config_manager.load_config()
|
|
(config_manager.config_dir.parent / "My Research").mkdir(parents=True, exist_ok=True)
|
|
config.projects["My Research"] = ProjectEntry(
|
|
path=str(config_manager.config_dir.parent / "My Research")
|
|
)
|
|
config_manager.save_config(config)
|
|
|
|
result = detect_project_from_url_prefix("memory://my-research/notes", config)
|
|
assert result == "My Research"
|
|
|
|
|
|
class TestGetProjectClientRoutingOrder:
|
|
"""Test that get_project_client respects explicit routing before workspace resolution."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_local_flag_skips_workspace_resolution(self, config_manager, monkeypatch):
|
|
"""--local flag should never trigger workspace resolution, even for cloud projects."""
|
|
from basic_memory.mcp.project_context import get_project_client
|
|
from basic_memory.config import ProjectEntry, ProjectMode
|
|
|
|
config = config_manager.load_config()
|
|
config.projects["cloud-proj"] = ProjectEntry(
|
|
path=str(config_manager.config_dir.parent / "cloud-proj"),
|
|
mode=ProjectMode.CLOUD,
|
|
)
|
|
config_manager.save_config(config)
|
|
|
|
# Set explicit local routing
|
|
monkeypatch.setenv("BASIC_MEMORY_EXPLICIT_ROUTING", "true")
|
|
monkeypatch.setenv("BASIC_MEMORY_FORCE_LOCAL", "true")
|
|
monkeypatch.delenv("BASIC_MEMORY_FORCE_CLOUD", raising=False)
|
|
|
|
# Should not raise "Multiple workspaces" — it should skip workspace entirely
|
|
# It will fail at project validation (no API running), which proves routing worked
|
|
with pytest.raises(Exception) as exc_info:
|
|
async with get_project_client(project="cloud-proj"):
|
|
pass
|
|
|
|
# The error should NOT be about workspaces
|
|
assert "workspace" not in str(exc_info.value).lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cloud_project_uses_per_project_workspace_id(self, config_manager, monkeypatch):
|
|
"""Cloud project with workspace_id in config should use it without network lookup."""
|
|
from basic_memory.mcp.project_context import get_project_client
|
|
from basic_memory.config import ProjectEntry, ProjectMode
|
|
|
|
config = config_manager.load_config()
|
|
config.projects["cloud-proj"] = ProjectEntry(
|
|
path=str(config_manager.config_dir.parent / "cloud-proj"),
|
|
mode=ProjectMode.CLOUD,
|
|
workspace_id="per-project-tenant-id",
|
|
)
|
|
config.cloud_api_key = "bmc_test123"
|
|
config_manager.save_config(config)
|
|
|
|
# Patch resolve_workspace_parameter to fail if called — it should be skipped
|
|
async def fail_if_called(**kwargs): # pragma: no cover
|
|
raise AssertionError(
|
|
"resolve_workspace_parameter should not be called when workspace_id is set"
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
"basic_memory.mcp.project_context.resolve_workspace_parameter",
|
|
fail_if_called,
|
|
)
|
|
|
|
# Will fail at cloud client creation (no real cloud), but proves workspace
|
|
# resolution was skipped
|
|
with pytest.raises(Exception) as exc_info:
|
|
async with get_project_client(project="cloud-proj"):
|
|
pass
|
|
|
|
# Should not be a workspace resolution error
|
|
error_msg = str(exc_info.value).lower()
|
|
assert "resolve_workspace_parameter should not be called" not in error_msg
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cloud_project_uses_default_workspace(self, config_manager, monkeypatch):
|
|
"""Cloud project without workspace_id should fall back to default_workspace."""
|
|
from basic_memory.mcp.project_context import get_project_client
|
|
from basic_memory.config import ProjectEntry, ProjectMode
|
|
|
|
config = config_manager.load_config()
|
|
config.projects["cloud-proj"] = ProjectEntry(
|
|
path=str(config_manager.config_dir.parent / "cloud-proj"),
|
|
mode=ProjectMode.CLOUD,
|
|
)
|
|
config.default_workspace = "global-default-tenant-id"
|
|
config.cloud_api_key = "bmc_test123"
|
|
config_manager.save_config(config)
|
|
|
|
# Patch resolve_workspace_parameter to fail if called — it should be skipped
|
|
async def fail_if_called(**kwargs): # pragma: no cover
|
|
raise AssertionError(
|
|
"resolve_workspace_parameter should not be called when default_workspace is set"
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
"basic_memory.mcp.project_context.resolve_workspace_parameter",
|
|
fail_if_called,
|
|
)
|
|
|
|
# Will fail at cloud client creation, but proves workspace resolution was skipped
|
|
with pytest.raises(Exception) as exc_info:
|
|
async with get_project_client(project="cloud-proj"):
|
|
pass
|
|
|
|
error_msg = str(exc_info.value).lower()
|
|
assert "resolve_workspace_parameter should not be called" not in error_msg
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cloud_only_project_routes_to_cloud(self, config_manager, monkeypatch):
|
|
"""Project NOT in local config should route to cloud (not default to LOCAL).
|
|
|
|
Cloud-only projects aren't registered in local config. The routing logic
|
|
should detect this and use CLOUD mode, falling back to default_workspace.
|
|
"""
|
|
from basic_memory.mcp.project_context import get_project_client
|
|
|
|
config = config_manager.load_config()
|
|
# Do NOT add "cloud-only-proj" to config.projects — it's cloud-only
|
|
config.default_workspace = "global-default-tenant-id"
|
|
config.cloud_api_key = "bmc_test123"
|
|
config_manager.save_config(config)
|
|
|
|
# Patch resolve_workspace_parameter to fail if called — it should be skipped
|
|
# because default_workspace is set (priority 3)
|
|
async def fail_if_called(**kwargs): # pragma: no cover
|
|
raise AssertionError(
|
|
"resolve_workspace_parameter should not be called when default_workspace is set"
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
"basic_memory.mcp.project_context.resolve_workspace_parameter",
|
|
fail_if_called,
|
|
)
|
|
|
|
# Will fail at cloud client creation (no real cloud), but proves cloud routing
|
|
# was selected instead of local routing
|
|
with pytest.raises(Exception) as exc_info:
|
|
async with get_project_client(project="cloud-only-proj"):
|
|
pass
|
|
|
|
# The error should NOT be about workspace resolution or local routing
|
|
error_msg = str(exc_info.value).lower()
|
|
assert "resolve_workspace_parameter should not be called" not in error_msg
|
|
# Should not get a local ASGI routing error
|
|
assert "no project found" not in error_msg
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_factory_mode_skips_workspace_resolution(self, config_manager, monkeypatch):
|
|
"""When a client factory is set (in-process cloud server), skip workspace resolution.
|
|
|
|
The cloud MCP server calls set_client_factory() so that get_client() routes
|
|
requests through TenantASGITransport. In this mode, workspace and tenant context
|
|
are already resolved by the transport layer. Attempting cloud workspace resolution
|
|
would call the production control-plane API and fail with 401.
|
|
"""
|
|
from contextlib import asynccontextmanager
|
|
|
|
from basic_memory.mcp import async_client
|
|
from basic_memory.mcp.project_context import get_project_client
|
|
from basic_memory.config import ProjectEntry, ProjectMode
|
|
|
|
config = config_manager.load_config()
|
|
config.projects["cloud-proj"] = ProjectEntry(
|
|
path=str(config_manager.config_dir.parent / "cloud-proj"),
|
|
mode=ProjectMode.CLOUD,
|
|
)
|
|
config_manager.save_config(config)
|
|
|
|
# Set up a factory (simulates what cloud MCP server does)
|
|
@asynccontextmanager
|
|
async def fake_factory():
|
|
from httpx import ASGITransport, AsyncClient
|
|
from basic_memory.api.app import app as fastapi_app
|
|
|
|
async with AsyncClient(
|
|
transport=ASGITransport(app=fastapi_app),
|
|
base_url="http://test",
|
|
) as client:
|
|
yield client
|
|
|
|
original_factory = async_client._client_factory
|
|
async_client.set_client_factory(fake_factory)
|
|
|
|
# Patch workspace resolution to fail if called — factory mode should skip it
|
|
async def fail_if_called(**kwargs): # pragma: no cover
|
|
raise AssertionError("resolve_workspace_parameter must not be called in factory mode")
|
|
|
|
monkeypatch.setattr(
|
|
"basic_memory.mcp.project_context.resolve_workspace_parameter",
|
|
fail_if_called,
|
|
)
|
|
|
|
# Patch get_cloud_control_plane_client to fail if called
|
|
@asynccontextmanager
|
|
async def fail_control_plane(): # pragma: no cover
|
|
raise AssertionError(
|
|
"get_cloud_control_plane_client must not be called in factory mode"
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
"basic_memory.mcp.async_client.get_cloud_control_plane_client",
|
|
fail_control_plane,
|
|
)
|
|
|
|
try:
|
|
# Will fail at project validation (no real project in DB), but proves
|
|
# workspace resolution and control-plane calls were skipped
|
|
with pytest.raises(Exception) as exc_info:
|
|
async with get_project_client(project="cloud-proj"):
|
|
pass
|
|
|
|
error_msg = str(exc_info.value).lower()
|
|
assert "resolve_workspace_parameter must not be called" not in error_msg
|
|
assert "get_cloud_control_plane_client must not be called" not in error_msg
|
|
finally:
|
|
# Restore original factory to avoid polluting other tests
|
|
async_client._client_factory = original_factory
|