refactor: split mcp auth catalog stores

This commit is contained in:
lda
2026-06-06 19:07:10 +07:00 Verified
parent 4b7088726b
commit 79a3ee4c7a
18 changed files with 303 additions and 61 deletions
+4 -3
View File
@@ -256,9 +256,10 @@ implementation state.
`server.stores.workflow`, `server.stores.auth`, `server.stores.workflow`, `server.stores.auth`,
`server.stores.source_registry`, and `server.stores.catalog_cache` `server.stores.source_registry`, and `server.stores.catalog_cache`
filesystem overrides. Missing roles still fall back to `server.store`. filesystem overrides. Missing roles still fall back to `server.store`.
MCP compatibility still uses one `FileStore` class for auth and catalog Follow-up complete: MCP compatibility auth and catalog/cache stores are now
snapshots internally; the separate catalog root is carried for the future split at the service boundary. `FileStore` remains as a compatibility
store split. wrapper, while neutral config role roots can drive `FileAuthStore` and
`FileCatalogStore` separately.
- Completed: `wf run watch` starts run progress UX with polling over existing - Completed: `wf run watch` starts run progress UX with polling over existing
`inspect_run` and optional bounded `read_run_trace`. SSE/WebSocket/MCP `inspect_run` and optional bounded `read_run_trace`. SSE/WebSocket/MCP
progress remains deferred until polling UX proves insufficient. progress remains deferred until polling UX proves insufficient.
@@ -253,7 +253,9 @@ optional role-specific overrides without breaking existing files:
``` ```
Implementation status: first filesystem-only slice implemented. Role overrides Implementation status: first filesystem-only slice implemented. Role overrides
are optional and fall back to `server.store`. are optional and fall back to `server.store`. MCP auth and catalog/cache
storage now have separate file-store adapters; `server.store` still remains
the fallback for missing roles.
Resolution rule: Resolution rule:
+3 -1
View File
@@ -58,7 +58,9 @@ workflow artifacts/deployments/runs, source registry entries, catalog cache, and
local/dev auth records. Role-specific store overrides are now supported via local/dev auth records. Role-specific store overrides are now supported via
`server.stores.*` (e.g. `server.stores.workflow`, `server.stores.auth`, `server.stores.*` (e.g. `server.stores.workflow`, `server.stores.auth`,
`server.stores.source_registry`, `server.stores.catalog_cache`); missing roles `server.stores.source_registry`, `server.stores.catalog_cache`); missing roles
continue to fall back to `server.store`. continue to fall back to `server.store`. For filesystem configs, role-specific
store overrides can split local/dev auth records and catalog cache from workflow
records.
Start a JSON-RPC server backed by MCP broker config and MCP-capable sources: Start a JSON-RPC server backed by MCP broker config and MCP-capable sources:
+7 -4
View File
@@ -15,7 +15,7 @@ from ..source_registry import (
FileSourceRegistryStore, FileSourceRegistryStore,
workflow_mcp_source_to_connection_config, workflow_mcp_source_to_connection_config,
) )
from ..storage import FileStore from ..storage import FileAuthStore, FileCatalogStore, FileStore
from .models import BrokerStoreRoots from .models import BrokerStoreRoots
from .service import WfMcpService from .service import WfMcpService
@@ -164,11 +164,14 @@ def build_service_from_config(config: BrokerConfig) -> WfMcpService:
runtime_factory = PersistentSessionFactory() runtime_factory = PersistentSessionFactory()
store_roots = config.store_roots or BrokerStoreRoots.from_default(config.store_root) store_roots = config.store_roots or BrokerStoreRoots.from_default(config.store_root)
workflow_stores = file_workflow_stores(store_roots.workflow_root) workflow_stores = file_workflow_stores(store_roots.workflow_root)
# FileStore still owns both auth files and catalog snapshots. Role roots are # Keep FileStore as the compatibility facade on WfMcpService.store while
# carried separately so a later FileStore split can move catalog_cache without a # focused services receive role-specific stores.
# config migration. auth_store = FileAuthStore(store_roots.auth_root)
catalog_store = FileCatalogStore(store_roots.catalog_cache_root)
service = WfMcpService( service = WfMcpService(
store=FileStore(store_roots.auth_root), store=FileStore(store_roots.auth_root),
auth_store=auth_store,
catalog_store=catalog_store,
artifact_store=workflow_stores.artifact_store, artifact_store=workflow_stores.artifact_store,
draft_workspace_store=workflow_stores.draft_workspace_store, draft_workspace_store=workflow_stores.draft_workspace_store,
run_store=workflow_stores.run_store, run_store=workflow_stores.run_store,
+1 -1
View File
@@ -68,7 +68,7 @@ def workflow_server_from_service(
admin = WorkflowAdminApi( admin = WorkflowAdminApi(
connections=service.connection_service, connections=service.connection_service,
events=service.events, events=service.events,
auth=McpAuthAdminProvider(store=service.store), auth=McpAuthAdminProvider(store=service.auth_store or service.store),
) )
registry_provider = SourceRegistryAdminProvider( registry_provider = SourceRegistryAdminProvider(
source_registry_store=source_registry_store, source_registry_store=source_registry_store,
+2 -2
View File
@@ -6,7 +6,7 @@ from typing import Any
from wf_api import WorkflowAdminAuthProvider from wf_api import WorkflowAdminAuthProvider
from wf_api.auth import AuthRecord as NeutralAuthRecord from wf_api.auth import AuthRecord as NeutralAuthRecord
from ...storage import Store from ...storage import AuthStore
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
@@ -17,7 +17,7 @@ class McpAuthAdminProvider(WorkflowAdminAuthProvider):
auth variants can provide richer safe display later. auth variants can provide richer safe display later.
""" """
store: Store store: AuthStore
def list_auth_records(self) -> list[dict[str, Any]]: def list_auth_records(self) -> list[dict[str, Any]]:
return [ return [
+8 -3
View File
@@ -37,7 +37,7 @@ from ...models import (
from ...runtime import ToolExecutor from ...runtime import ToolExecutor
from ...sdk import BackendAdapter from ...sdk import BackendAdapter
from ...source_registry import SourceRegistryStore from ...source_registry import SourceRegistryStore
from ...storage import Store from ...storage import AuthStore, CatalogStore, Store
from ..admin_capabilities import admin_source from ..admin_capabilities import admin_source
from ..catalog import CombinedCatalog from ..catalog import CombinedCatalog
from .builtins import builtin_sources from .builtins import builtin_sources
@@ -60,6 +60,8 @@ class WfMcpService:
""" """
store: Store store: Store
auth_store: AuthStore | None = None
catalog_store: CatalogStore | None = None
default_catalog_max_age_seconds: int = 300 default_catalog_max_age_seconds: int = 300
event_bus: EventBus = field(default_factory=EventBus) event_bus: EventBus = field(default_factory=EventBus)
include_builtin_specs: bool = True include_builtin_specs: bool = True
@@ -83,13 +85,16 @@ class WfMcpService:
""" """
self.events = BrokerEventRecorder(self.event_bus) self.events = BrokerEventRecorder(self.event_bus)
self.connection_service = ConnectionService(events=self.events) self.connection_service = ConnectionService(events=self.events)
auth_store = self.auth_store or self.store
catalog_store = self.catalog_store or self.store
self.upstream = UpstreamTransportService( self.upstream = UpstreamTransportService(
store=self.store, auth_store=auth_store,
catalog_store=catalog_store,
event_sink=self.events.record_event, event_sink=self.events.record_event,
tool_executor=self.tool_executor, tool_executor=self.tool_executor,
) )
self.source_catalog = SourceCatalogService( self.source_catalog = SourceCatalogService(
store=self.store, store=catalog_store,
connection_lookup=self.connection_service.get, connection_lookup=self.connection_service.get,
connection_list_enabled=self.connection_service.list_enabled, connection_list_enabled=self.connection_service.list_enabled,
connection_list_all=self.connection_service.list_all, connection_list_all=self.connection_service.list_all,
+2 -2
View File
@@ -30,7 +30,7 @@ from ...models import (
CatalogSnapshot, CatalogSnapshot,
) )
from ...runtime import ToolExecutor from ...runtime import ToolExecutor
from ...storage import Store from ...storage import CatalogStore
from ...workflow.wrappers import _model_from_schema from ...workflow.wrappers import _model_from_schema
from ..catalog import CombinedCatalog, snapshot_from_specs from ..catalog import CombinedCatalog, snapshot_from_specs
from .specs import get_qualified_spec, qualify_spec from .specs import get_qualified_spec, qualify_spec
@@ -51,7 +51,7 @@ class SourceCatalogService:
the broker's configured tool executor. the broker's configured tool executor.
""" """
store: Store store: CatalogStore
connection_lookup: ConnectionLookup connection_lookup: ConnectionLookup
connection_list_enabled: ConnectionList connection_list_enabled: ConnectionList
connection_list_all: ConnectionList connection_list_all: ConnectionList
@@ -28,7 +28,7 @@ from wf_mcp.models import CatalogSnapshot, ConnectionConfig
from wf_mcp.runtime import ToolExecutor from wf_mcp.runtime import ToolExecutor
from wf_mcp.sdk import BackendAdapter from wf_mcp.sdk import BackendAdapter
from wf_mcp.shared.errors import error_payload from wf_mcp.shared.errors import error_payload
from wf_mcp.storage import Store from wf_mcp.storage import AuthStore, CatalogStore
from ...auth import connection_auth_diagnostic from ...auth import connection_auth_diagnostic
from .adapters import require_adapter from .adapters import require_adapter
@@ -45,7 +45,8 @@ class UpstreamTransportService:
admin calls, discovery, generated workflow NodeSpecs, and live source checks. admin calls, discovery, generated workflow NodeSpecs, and live source checks.
""" """
store: Store auth_store: AuthStore
catalog_store: CatalogStore
event_sink: EventSink event_sink: EventSink
adapters: dict[str, BackendAdapter] = field(default_factory=dict) adapters: dict[str, BackendAdapter] = field(default_factory=dict)
tool_executor: ToolExecutor | None = None tool_executor: ToolExecutor | None = None
@@ -54,7 +55,7 @@ class UpstreamTransportService:
self.adapters[server] = adapter self.adapters[server] = adapter
def save_auth(self, record: AuthRecord) -> None: def save_auth(self, record: AuthRecord) -> None:
self.store.save_auth(record) self.auth_store.save_auth(record)
self.event_sink( self.event_sink(
make_event( make_event(
"auth_saved", "auth_saved",
@@ -64,7 +65,7 @@ class UpstreamTransportService:
) )
def load_auth(self, connection_id: str) -> AuthRecord | None: def load_auth(self, connection_id: str) -> AuthRecord | None:
return self.store.load_auth(connection_id) return self.auth_store.load_auth(connection_id)
def load_connection_auth(self, connection: ConnectionConfig) -> AuthRecord | None: def load_connection_auth(self, connection: ConnectionConfig) -> AuthRecord | None:
"""Resolve auth for a connection, preferring explicit source auth_ref. """Resolve auth for a connection, preferring explicit source auth_ref.
@@ -251,7 +252,7 @@ class UpstreamTransportService:
fetched_at_epoch_ms=int(time.time() * 1000), fetched_at_epoch_ms=int(time.time() * 1000),
max_age_seconds=max_age_seconds or default_catalog_max_age_seconds, max_age_seconds=max_age_seconds or default_catalog_max_age_seconds,
) )
self.store.save_catalog(snapshot) self.catalog_store.save_catalog(snapshot)
record_catalog_change_events(connection.id, snapshot, "catalog_refresh") record_catalog_change_events(connection.id, snapshot, "catalog_refresh")
self.event_sink( self.event_sink(
make_event( make_event(
+16 -2
View File
@@ -1,3 +1,17 @@
from .store import FileStore, Store from .store import (
AuthStore,
CatalogStore,
FileAuthStore,
FileCatalogStore,
FileStore,
Store,
)
__all__ = ["FileStore", "Store"] __all__ = [
"AuthStore",
"CatalogStore",
"FileAuthStore",
"FileCatalogStore",
"FileStore",
"Store",
]
+79 -22
View File
@@ -19,7 +19,7 @@ from ..models import (
) )
class Store: class AuthStore:
def save_auth(self, record: AuthRecord) -> None: def save_auth(self, record: AuthRecord) -> None:
raise NotImplementedError raise NotImplementedError
@@ -41,6 +41,8 @@ class Store:
def delete_auth_record(self, auth_ref: str) -> bool: def delete_auth_record(self, auth_ref: str) -> bool:
raise NotImplementedError raise NotImplementedError
class CatalogStore:
def save_catalog(self, snapshot: CatalogSnapshot) -> None: def save_catalog(self, snapshot: CatalogSnapshot) -> None:
raise NotImplementedError raise NotImplementedError
@@ -48,21 +50,20 @@ class Store:
raise NotImplementedError raise NotImplementedError
class FileStore(Store): class Store(AuthStore, CatalogStore):
"""Compatibility store combining MCP auth and catalog/cache storage."""
class FileAuthStore(AuthStore):
def __init__(self, root: Path) -> None: def __init__(self, root: Path) -> None:
self.root = root self.root = root
self.root.mkdir(parents=True, exist_ok=True) self.root.mkdir(parents=True, exist_ok=True)
self.auth_dir.mkdir(parents=True, exist_ok=True) self.auth_dir.mkdir(parents=True, exist_ok=True)
self.catalog_dir.mkdir(parents=True, exist_ok=True)
@property @property
def auth_dir(self) -> Path: def auth_dir(self) -> Path:
return self.root / "auth" return self.root / "auth"
@property
def catalog_dir(self) -> Path:
return self.root / "catalog"
def _auth_path(self, auth_ref: str) -> Path: def _auth_path(self, auth_ref: str) -> Path:
"""Map one auth ref to one file. """Map one auth ref to one file.
@@ -78,21 +79,6 @@ class FileStore(Store):
raise ValueError(f"auth ref escapes store directory: {auth_ref!r}") raise ValueError(f"auth ref escapes store directory: {auth_ref!r}")
return path return path
def _catalog_path(self, connection_id: str) -> Path:
return self._connection_path(self.catalog_dir, connection_id)
@staticmethod
def _connection_path(directory: Path, connection_id: str) -> Path:
"""Map one validated connection id to one file inside a store directory."""
parse_connection_id(connection_id)
root = directory.resolve()
path = (directory / f"{connection_id}.json").resolve()
if path.parent != root:
raise ValueError(
f"connection id escapes store directory: {connection_id!r}"
)
return path
def save_auth(self, record: AuthRecord) -> None: def save_auth(self, record: AuthRecord) -> None:
self._auth_path(record.connection_id).write_text( self._auth_path(record.connection_id).write_text(
json.dumps( json.dumps(
@@ -142,6 +128,32 @@ class FileStore(Store):
"""Delete neutral auth through the legacy MCP file shape.""" """Delete neutral auth through the legacy MCP file shape."""
return self.delete_auth(auth_ref) return self.delete_auth(auth_ref)
class FileCatalogStore(CatalogStore):
def __init__(self, root: Path) -> None:
self.root = root
self.root.mkdir(parents=True, exist_ok=True)
self.catalog_dir.mkdir(parents=True, exist_ok=True)
@property
def catalog_dir(self) -> Path:
return self.root / "catalog"
def _catalog_path(self, connection_id: str) -> Path:
return self._connection_path(self.catalog_dir, connection_id)
@staticmethod
def _connection_path(directory: Path, connection_id: str) -> Path:
"""Map one validated connection id to one file inside a store directory."""
parse_connection_id(connection_id)
root = directory.resolve()
path = (directory / f"{connection_id}.json").resolve()
if path.parent != root:
raise ValueError(
f"connection id escapes store directory: {connection_id!r}"
)
return path
def save_catalog(self, snapshot: CatalogSnapshot) -> None: def save_catalog(self, snapshot: CatalogSnapshot) -> None:
self._catalog_path(snapshot.connection_id).write_text( self._catalog_path(snapshot.connection_id).write_text(
json.dumps(dump_catalog_snapshot(snapshot), indent=2), json.dumps(dump_catalog_snapshot(snapshot), indent=2),
@@ -167,3 +179,48 @@ class FileStore(Store):
], ],
metadata=data.get("metadata", {}), metadata=data.get("metadata", {}),
) )
class FileStore(Store):
"""Compatibility file store that combines auth and catalog stores."""
def __init__(self, root: Path) -> None:
self.root = root
self.root.mkdir(parents=True, exist_ok=True)
self._auth = FileAuthStore(root)
self._catalog = FileCatalogStore(root)
@property
def auth_dir(self) -> Path:
return self._auth.auth_dir
@property
def catalog_dir(self) -> Path:
return self._catalog.catalog_dir
def save_auth(self, record: AuthRecord) -> None:
self._auth.save_auth(record)
def load_auth(self, connection_id: str) -> AuthRecord | None:
return self._auth.load_auth(connection_id)
def list_auth_refs(self) -> list[str]:
return self._auth.list_auth_refs()
def save_auth_record(self, record: NeutralAuthRecord) -> None:
self._auth.save_auth_record(record)
def load_auth_record(self, auth_ref: str) -> NeutralAuthRecord | None:
return self._auth.load_auth_record(auth_ref)
def delete_auth(self, connection_id: str) -> bool:
return self._auth.delete_auth(connection_id)
def delete_auth_record(self, auth_ref: str) -> bool:
return self._auth.delete_auth_record(auth_ref)
def save_catalog(self, snapshot: CatalogSnapshot) -> None:
self._catalog.save_catalog(snapshot)
def load_catalog(self, connection_id: str) -> CatalogSnapshot | None:
return self._catalog.load_catalog(connection_id)
+34 -2
View File
@@ -6,8 +6,8 @@ from wf_authoring import NodeSpec
from wf_core import RunStatus from wf_core import RunStatus
from wf_mcp.broker import WfMcpService from wf_mcp.broker import WfMcpService
from wf_mcp.broker.service.source_catalog import SourceCatalogService from wf_mcp.broker.service.source_catalog import SourceCatalogService
from wf_mcp.models import ConnectionConfig from wf_mcp.models import CatalogSnapshot, ConnectionConfig
from wf_mcp.storage import FileStore from wf_mcp.storage import FileCatalogStore, FileStore
from wf_platform import ( from wf_platform import (
CapabilityBuckets, CapabilityBuckets,
CapabilitySource, CapabilitySource,
@@ -500,3 +500,35 @@ def test_source_catalog_finds_local_documentation_resource_directly() -> None:
assert result is not None assert result is not None
assert result.uri == test_resource.uri assert result.uri == test_resource.uri
def test_source_catalog_uses_catalog_store_only(tmp_path) -> None:
catalog_store = FileCatalogStore(tmp_path / "catalog")
service = SourceCatalogService(
store=catalog_store,
connection_lookup=lambda connection_id: ConnectionConfig(
id=connection_id,
server="demo",
account="personal",
),
connection_list_enabled=lambda: [],
connection_list_all=lambda: [],
tool_executor_for=lambda connection: (_ for _ in ()).throw(
AssertionError("unexpected executor")
),
load_auth=lambda connection: None,
emit_event=lambda event: None,
)
snapshot = CatalogSnapshot(
connection_id="demo.personal",
fetched_at_epoch_ms=1,
max_age_seconds=300,
nodes=[],
resources=[],
prompts=[],
metadata={},
)
service.store.save_catalog(snapshot)
assert service.store.load_catalog("demo.personal") == snapshot
assert (tmp_path / "catalog" / "catalog" / "demo.personal.json").exists()
+2 -1
View File
@@ -30,7 +30,8 @@ def _make_content_access(
events = BrokerEventRecorder(EventBus()) events = BrokerEventRecorder(EventBus())
connection_service = ConnectionService(events=events) connection_service = ConnectionService(events=events)
upstream = UpstreamTransportService( upstream = UpstreamTransportService(
store=store, auth_store=store,
catalog_store=store,
event_sink=events.record_event, event_sink=events.record_event,
) )
source_catalog = SourceCatalogService( source_catalog = SourceCatalogService(
+49 -10
View File
@@ -8,8 +8,8 @@ from wf_mcp.broker.service.source_catalog import SourceCatalogService
from wf_mcp.broker.service.upstream_transport import UpstreamTransportService from wf_mcp.broker.service.upstream_transport import UpstreamTransportService
from wf_mcp.connections import ConnectionRegistry from wf_mcp.connections import ConnectionRegistry
from wf_mcp.events import McpEvent from wf_mcp.events import McpEvent
from wf_mcp.models import AuthRecord, ConnectionConfig from wf_mcp.models import AuthRecord, CatalogSnapshot, ConnectionConfig
from wf_mcp.storage import FileStore from wf_mcp.storage import FileAuthStore, FileCatalogStore, FileStore
from wf_platform import CapabilityBuckets, CapabilitySource, SourcePermissions from wf_platform import CapabilityBuckets, CapabilitySource, SourcePermissions
from ..test_support import FakeAdapter, local_temp_root from ..test_support import FakeAdapter, local_temp_root
@@ -19,7 +19,8 @@ from ..workflow_surface.conftest import echo_artifact
def _transport(root: Path) -> UpstreamTransportService: def _transport(root: Path) -> UpstreamTransportService:
events: list[McpEvent] = [] events: list[McpEvent] = []
return UpstreamTransportService( return UpstreamTransportService(
store=FileStore(root), auth_store=FileStore(root),
catalog_store=FileStore(root),
event_sink=events.append, event_sink=events.append,
) )
@@ -27,7 +28,8 @@ def _transport(root: Path) -> UpstreamTransportService:
def test_upstream_transport_registers_adapter() -> None: def test_upstream_transport_registers_adapter() -> None:
events: list[McpEvent] = [] events: list[McpEvent] = []
transport = UpstreamTransportService( transport = UpstreamTransportService(
store=FileStore(local_temp_root() / "upstream_adapter"), auth_store=FileStore(local_temp_root() / "upstream_adapter"),
catalog_store=FileStore(local_temp_root() / "upstream_adapter"),
event_sink=events.append, event_sink=events.append,
) )
adapter = FakeAdapter() adapter = FakeAdapter()
@@ -40,7 +42,8 @@ def test_upstream_transport_registers_adapter() -> None:
def test_upstream_transport_saves_and_loads_auth_with_event() -> None: def test_upstream_transport_saves_and_loads_auth_with_event() -> None:
events: list[McpEvent] = [] events: list[McpEvent] = []
transport = UpstreamTransportService( transport = UpstreamTransportService(
store=FileStore(local_temp_root() / "upstream_auth"), auth_store=FileStore(local_temp_root() / "upstream_auth"),
catalog_store=FileStore(local_temp_root() / "upstream_auth"),
event_sink=events.append, event_sink=events.append,
) )
record = AuthRecord(connection_id="demo.personal", scheme="bearer") record = AuthRecord(connection_id="demo.personal", scheme="bearer")
@@ -74,7 +77,8 @@ async def test_upstream_transport_invokes_raw_method_and_records_events() -> Non
ConnectionConfig(id="demo.personal", server="demo", account="personal") ConnectionConfig(id="demo.personal", server="demo", account="personal")
) )
transport = UpstreamTransportService( transport = UpstreamTransportService(
store=FileStore(local_temp_root() / "upstream_raw_method"), auth_store=FileStore(local_temp_root() / "upstream_raw_method"),
catalog_store=FileStore(local_temp_root() / "upstream_raw_method"),
event_sink=events.append, event_sink=events.append,
) )
transport.register_adapter("demo", FakeAdapter()) transport.register_adapter("demo", FakeAdapter())
@@ -98,7 +102,11 @@ async def test_upstream_transport_refreshes_catalog_directly() -> None:
connections = ConnectionRegistry() connections = ConnectionRegistry()
connection = ConnectionConfig(id="demo.personal", server="demo", account="personal") connection = ConnectionConfig(id="demo.personal", server="demo", account="personal")
connections.register(connection) connections.register(connection)
transport = UpstreamTransportService(store=store, event_sink=events.append) transport = UpstreamTransportService(
auth_store=store,
catalog_store=store,
event_sink=events.append,
)
transport.register_adapter("demo", FakeAdapter()) transport.register_adapter("demo", FakeAdapter())
source_catalog = SourceCatalogService( source_catalog = SourceCatalogService(
store=store, store=store,
@@ -126,7 +134,8 @@ async def test_upstream_transport_refreshes_catalog_directly() -> None:
async def test_upstream_transport_live_diagnostics_report_missing_connection() -> None: async def test_upstream_transport_live_diagnostics_report_missing_connection() -> None:
transport = UpstreamTransportService( transport = UpstreamTransportService(
store=FileStore(local_temp_root() / "upstream_live_missing"), auth_store=FileStore(local_temp_root() / "upstream_live_missing"),
catalog_store=FileStore(local_temp_root() / "upstream_live_missing"),
event_sink=lambda event: None, event_sink=lambda event: None,
) )
@@ -134,7 +143,7 @@ async def test_upstream_transport_live_diagnostics_report_missing_connection() -
raise KeyError(connection_id) raise KeyError(connection_id)
source_catalog = SourceCatalogService( source_catalog = SourceCatalogService(
store=transport.store, store=transport.catalog_store,
connection_lookup=_raise_missing_connection, connection_lookup=_raise_missing_connection,
connection_list_enabled=lambda: [], connection_list_enabled=lambda: [],
connection_list_all=lambda: [], connection_list_all=lambda: [],
@@ -260,7 +269,11 @@ async def test_upstream_transport_live_diagnostics_report_missing_auth_ref(
metadata={"auth_ref": "github.creds"}, metadata={"auth_ref": "github.creds"},
) )
connections.register(connection) connections.register(connection)
transport = UpstreamTransportService(store=store, event_sink=events.append) transport = UpstreamTransportService(
auth_store=store,
catalog_store=store,
event_sink=events.append,
)
transport.register_adapter("demo", FakeAdapter()) transport.register_adapter("demo", FakeAdapter())
source_catalog = SourceCatalogService( source_catalog = SourceCatalogService(
store=store, store=store,
@@ -296,3 +309,29 @@ async def test_upstream_transport_live_diagnostics_report_missing_auth_ref(
assert diagnostics[0].code == "auth_not_found" assert diagnostics[0].code == "auth_not_found"
assert diagnostics[0].bound_source == "github.work" assert diagnostics[0].bound_source == "github.work"
assert "github.creds" in diagnostics[0].message assert "github.creds" in diagnostics[0].message
def test_upstream_transport_uses_separate_auth_and_catalog_stores(tmp_path) -> None:
auth_store = FileAuthStore(tmp_path / "auth")
catalog_store = FileCatalogStore(tmp_path / "catalog")
events = []
transport = UpstreamTransportService(
auth_store=auth_store,
catalog_store=catalog_store,
event_sink=events.append,
)
record = AuthRecord(connection_id="demo.personal", scheme="bearer")
transport.save_auth(record)
snapshot = CatalogSnapshot(
connection_id="demo.personal",
fetched_at_epoch_ms=1,
max_age_seconds=300,
nodes=[],
resources=[],
prompts=[],
metadata={},
)
transport.catalog_store.save_catalog(snapshot)
assert (tmp_path / "auth" / "auth" / "demo.personal.json").exists()
assert (tmp_path / "catalog" / "catalog" / "demo.personal.json").exists()
+33 -1
View File
@@ -16,7 +16,7 @@ from wf_mcp.source_registry import (
McpSourceRegistryEntry, McpSourceRegistryEntry,
SourceRegistryFile, SourceRegistryFile,
) )
from wf_mcp.storage import FileStore from wf_mcp.storage import FileAuthStore, FileStore
from wf_server import WorkflowServer from wf_server import WorkflowServer
@@ -131,3 +131,35 @@ async def test_workflow_server_from_service_exposes_auth_admin(tmp_path) -> None
"payload_keys": ["token"], "payload_keys": ["token"],
} }
] ]
async def test_workflow_server_from_service_uses_focused_auth_store(tmp_path) -> None:
from wf_artifacts import (
FileDraftWorkspaceStore,
FileRunStore,
FileWorkflowArtifactStore,
)
from wf_mcp.models import AuthRecord
auth_store = FileAuthStore(tmp_path / "auth")
service = WfMcpService(
store=FileStore(tmp_path / "compat"),
auth_store=auth_store,
artifact_store=FileWorkflowArtifactStore(tmp_path / "workflow"),
draft_workspace_store=FileDraftWorkspaceStore(tmp_path / "workflow"),
run_store=FileRunStore(tmp_path / "workflow"),
)
auth_store.save_auth(AuthRecord(connection_id="drive.work", scheme="bearer"))
config = BrokerConfig(
store_root=tmp_path / "store",
connections=[],
)
server = workflow_server_from_service(
service,
config=config,
source_registry_store=FileSourceRegistryStore(tmp_path / "store"),
)
result = await server.admin.inspect_auth_record("drive.work")
assert result["id"] == "drive.work"
+35 -1
View File
@@ -5,7 +5,7 @@ import pytest
from wf_api.auth import AuthRecord as NeutralAuthRecord from wf_api.auth import AuthRecord as NeutralAuthRecord
from wf_mcp.connections import parse_connection_id from wf_mcp.connections import parse_connection_id
from wf_mcp.models import AuthRecord, CatalogSnapshot from wf_mcp.models import AuthRecord, CatalogSnapshot
from wf_mcp.storage import FileStore from wf_mcp.storage import FileAuthStore, FileCatalogStore, FileStore
from .test_support import local_temp_root from .test_support import local_temp_root
@@ -108,3 +108,37 @@ def test_file_store_accepts_neutral_auth_ref_without_connection_shape(
assert loaded.scheme == "bearer" assert loaded.scheme == "bearer"
assert loaded.payload == {"token": "secret"} assert loaded.payload == {"token": "secret"}
assert store.delete_auth_record("api_ci-1") is True assert store.delete_auth_record("api_ci-1") is True
def test_file_auth_store_uses_own_root(tmp_path) -> None:
store = FileAuthStore(tmp_path / "auth_root")
record = AuthRecord(
connection_id="drive.work",
scheme="bearer",
payload={"token": "secret"},
)
store.save_auth(record)
assert store.load_auth("drive.work") == record
assert (tmp_path / "auth_root" / "auth" / "drive.work.json").exists()
assert not (tmp_path / "auth_root" / "catalog").exists()
def test_file_catalog_store_uses_own_root(tmp_path) -> None:
store = FileCatalogStore(tmp_path / "catalog_root")
snapshot = CatalogSnapshot(
connection_id="drive.work",
fetched_at_epoch_ms=1,
max_age_seconds=300,
nodes=[],
resources=[],
prompts=[],
metadata={},
)
store.save_catalog(snapshot)
assert store.load_catalog("drive.work") == snapshot
assert (tmp_path / "catalog_root" / "catalog" / "drive.work.json").exists()
assert not (tmp_path / "catalog_root" / "auth").exists()
@@ -4,6 +4,7 @@ from pathlib import Path
from wf_config import WorkflowConfigFile from wf_config import WorkflowConfigFile
from wf_mcp.broker.config import broker_config_from_workflow_config from wf_mcp.broker.config import broker_config_from_workflow_config
from wf_mcp.models import CatalogSnapshot
def test_broker_config_from_workflow_config_converts_mcp_sources( def test_broker_config_from_workflow_config_converts_mcp_sources(
@@ -187,7 +188,25 @@ def test_build_service_from_neutral_config_uses_role_store_roots(
service = build_service_from_config(broker) service = build_service_from_config(broker)
assert service.store.root == tmp_path / "auth" assert service.store.root == tmp_path / "auth"
assert service.auth_store is not None
assert service.catalog_store is not None
assert service.auth_store.root == tmp_path / "auth"
assert service.catalog_store.root == tmp_path / "catalog"
assert service.artifact_store.root == tmp_path / "workflow" assert service.artifact_store.root == tmp_path / "workflow"
assert service.draft_workspace_store.root == tmp_path / "workflow" assert service.draft_workspace_store.root == tmp_path / "workflow"
assert service.run_store.root == tmp_path / "workflow" assert service.run_store.root == tmp_path / "workflow"
assert (tmp_path / "sources").exists() assert (tmp_path / "sources").exists()
service.source_catalog.store.save_catalog(
CatalogSnapshot(
connection_id="everything.default",
fetched_at_epoch_ms=1,
max_age_seconds=300,
nodes=[],
resources=[],
prompts=[],
metadata={},
)
)
assert (tmp_path / "catalog" / "catalog" / "everything.default.json").exists()
assert not (tmp_path / "auth" / "catalog" / "everything.default.json").exists()