From a15d04cd181787c280ae4556396c547018a5d18d Mon Sep 17 00:00:00 2001 From: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> Date: Wed, 21 Oct 2026 21:48:05 -0700 Subject: [PATCH 1/2] fix: select provider sources before configuration imports Generated with Amplifier Co-Authored-By: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> --- AGENTS.md | 11 + amplifier_app_cli/provider_loader.py | 65 ++- amplifier_app_cli/provider_manager.py | 64 +-- amplifier_app_cli/provider_sources.py | 21 +- .../test_discover_providers_clean_install.py | 163 ++++-- tests/test_provider_source_origin.py | 495 ++++++++++++++++++ 6 files changed, 734 insertions(+), 85 deletions(-) create mode 100644 tests/test_provider_source_origin.py diff --git a/AGENTS.md b/AGENTS.md index 17949868..0002f34d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -228,6 +228,17 @@ the first install and the entire file is identical after a repeat install. ## Lightweight provider configuration +Resolve the effective runtime source before configuration-time provider imports. +Installed-package checks and picker inventories must not execute provider packages: +a wheel's cached children cannot be mixed with the selected source at startup. +Activate provider dependencies before their dependents; preserve installed-only +unknown providers and explicit activated paths. Selected-source/origin failures +must not fall back to a wheel, evict imports, or trigger an install to hide the +conflict. Run `tests/test_provider_source_origin.py` and +`tests/test_provider_source_precedence.py`; verify first-run startup with a +preinstalled provider wheel in an isolated environment, without weakening Core's +origin checks. + Model discovery and login must expand an independent full configuration copy once with `lib.env_vars.expand_env_vars`, before deriving constructor arguments. Do not discard saved account settings or expand returned environment values a diff --git a/amplifier_app_cli/provider_loader.py b/amplifier_app_cli/provider_loader.py index 48268243..1138fe2c 100644 --- a/amplifier_app_cli/provider_loader.py +++ b/amplifier_app_cli/provider_loader.py @@ -24,6 +24,10 @@ logger = logging.getLogger(__name__) +class ProviderSourceError(ImportError): + """The selected provider source cannot be used without mixing origins.""" + + def _get_provider_module_name(provider_id: str) -> str: """Convert provider ID to Python module name. @@ -50,7 +54,7 @@ def _load_provider_module_from_source_path(provider_id: str, source_path: Path) module_name = _get_provider_module_name(provider_id) expected_file = (source_path / module_name / "__init__.py").resolve() if not expected_file.is_file(): - raise ImportError( + raise ProviderSourceError( f"Configured source for provider '{provider_id}' does not contain " f"expected package '{module_name}'" ) @@ -59,23 +63,63 @@ def _load_provider_module_from_source_path(provider_id: str, source_path: Path) if cached_module is not None: cached_file = getattr(cached_module, "__file__", None) if cached_file is None or Path(cached_file).resolve() != expected_file: - raise ImportError( + raise ProviderSourceError( f"Provider '{provider_id}' is already loaded from a different source" ) + # A prior failed import can leave children behind without the parent. + # Never evict them or combine them with another copy of the package. + expected_package = expected_file.parent + for name, child in tuple(sys.modules.items()): + if name.startswith(module_name + ".") and child is not None: + child_file = getattr(child, "__file__", None) + if child_file is not None and not Path(child_file).resolve().is_relative_to( + expected_package + ): + raise ProviderSourceError( + f"Provider '{provider_id}' has a submodule loaded from a different source" + ) + module = importlib.import_module(module_name) loaded_file = getattr(module, "__file__", None) if loaded_file is None or Path(loaded_file).resolve() != expected_file: - raise ImportError( + raise ProviderSourceError( f"Provider '{provider_id}' loaded from a different source than configured" ) return module +def _load_selected_provider_module( + module_id: str, sources: dict[str, str] +) -> Any: + """Resolve the runtime source before the first configuration-time import.""" + from .provider_sources import PROVIDER_DEPENDENCIES, source_from_uri + + # A provider can import another provider at module scope (Azure/OpenAI). + # Resolve that dependency first, not whichever wheel happens to be installed. + for dependency in PROVIDER_DEPENDENCIES.get(module_id, []): + if dependency in sources: + _load_selected_provider_module(dependency, sources) + + try: + source_path = source_from_uri(sources[module_id]).resolve() + except Exception as exc: + raise ProviderSourceError( + f"Could not resolve the selected source for provider '{module_id}'" + ) from exc + + path = str(source_path) + if path not in sys.path: + sys.path.insert(0, path) + importlib.invalidate_caches() + return _load_provider_module_from_source_path(module_id, source_path) + + def _load_provider_module(provider_id: str, *, source_path: Path | None = None) -> Any: """Load a provider module. - Tries entry points first, then direct import. + Uses the selected runtime source before entry points or direct import. + Installed-only providers remain discoverable when no source is configured. Args: provider_id: Provider ID (e.g., "provider-anthropic") @@ -96,6 +140,13 @@ def _load_provider_module(provider_id: str, *, source_path: Path | None = None) else f"provider-{provider_id}" ) + from .lib.settings import AppSettings + from .provider_sources import get_effective_provider_sources + + sources = get_effective_provider_sources(AppSettings()) + if module_id in sources: + return _load_selected_provider_module(module_id, sources) + # Try entry point first try: eps = importlib.metadata.entry_points(group="amplifier.modules") @@ -111,6 +162,8 @@ def _load_provider_module(provider_id: str, *, source_path: Path | None = None) module_name = _get_provider_module_name(provider_id) try: return importlib.import_module(module_name) + except ProviderSourceError: + raise except ImportError as e: raise ImportError(f"Could not load provider module '{provider_id}': {e}") from e @@ -168,6 +221,8 @@ def load_provider_class( logger.warning(f"No provider class found in module for '{provider_id}'") return None + except ProviderSourceError: + raise except ImportError as e: if source_path is not None: raise @@ -412,6 +467,8 @@ def get_provider_info( info = provider.get_info() return info.model_dump() if hasattr(info, "model_dump") else vars(info) + except ProviderSourceError: + raise except Exception as e: if source_path is not None: raise diff --git a/amplifier_app_cli/provider_manager.py b/amplifier_app_cli/provider_manager.py index 5089b25c..ff60ed3c 100644 --- a/amplifier_app_cli/provider_manager.py +++ b/amplifier_app_cli/provider_manager.py @@ -1,5 +1,6 @@ """Provider configuration management.""" +import importlib.metadata import logging import subprocess import sys @@ -9,6 +10,7 @@ from .lib.home_env import uv_target_args from .lib.settings import AppSettings from .lib.settings import ScopeType +from .provider_loader import ProviderSourceError from .provider_loader import get_provider_info from .provider_sources import _get_ordered_providers from .provider_sources import ensure_provider_installed @@ -286,33 +288,20 @@ def list_providers(self) -> list[tuple[str, str, str]]: Returns: List of (module_id, display_name, description) tuples """ - import asyncio - - from amplifier_core.loader import ModuleLoader - - providers: dict[str, tuple[str, str, str]] = {} - - # Discover installed providers via entry points - loader = ModuleLoader() - modules = asyncio.run(loader.discover()) - - for module in modules: - if module.type == "provider": - # Try to get proper display_name from provider's get_info() - info = get_provider_info(module.id) - if info and "display_name" in info: - display_name = info["display_name"] - else: - # Fallback to module's name from entry point - display_name = module.name - providers[module.id] = (module.id, display_name, module.description) - - # Always merge source-discovered providers (handles local file paths, - # user-added providers, and modules installed with --target flag) - source_providers = self._discover_providers_from_sources() - for module_id, provider_info in source_providers.items(): - if module_id not in providers: - providers[module_id] = provider_info + # Select sources before any provider import. Full module discovery can + # execute installed packages and cache their children from another root. + providers = self._discover_providers_from_sources() + selected_sources = get_effective_provider_sources(self.config) + for ep in importlib.metadata.entry_points(group="amplifier.modules"): + module_id = ep.name + if not module_id.startswith("provider-") or module_id in selected_sources: + continue + info = get_provider_info(module_id) + providers[module_id] = ( + module_id, + (info or {}).get("display_name", _get_provider_display_name(module_id)), + (info or {}).get("description", f"Provider: {module_id}"), + ) # Sort alphabetically by display name (second element of tuple) return sorted(providers.values(), key=lambda p: p[1].lower()) @@ -346,8 +335,13 @@ def _discover_providers_from_sources(self) -> dict[str, tuple[str, str, str]]: for module_id, source_uri in ordered_providers: try: # Resolve source to path (handles both git URLs and local paths) - source = source_from_uri(source_uri) - module_path = source.resolve() + try: + source = source_from_uri(source_uri) + module_path = source.resolve() + except Exception as exc: + raise ProviderSourceError( + f"Could not resolve the selected source for provider '{module_id}'" + ) from exc # Add to sys.path if not already there path_str = str(module_path) @@ -359,7 +353,7 @@ def _discover_providers_from_sources(self) -> dict[str, tuple[str, str, str]]: importlib.invalidate_caches() # Try to get provider info via direct import - info = get_provider_info(module_id) + info = get_provider_info(module_id, source_path=module_path) if info: display_name = info.get( "display_name", _get_provider_display_name(module_id) @@ -384,6 +378,10 @@ def _discover_providers_from_sources(self) -> dict[str, tuple[str, str, str]]: f"Discovered provider from source (no info): {module_id}" ) + except ProviderSourceError: + # An origin conflict is not a missing SDK. Do not overwrite an + # installed provider or fall back to a different copy to hide it. + raise except Exception as e: # For local sources, try to install with deps (same as install_known_providers) if is_local_path(source_uri): @@ -420,7 +418,7 @@ def _discover_providers_from_sources(self) -> dict[str, tuple[str, str, str]]: sys.path.insert(0, path_str) # Retry getting provider info - info = get_provider_info(module_id) + info = get_provider_info(module_id, source_path=module_path) if info: display_name = info.get( "display_name", _get_provider_display_name(module_id) @@ -449,6 +447,8 @@ def _discover_providers_from_sources(self) -> dict[str, tuple[str, str, str]]: f"Discovered local provider after install (no info): {module_id}" ) + except ProviderSourceError: + raise except Exception as install_error: # Install failed - show fallback with consistent description logger.warning( @@ -477,7 +477,7 @@ def _discover_providers_from_sources(self) -> dict[str, tuple[str, str, str]]: if installed: # Refresh Python's view of newly installed packages importlib.invalidate_caches() - info = get_provider_info(module_id) + info = get_provider_info(module_id, source_path=module_path) if info: display_name = info.get( "display_name", _get_provider_display_name(module_id) diff --git a/amplifier_app_cli/provider_sources.py b/amplifier_app_cli/provider_sources.py index 2df81702..06c3eca4 100644 --- a/amplifier_app_cli/provider_sources.py +++ b/amplifier_app_cli/provider_sources.py @@ -113,11 +113,11 @@ def get_effective_provider_sources( if config_manager: # 1. Apply source overrides for known providers (sources.modules) overrides = config_manager.get_module_sources() - for module_id in list(sources.keys()): - if module_id in overrides: - sources[module_id] = overrides[module_id] + for module_id, source in overrides.items(): + if module_id.startswith("provider-") or module_id in sources: + sources[module_id] = source logger.debug( - f"Using override source for {module_id}: {overrides[module_id]}" + f"Using override source for {module_id}: {source}" ) # 2. Add user-added provider modules from settings @@ -172,7 +172,7 @@ def get_effective_provider_sources( def is_provider_module_installed(provider_id: str) -> bool: - """Check whether a provider module is installed and importable. + """Check whether a provider package is discoverable without importing it. Args: provider_id: Provider module ID (e.g., "provider-anthropic"). A bare @@ -208,20 +208,21 @@ def is_provider_module_installed(provider_id: str) -> bool: for ep in eps: if ep.name == module_id: try: - return importlib.util.find_spec(ep.module) is not None + # find_spec on a dotted name imports its parent. That can + # cache wheel submodules before the runtime source is chosen. + return importlib.util.find_spec(ep.module.split(".", 1)[0]) is not None except (ImportError, AttributeError, ValueError): # Parent package missing/broken -> treat as not installed. return False except Exception: pass - # Fall back to direct import check when no entry point is registered. + # The conventional top-level package check is also import-free. try: provider_name = module_id.replace("provider-", "") module_name = f"amplifier_module_provider_{provider_name.replace('-', '_')}" - importlib.import_module(module_name) - return True - except ImportError: + return importlib.util.find_spec(module_name) is not None + except (ImportError, AttributeError, ValueError): return False diff --git a/tests/test_discover_providers_clean_install.py b/tests/test_discover_providers_clean_install.py index 95b76dad..52b30aee 100644 --- a/tests/test_discover_providers_clean_install.py +++ b/tests/test_discover_providers_clean_install.py @@ -7,7 +7,8 @@ still appears in the picker via _get_provider_display_name() fallback. """ -from unittest.mock import MagicMock, patch +from pathlib import Path +from unittest.mock import MagicMock, call, patch from amplifier_app_cli.provider_manager import ProviderManager @@ -24,10 +25,13 @@ def _make_manager() -> ProviderManager: return ProviderManager() -def _failing_source(error: Exception | None = None) -> MagicMock: - """Return a mock source whose resolve() raises, simulating a missing install.""" +_SOURCE_PATH = Path(__file__).parent / "resolved-provider-fixture" + + +def _resolved_source() -> MagicMock: + """Source resolution succeeds; missing SDKs fail at get_provider_info().""" mock = MagicMock() - mock.resolve.side_effect = error or ImportError("No module named 'anthropic'") + mock.resolve.return_value = _SOURCE_PATH return mock @@ -66,7 +70,7 @@ def test_provider_appears_after_successful_install(self): ), patch( "amplifier_app_cli.provider_manager.source_from_uri", - return_value=_failing_source(), + return_value=_resolved_source(), ), patch( "amplifier_app_cli.provider_manager.ensure_provider_installed", @@ -74,8 +78,11 @@ def test_provider_appears_after_successful_install(self): ) as mock_ensure, patch( "amplifier_app_cli.provider_manager.get_provider_info", - return_value=provider_info, - ), + side_effect=[ + ModuleNotFoundError("No module named 'anthropic'"), + provider_info, + ], + ) as mock_get_info, ): providers = manager._discover_providers_from_sources() @@ -87,6 +94,10 @@ def test_provider_appears_after_successful_install(self): mock_ensure.assert_called_once_with( "provider-anthropic", config_manager=manager.config ) + assert mock_get_info.call_args_list == [ + call("provider-anthropic", source_path=_SOURCE_PATH), + call("provider-anthropic", source_path=_SOURCE_PATH), + ] def test_ensure_provider_installed_called_with_correct_module_id(self): """Verify module_id passed to ensure_provider_installed is the full prefixed ID.""" @@ -110,7 +121,7 @@ def test_ensure_provider_installed_called_with_correct_module_id(self): ), patch( "amplifier_app_cli.provider_manager.source_from_uri", - return_value=_failing_source(), + return_value=_resolved_source(), ), patch( "amplifier_app_cli.provider_manager.ensure_provider_installed", @@ -118,17 +129,25 @@ def test_ensure_provider_installed_called_with_correct_module_id(self): ) as mock_ensure, patch( "amplifier_app_cli.provider_manager.get_provider_info", - return_value={ - "display_name": "OpenAI", - "description": "OpenAI provider", - }, - ), + side_effect=[ + ModuleNotFoundError("No module named 'openai'"), + { + "display_name": "OpenAI", + "description": "OpenAI provider", + }, + ], + ) as mock_get_info, ): manager._discover_providers_from_sources() call_args = mock_ensure.call_args assert call_args[0][0] == "provider-openai" assert call_args[1]["config_manager"] is manager.config + mock_ensure.assert_called_once() + assert mock_get_info.call_args_list == [ + call("provider-openai", source_path=_SOURCE_PATH), + call("provider-openai", source_path=_SOURCE_PATH), + ] # --------------------------------------------------------------------------- @@ -161,8 +180,12 @@ def test_provider_appears_with_display_name_when_install_returns_false(self): ), patch( "amplifier_app_cli.provider_manager.source_from_uri", - return_value=_failing_source(), + return_value=_resolved_source(), ), + patch( + "amplifier_app_cli.provider_manager.get_provider_info", + side_effect=ModuleNotFoundError("No module named 'openai'"), + ) as mock_get_info, patch( "amplifier_app_cli.provider_manager.ensure_provider_installed", return_value=False, @@ -176,6 +199,9 @@ def test_provider_appears_with_display_name_when_install_returns_false(self): assert module_id == "provider-openai" assert display_name == "OpenAI" mock_ensure.assert_called_once() + mock_get_info.assert_called_once_with( + "provider-openai", source_path=_SOURCE_PATH + ) def test_provider_display_names_are_correct_for_all_fallbacks(self): """When all installs fail, each provider gets its proper display name from the dict.""" @@ -211,12 +237,16 @@ def test_provider_display_names_are_correct_for_all_fallbacks(self): ), patch( "amplifier_app_cli.provider_manager.source_from_uri", - return_value=_failing_source(), + return_value=_resolved_source(), ), + patch( + "amplifier_app_cli.provider_manager.get_provider_info", + side_effect=ModuleNotFoundError("No module named 'fixture_sdk'"), + ) as mock_get_info, patch( "amplifier_app_cli.provider_manager.ensure_provider_installed", return_value=False, - ), + ) as mock_ensure, ): providers = manager._discover_providers_from_sources() @@ -226,6 +256,10 @@ def test_provider_display_names_are_correct_for_all_fallbacks(self): f"{module_id}: expected '{expected_display}', " f"got '{providers[module_id][1]}'" ) + assert mock_ensure.call_count == len(expected_names) + assert mock_get_info.call_args_list == [ + call(module_id, source_path=_SOURCE_PATH) for module_id in expected_names + ] # --------------------------------------------------------------------------- @@ -252,12 +286,16 @@ def test_all_providers_appear_when_all_installs_fail(self): ), patch( "amplifier_app_cli.provider_manager.source_from_uri", - return_value=_failing_source(), + return_value=_resolved_source(), ), + patch( + "amplifier_app_cli.provider_manager.get_provider_info", + side_effect=ModuleNotFoundError("No module named 'fixture_sdk'"), + ) as mock_get_info, patch( "amplifier_app_cli.provider_manager.ensure_provider_installed", return_value=False, - ), + ) as mock_ensure, ): providers = manager._discover_providers_from_sources() @@ -265,13 +303,24 @@ def test_all_providers_appear_when_all_installs_fail(self): assert module_id in providers, ( f"Provider {module_id} missing — should appear even when install fails" ) + assert mock_ensure.call_count == len(DEFAULT_PROVIDER_SOURCES) + assert mock_get_info.call_args_list == [ + call(module_id, source_path=_SOURCE_PATH) + for module_id in DEFAULT_PROVIDER_SOURCES + ] def test_all_providers_appear_when_all_installs_succeed(self): """Happy path: every install succeeds — every known provider appears with full info.""" manager = _make_manager() ordered = list(DEFAULT_PROVIDER_SOURCES.items()) - def fake_get_info(module_id: str) -> dict: + attempted: set[str] = set() + + def fake_get_info(module_id: str, *, source_path: Path) -> dict: + assert source_path == _SOURCE_PATH + if module_id not in attempted: + attempted.add(module_id) + raise ModuleNotFoundError("No module named 'fixture_sdk'") name = module_id.replace("provider-", "") return { "display_name": name.title(), @@ -289,22 +338,28 @@ def fake_get_info(module_id: str) -> dict: ), patch( "amplifier_app_cli.provider_manager.source_from_uri", - return_value=_failing_source(), + return_value=_resolved_source(), ), patch( "amplifier_app_cli.provider_manager.ensure_provider_installed", return_value=True, - ), + ) as mock_ensure, patch( "amplifier_app_cli.provider_manager.get_provider_info", side_effect=fake_get_info, - ), + ) as mock_get_info, ): providers = manager._discover_providers_from_sources() assert len(providers) == len(DEFAULT_PROVIDER_SOURCES) for module_id in DEFAULT_PROVIDER_SOURCES: assert module_id in providers + assert mock_ensure.call_count == len(DEFAULT_PROVIDER_SOURCES) + assert mock_get_info.call_args_list == [ + call(module_id, source_path=_SOURCE_PATH) + for module_id in DEFAULT_PROVIDER_SOURCES + for _ in range(2) + ] # --------------------------------------------------------------------------- @@ -337,25 +392,33 @@ def test_display_name_fallback_when_info_still_none_after_install(self): ), patch( "amplifier_app_cli.provider_manager.source_from_uri", - return_value=_failing_source(), + return_value=_resolved_source(), ), patch( "amplifier_app_cli.provider_manager.ensure_provider_installed", return_value=True, - ), + ) as mock_ensure, patch( "amplifier_app_cli.provider_manager.get_provider_info", - return_value=None, # Still None even after successful install - ), + side_effect=[ + ModuleNotFoundError("No module named 'google'"), + None, # Still None even after successful install + ], + ) as mock_get_info, ): providers = manager._discover_providers_from_sources() assert "provider-gemini" in providers _, display_name, _ = providers["provider-gemini"] assert display_name == "Google Gemini" + mock_ensure.assert_called_once() + assert mock_get_info.call_args_list == [ + call("provider-gemini", source_path=_SOURCE_PATH), + call("provider-gemini", source_path=_SOURCE_PATH), + ] def test_get_provider_info_called_after_ensure_installed_succeeds(self): - """get_provider_info() is called exactly once in the retry path after install.""" + """The initial SDK failure and post-install retry use the same source path.""" manager = _make_manager() with ( @@ -376,25 +439,30 @@ def test_get_provider_info_called_after_ensure_installed_succeeds(self): ), patch( "amplifier_app_cli.provider_manager.source_from_uri", - return_value=_failing_source(), + return_value=_resolved_source(), ), patch( "amplifier_app_cli.provider_manager.ensure_provider_installed", return_value=True, - ), + ) as mock_ensure, patch( "amplifier_app_cli.provider_manager.get_provider_info", - return_value={ - "display_name": "Ollama", - "description": "Ollama provider", - }, + side_effect=[ + ModuleNotFoundError("No module named 'ollama'"), + { + "display_name": "Ollama", + "description": "Ollama provider", + }, + ], ) as mock_get_info, ): providers = manager._discover_providers_from_sources() - # get_provider_info should be called once (in the retry, not the initial try - # since source.resolve() raised before reaching it) - mock_get_info.assert_called_once_with("provider-ollama") + mock_ensure.assert_called_once() + assert mock_get_info.call_args_list == [ + call("provider-ollama", source_path=_SOURCE_PATH), + call("provider-ollama", source_path=_SOURCE_PATH), + ] assert "provider-ollama" in providers @@ -421,7 +489,15 @@ def test_ensure_not_called_for_local_source(self): ), patch( "amplifier_app_cli.provider_manager.source_from_uri", - return_value=_failing_source(), + return_value=_resolved_source(), + ), + patch( + "amplifier_app_cli.provider_manager.get_provider_info", + side_effect=ModuleNotFoundError("No module named 'fixture_sdk'"), + ) as mock_get_info, + patch( + "amplifier_app_cli.provider_manager.uv_target_args", + return_value=[], ), patch( "amplifier_app_cli.provider_manager.ensure_provider_installed", @@ -429,9 +505,18 @@ def test_ensure_not_called_for_local_source(self): patch( "amplifier_app_cli.provider_manager.subprocess.run", return_value=MagicMock(returncode=1, stderr="local install failed"), - ), + ) as mock_install, ): - manager._discover_providers_from_sources() + providers = manager._discover_providers_from_sources() # The new git-source code path must NOT fire for local providers mock_ensure.assert_not_called() + mock_install.assert_called_once() + mock_get_info.assert_called_once_with( + "provider-local", source_path=_SOURCE_PATH + ) + assert providers["local"] == ( + "local", + "Local", + "Provider: local (install failed - check pyproject.toml)", + ) diff --git a/tests/test_provider_source_origin.py b/tests/test_provider_source_origin.py new file mode 100644 index 00000000..ac227b87 --- /dev/null +++ b/tests/test_provider_source_origin.py @@ -0,0 +1,495 @@ +"""Discovery and runtime must import one provider source, not mix wheel children.""" + +import json +import os +from pathlib import Path +import subprocess +import sys +import textwrap + +import pytest + + +def _package(root: Path, name: str, marker: str) -> None: + package = root / name + package.mkdir(parents=True) + (package / "_cost.py").write_text(f"ORIGIN = {marker!r}\n", encoding="utf-8") + (package / "__init__.py").write_text( + textwrap.dedent( + """ + from types import SimpleNamespace + from . import _cost + __all__ = ["FixtureProvider"] + class FixtureProvider: + def get_info(self): + return SimpleNamespace( + id="origin-fixture", display_name=_cost.ORIGIN, + description="fixture provider", config_fields=[]) + def mount(*args, **kwargs): + return None + """ + ), + encoding="utf-8", + ) + (package / "entry.py").write_text("from . import mount\n", encoding="utf-8") + + +def _run(tmp_path: Path, body: str, *, configured: bool = True) -> dict: + """Use fresh real imports; mock packages, not import machinery or guards.""" + wheel = tmp_path / "site-packages" + selected = tmp_path / "selected-source" + name = "amplifier_module_provider_origin_fixture" + _package(wheel, name, "wheel") + _package(selected, name, "selected") + dist = wheel / "origin_fixture-1.0.dist-info" + dist.mkdir() + (dist / "METADATA").write_text( + "Metadata-Version: 2.1\nName: origin-fixture\nVersion: 1.0\n", + encoding="utf-8", + ) + (dist / "entry_points.txt").write_text( + f"[amplifier.modules]\nprovider-origin-fixture = {name}.entry:mount\n", + encoding="utf-8", + ) + home = tmp_path / "home" + config = home / ".amplifier" / "settings.yaml" + config.parent.mkdir(parents=True) + config.write_text( + "config:\n providers:\n - module: provider-origin-fixture\n" + f" source: {json.dumps(str(selected))}\n" + if configured + else "{}\n", + encoding="utf-8", + ) + env = { + key: value + for key, value in os.environ.items() + if not key.startswith(("AMPLIFIER_", "PYTHON")) + and "API_KEY" not in key + and not key.endswith("_TOKEN") + } + env.update( + HOME=str(home), + USERPROFILE=str(home), + AMPLIFIER_SKIP_STARTUP_UPDATE_CHECK="1", + PYTHONDONTWRITEBYTECODE="1", + ) + script = ( + "import importlib, json, sys\n" + "from pathlib import Path\n" + f"sys.path.insert(0, {str(wheel)!r})\n" + f"name = {name!r}\n" + f"selected = Path({str(selected)!r})\n" + "def deny_network(event, args):\n" + " if event in {'socket.connect', 'socket.getaddrinfo'}:\n" + " raise PermissionError('No network in provider-origin fixture')\n" + "sys.addaudithook(deny_network)\n" + textwrap.dedent(body) + ) + result = subprocess.run( + [sys.executable, "-c", script], + cwd=tmp_path, + env=env, + capture_output=True, + text=True, + timeout=30, + ) + assert result.returncode == 0, result.stdout + result.stderr + return json.loads(result.stdout) + + +def test_installed_check_does_not_import_entry_point_parent(tmp_path): + result = _run( + tmp_path, + """ + from amplifier_app_cli.provider_sources import is_provider_module_installed + assert is_provider_module_installed("provider-origin-fixture") + assert not any(n == name or n.startswith(name + ".") for n in sys.modules) + print(json.dumps({"imported": False})) + """, + ) + assert result == {"imported": False} + + +def test_metadata_and_later_source_load_share_package_and_children(tmp_path): + result = _run( + tmp_path, + """ + from amplifier_app_cli.provider_loader import get_provider_info + from amplifier_app_cli.provider_loader import _load_provider_module_from_source_path + info = get_provider_info("origin-fixture") + assert info["display_name"] == "selected", info + module = _load_provider_module_from_source_path("origin-fixture", selected) + assert module._cost.ORIGIN == "selected" + for n, m in tuple(sys.modules.items()): + if n == name or n.startswith(name + "."): + assert Path(m.__file__).resolve().is_relative_to(selected.resolve()) + assert get_provider_info("origin-fixture") == info + print(json.dumps({"origin": module._cost.ORIGIN})) + """, + ) + assert result == {"origin": "selected"} + + +def test_unknown_provider_source_modules_is_used_before_installed_wheel(tmp_path): + result = _run( + tmp_path, + """ + config = Path.home() / ".amplifier/settings.yaml" + config.write_text("sources:\\n modules:\\n provider-origin-fixture: " + + json.dumps(str(selected)) + "\\n") + from amplifier_app_cli.provider_loader import get_provider_info + assert get_provider_info("origin-fixture")["display_name"] == "selected" + print(json.dumps({"origin": "selected"})) + """, + ) + assert result == {"origin": "selected"} + + +@pytest.mark.parametrize("preload", ["package", "child"]) +def test_wrong_preloaded_origin_is_rejected_not_evicted(tmp_path, preload): + result = _run( + tmp_path, + f""" + from amplifier_app_cli.provider_loader import get_provider_info + if {preload!r} == "package": + original = importlib.import_module(name) + else: + import importlib.util + spec = importlib.util.spec_from_file_location(name + "._cost", + Path(sys.path[0]) / name / "_cost.py") + original = importlib.util.module_from_spec(spec) + sys.modules[name + "._cost"] = original + spec.loader.exec_module(original) + before = original.__file__ + try: + get_provider_info("origin-fixture") + except ImportError: + pass + else: + raise AssertionError("Wrong origin was silently accepted or hidden") + assert original.__file__ == before + assert original in sys.modules.values() + print(json.dumps({{"rejected": True}})) + """, + ) + assert result == {"rejected": True} + + +def test_unavailable_selected_source_does_not_fall_back_to_wheel(tmp_path): + result = _run( + tmp_path, + """ + import shutil + shutil.rmtree(selected) + from amplifier_app_cli.provider_loader import get_provider_info + try: + get_provider_info("origin-fixture") + except ImportError: + pass + else: + raise AssertionError("Unavailable selected source silently used a wheel") + assert name not in sys.modules + print(json.dumps({"rejected": True})) + """, + ) + assert result == {"rejected": True} + + +def test_installed_only_unknown_provider_remains_supported(tmp_path): + result = _run( + tmp_path, + """ + from amplifier_app_cli.provider_loader import get_provider_info + assert get_provider_info("origin-fixture")["display_name"] == "wheel" + print(json.dumps({"origin": "wheel"})) + """, + configured=False, + ) + assert result == {"origin": "wheel"} + + +def test_explicit_activated_path_wins_over_saved_source(tmp_path): + result = _run( + tmp_path, + """ + from amplifier_app_cli.provider_loader import get_provider_info + explicit = Path(sys.path[0]) + info = get_provider_info("origin-fixture", source_path=explicit) + assert info["display_name"] == "wheel" + print(json.dumps({"origin": info["display_name"]})) + """, + ) + assert result == {"origin": "wheel"} + + +def test_installed_check_without_entry_point_does_not_import(tmp_path): + result = _run( + tmp_path, + """ + from unittest.mock import patch + from amplifier_app_cli.provider_sources import is_provider_module_installed + with patch("importlib.metadata.entry_points", return_value=[]): + assert is_provider_module_installed("provider-origin-fixture") + assert name not in sys.modules + print(json.dumps({"imported": False})) + """, + ) + assert result == {"imported": False} + + +def test_picker_uses_selected_source_before_entry_point_import(tmp_path): + result = _run( + tmp_path, + """ + from unittest.mock import patch + from amplifier_app_cli.provider_manager import ProviderManager + sources = {"provider-origin-fixture": str(selected)} + with patch("amplifier_app_cli.provider_manager.get_effective_provider_sources", + return_value=sources): + providers = ProviderManager().list_providers() + assert ("provider-origin-fixture", "selected", "fixture provider") in providers + assert sys.modules[name]._cost.ORIGIN == "selected" + print(json.dumps({"origin": "selected"})) + """, + ) + assert result == {"origin": "selected"} + + +def test_provider_dependency_is_selected_before_dependent_import(tmp_path): + result = _run( + tmp_path, + """ + from unittest.mock import patch + dependency = "amplifier_module_provider_origin_base" + for root, marker in [(Path(sys.path[0]), "wheel"), (selected, "selected")]: + package = root / dependency + package.mkdir() + (package / "__init__.py").write_text("ORIGIN = " + repr(marker) + "\\n") + path = selected / name / "__init__.py" + path.write_text("from " + dependency + " import ORIGIN as DEPENDENCY\\n" + + path.read_text()) + config = Path.home() / ".amplifier/settings.yaml" + config.write_text("sources:\\n modules:\\n provider-origin-fixture: " + + json.dumps(str(selected)) + + "\\n provider-origin-base: " + json.dumps(str(selected)) + "\\n") + from amplifier_app_cli.provider_loader import get_provider_info + with patch("amplifier_app_cli.provider_sources.PROVIDER_DEPENDENCIES", + {"provider-origin-fixture": ["provider-origin-base"]}): + assert get_provider_info("origin-fixture")["display_name"] == "selected" + assert sys.modules[name].DEPENDENCY == "selected" + assert Path(sys.modules[dependency].__file__).resolve().is_relative_to(selected) + print(json.dumps({"dependency_origin": "selected"})) + """, + ) + assert result == {"dependency_origin": "selected"} + + +def test_picker_dependency_resolution_failure_does_not_import_or_repair(tmp_path): + result = _run( + tmp_path, + """ + from unittest.mock import patch + from amplifier_app_cli.lib.settings import AppSettings + from amplifier_app_cli.provider_loader import ProviderSourceError, get_provider_info + from amplifier_app_cli.provider_manager import ProviderManager + from amplifier_app_cli.provider_sources import is_provider_module_installed + + dependency = "amplifier_module_provider_origin_base" + wheel = Path(sys.path[0]) + package = wheel / dependency + package.mkdir() + (package / "__init__.py").write_text("ORIGIN = 'wheel'\\n") + dist = wheel / "origin_base-1.0.dist-info" + dist.mkdir() + (dist / "METADATA").write_text( + "Metadata-Version: 2.1\\nName: origin-base\\nVersion: 1.0\\n") + (dist / "entry_points.txt").write_text( + "[amplifier.modules]\\nprovider-origin-base = " + dependency + ":mount\\n") + path = selected / name / "__init__.py" + path.write_text("from " + dependency + " import ORIGIN as DEPENDENCY\\n" + + path.read_text()) + missing = selected.parent / "missing-base-source" + assert not missing.exists() + assert path.is_file() + config = Path.home() / ".amplifier/settings.yaml" + config.write_text(json.dumps({"sources": {"modules": { + "provider-origin-base": str(missing), + "provider-origin-fixture": str(selected), + }}})) + assert is_provider_module_installed("provider-origin-base") + assert is_provider_module_installed("provider-origin-fixture") + with ( + patch("amplifier_app_cli.provider_sources.DEFAULT_PROVIDER_SOURCES", {}), + patch("amplifier_app_cli.provider_sources.PROVIDER_DEPENDENCIES", + {"provider-origin-fixture": ["provider-origin-base"]}), + patch("amplifier_app_cli.provider_manager.ensure_provider_installed") as ensure, + patch("subprocess.run") as install, + ): + try: + ProviderManager(AppSettings()).list_providers() + except ProviderSourceError as error: + assert isinstance(error, ImportError) + assert error.__cause__ is not None + assert "provider-origin-base" in str(error) + else: + raise AssertionError("Picker hid the selected dependency source failure") + # Also exercise the central loader's dependency-resolution boundary. + try: + get_provider_info("origin-fixture") + except ProviderSourceError as error: + assert error.__cause__ is not None + assert "provider-origin-base" in str(error) + else: + raise AssertionError("Loader used the installed dependency wheel") + ensure.assert_not_called() + install.assert_not_called() + assert not any(n == name or n.startswith(name + ".") + or n == dependency or n.startswith(dependency + ".") + for n in sys.modules) + print(json.dumps({"rejected": True, "imported": False, "repaired": False})) + """, + ) + assert result == {"rejected": True, "imported": False, "repaired": False} + + +@pytest.mark.parametrize("source_kind", ["local", "git"]) +def test_sdk_repair_retry_retains_custom_settings_source_path(tmp_path, source_kind): + result = _run( + tmp_path, + """ + from types import SimpleNamespace + from unittest.mock import patch + from amplifier_app_cli.lib.settings import AppSettings, SettingsPaths + from amplifier_app_cli.provider_loader import get_provider_info as real_get_info + from amplifier_app_cli.provider_manager import ProviderManager + from amplifier_app_cli.provider_sources import get_effective_provider_sources + + source_kind = SOURCE_KIND + module_id = "provider-origin-fixture" + wheel = Path(sys.path[0]) + ambient = Path.home() / ".amplifier/settings.yaml" + ambient.write_text(json.dumps({"config": {"providers": [ + {"module": module_id, "source": str(wheel)} + ]}})) + source_uri = (str(selected) if source_kind == "local" else + "git+https://example.invalid/provider-origin-fixture@selected") + custom = selected.parent / "manager-settings" + custom.mkdir() + paths = SettingsPaths(custom / "global.yaml", custom / "project.yaml", + custom / "local.yaml") + paths.global_settings.write_text(json.dumps({"config": {"providers": [ + {"module": module_id, "source": source_uri} + ]}})) + settings = AppSettings(paths) + attempts = [] + repaired = False + + def resolve_source(uri): + root = {source_uri: selected, str(wheel): wheel}[uri] + return SimpleNamespace(resolve=lambda: root) + + def get_info(provider_id, *, source_path=None): + attempts.append(source_path) + assert provider_id == module_id + if len(attempts) == 1: + raise ModuleNotFoundError("No module named 'fixture_sdk'", + name="fixture_sdk") + assert repaired + return real_get_info(provider_id, source_path=source_path) + + def install_local(command, **kwargs): + global repaired + assert command == ["uv", "pip", "install", "-e", str(selected)] + repaired = True + return SimpleNamespace(returncode=0, stderr="") + + def install_git(provider_id, *, config_manager): + global repaired + assert provider_id == module_id + assert config_manager is settings + repaired = True + return True + + with ( + patch("amplifier_app_cli.provider_sources.DEFAULT_PROVIDER_SOURCES", {}), + patch("amplifier_app_cli.provider_manager.source_from_uri", + side_effect=resolve_source), + patch("amplifier_app_cli.provider_sources.source_from_uri", + side_effect=resolve_source), + patch("amplifier_app_cli.provider_manager.get_provider_info", + side_effect=get_info), + patch("amplifier_app_cli.provider_manager.uv_target_args", return_value=[]), + patch("amplifier_app_cli.provider_manager.subprocess.run", + side_effect=install_local) as local_install, + patch("amplifier_app_cli.provider_manager.ensure_provider_installed", + side_effect=install_git) as git_install, + ): + assert get_effective_provider_sources(settings) == {module_id: source_uri} + assert get_effective_provider_sources(AppSettings()) == {module_id: str(wheel)} + providers = ProviderManager(settings)._discover_providers_from_sources() + if source_kind == "local": + local_install.assert_called_once() + git_install.assert_not_called() + else: + git_install.assert_called_once_with(module_id, config_manager=settings) + local_install.assert_not_called() + assert attempts == [selected, selected] + assert providers[module_id] == (module_id, "selected", "fixture provider") + assert sys.modules[name]._cost.ORIGIN == "selected" + assert Path(sys.modules[name].__file__).resolve().is_relative_to(selected) + print(json.dumps({"origin": "selected", "attempts": len(attempts)})) + """.replace("SOURCE_KIND", repr(source_kind)), + ) + assert result == {"origin": "selected", "attempts": 2} + + +def test_local_retry_origin_error_propagates_without_purging_imports(tmp_path): + result = _run( + tmp_path, + """ + from types import SimpleNamespace + from unittest.mock import patch + from amplifier_app_cli.provider_loader import ProviderSourceError + from amplifier_app_cli.provider_loader import get_provider_info as real_get_info + from amplifier_app_cli.provider_manager import ProviderManager + + original = importlib.import_module(name) + before = {n: (m, m.__file__) for n, m in tuple(sys.modules.items()) + if n == name or n.startswith(name + ".")} + attempts = [] + + def get_info(provider_id, *, source_path=None): + attempts.append(source_path) + if len(attempts) == 1: + raise ModuleNotFoundError("No module named 'fixture_sdk'", + name="fixture_sdk") + return real_get_info(provider_id, source_path=source_path) + + with ( + patch("amplifier_app_cli.provider_manager.get_effective_provider_sources", + return_value={"provider-origin-fixture": str(selected)}), + patch("amplifier_app_cli.provider_manager.get_provider_info", + side_effect=get_info), + patch("amplifier_app_cli.provider_manager.uv_target_args", return_value=[]), + patch("amplifier_app_cli.provider_manager.subprocess.run", + return_value=SimpleNamespace(returncode=0, stderr="")) as install, + patch("amplifier_app_cli.provider_manager.ensure_provider_installed") as ensure, + ): + try: + ProviderManager().list_providers() + except ProviderSourceError as error: + assert "different source" in str(error) + else: + raise AssertionError("Local retry hid an origin conflict as fallback metadata") + install.assert_called_once() + ensure.assert_not_called() + assert attempts == [selected, selected] + assert sys.modules[name] is original + assert original._cost.ORIGIN == "wheel" + for n, (module, filename) in before.items(): + assert sys.modules[n] is module + assert module.__file__ == filename + print(json.dumps({"rejected": True, "preserved": True})) + """, + ) + assert result == {"rejected": True, "preserved": True} From 23b67b22ee023c42bc2dcffc60d061fd72a8045d Mon Sep 17 00:00:00 2001 From: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> Date: Wed, 21 Oct 2026 22:04:25 -0700 Subject: [PATCH 2/2] test: use portable local provider fixture sources Generated with Amplifier Co-Authored-By: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> --- tests/test_provider_source_origin.py | 71 +++++++++++++++++++++------- 1 file changed, 53 insertions(+), 18 deletions(-) diff --git a/tests/test_provider_source_origin.py b/tests/test_provider_source_origin.py index ac227b87..7caeeed0 100644 --- a/tests/test_provider_source_origin.py +++ b/tests/test_provider_source_origin.py @@ -56,7 +56,7 @@ def _run(tmp_path: Path, body: str, *, configured: bool = True) -> dict: config.parent.mkdir(parents=True) config.write_text( "config:\n providers:\n - module: provider-origin-fixture\n" - f" source: {json.dumps(str(selected))}\n" + f" source: {json.dumps(f'file://{selected}')}\n" if configured else "{}\n", encoding="utf-8", @@ -80,6 +80,8 @@ def _run(tmp_path: Path, body: str, *, configured: bool = True) -> dict: f"sys.path.insert(0, {str(wheel)!r})\n" f"name = {name!r}\n" f"selected = Path({str(selected)!r})\n" + "selected_uri = f'file://{selected}'\n" + f"wheel_uri = {'file://' + str(wheel)!r}\n" "def deny_network(event, args):\n" " if event in {'socket.connect', 'socket.getaddrinfo'}:\n" " raise PermissionError('No network in provider-origin fixture')\n" @@ -110,6 +112,37 @@ def test_installed_check_does_not_import_entry_point_parent(tmp_path): assert result == {"imported": False} +def test_fixture_uses_real_local_resolver_and_retains_strict_network_guard(tmp_path): + result = _run( + tmp_path, + """ + import socket + from amplifier_app_cli.lib.bundle_loader.resolvers import FoundationFileSource + from amplifier_app_cli.provider_sources import source_from_uri + source = source_from_uri(selected_uri) + assert isinstance(source, FoundationFileSource) + assert source.resolve() == selected.resolve() + blocked = 0 + try: + socket.getaddrinfo("example.invalid", 443) + except PermissionError as error: + assert str(error) == "No network in provider-origin fixture" + blocked += 1 + for address in [("192.0.2.1", 443), ("127.0.0.1", 9)]: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as connection: + try: + connection.connect(address) + except PermissionError as error: + assert str(error) == "No network in provider-origin fixture" + blocked += 1 + assert blocked == 3 + assert name not in sys.modules + print(json.dumps({"local_source": True, "network_controls_blocked": blocked})) + """, + ) + assert result == {"local_source": True, "network_controls_blocked": 3} + + def test_metadata_and_later_source_load_share_package_and_children(tmp_path): result = _run( tmp_path, @@ -136,7 +169,7 @@ def test_unknown_provider_source_modules_is_used_before_installed_wheel(tmp_path """ config = Path.home() / ".amplifier/settings.yaml" config.write_text("sources:\\n modules:\\n provider-origin-fixture: " - + json.dumps(str(selected)) + "\\n") + + json.dumps(selected_uri) + "\\n") from amplifier_app_cli.provider_loader import get_provider_info assert get_provider_info("origin-fixture")["display_name"] == "selected" print(json.dumps({"origin": "selected"})) @@ -150,7 +183,7 @@ def test_wrong_preloaded_origin_is_rejected_not_evicted(tmp_path, preload): result = _run( tmp_path, f""" - from amplifier_app_cli.provider_loader import get_provider_info + from amplifier_app_cli.provider_loader import ProviderSourceError, get_provider_info if {preload!r} == "package": original = importlib.import_module(name) else: @@ -163,8 +196,8 @@ def test_wrong_preloaded_origin_is_rejected_not_evicted(tmp_path, preload): before = original.__file__ try: get_provider_info("origin-fixture") - except ImportError: - pass + except ProviderSourceError as error: + assert "different source" in str(error) else: raise AssertionError("Wrong origin was silently accepted or hidden") assert original.__file__ == before @@ -181,11 +214,13 @@ def test_unavailable_selected_source_does_not_fall_back_to_wheel(tmp_path): """ import shutil shutil.rmtree(selected) - from amplifier_app_cli.provider_loader import get_provider_info + from amplifier_app_cli.lib.bundle_loader.resolvers import ModuleResolutionError + from amplifier_app_cli.provider_loader import ProviderSourceError, get_provider_info try: get_provider_info("origin-fixture") - except ImportError: - pass + except ProviderSourceError as error: + assert isinstance(error.__cause__, ModuleResolutionError) + assert "Module path not found" in str(error.__cause__) else: raise AssertionError("Unavailable selected source silently used a wheel") assert name not in sys.modules @@ -243,7 +278,7 @@ def test_picker_uses_selected_source_before_entry_point_import(tmp_path): """ from unittest.mock import patch from amplifier_app_cli.provider_manager import ProviderManager - sources = {"provider-origin-fixture": str(selected)} + sources = {"provider-origin-fixture": selected_uri} with patch("amplifier_app_cli.provider_manager.get_effective_provider_sources", return_value=sources): providers = ProviderManager().list_providers() @@ -270,8 +305,8 @@ def test_provider_dependency_is_selected_before_dependent_import(tmp_path): + path.read_text()) config = Path.home() / ".amplifier/settings.yaml" config.write_text("sources:\\n modules:\\n provider-origin-fixture: " - + json.dumps(str(selected)) - + "\\n provider-origin-base: " + json.dumps(str(selected)) + "\\n") + + json.dumps(selected_uri) + + "\\n provider-origin-base: " + json.dumps(selected_uri) + "\\n") from amplifier_app_cli.provider_loader import get_provider_info with patch("amplifier_app_cli.provider_sources.PROVIDER_DEPENDENCIES", {"provider-origin-fixture": ["provider-origin-base"]}): @@ -313,8 +348,8 @@ def test_picker_dependency_resolution_failure_does_not_import_or_repair(tmp_path assert path.is_file() config = Path.home() / ".amplifier/settings.yaml" config.write_text(json.dumps({"sources": {"modules": { - "provider-origin-base": str(missing), - "provider-origin-fixture": str(selected), + "provider-origin-base": f"file://{missing}", + "provider-origin-fixture": selected_uri, }}})) assert is_provider_module_installed("provider-origin-base") assert is_provider_module_installed("provider-origin-fixture") @@ -369,9 +404,9 @@ def test_sdk_repair_retry_retains_custom_settings_source_path(tmp_path, source_k wheel = Path(sys.path[0]) ambient = Path.home() / ".amplifier/settings.yaml" ambient.write_text(json.dumps({"config": {"providers": [ - {"module": module_id, "source": str(wheel)} + {"module": module_id, "source": wheel_uri} ]}})) - source_uri = (str(selected) if source_kind == "local" else + source_uri = (selected_uri if source_kind == "local" else "git+https://example.invalid/provider-origin-fixture@selected") custom = selected.parent / "manager-settings" custom.mkdir() @@ -385,7 +420,7 @@ def test_sdk_repair_retry_retains_custom_settings_source_path(tmp_path, source_k repaired = False def resolve_source(uri): - root = {source_uri: selected, str(wheel): wheel}[uri] + root = {source_uri: selected, wheel_uri: wheel}[uri] return SimpleNamespace(resolve=lambda: root) def get_info(provider_id, *, source_path=None): @@ -425,7 +460,7 @@ def install_git(provider_id, *, config_manager): side_effect=install_git) as git_install, ): assert get_effective_provider_sources(settings) == {module_id: source_uri} - assert get_effective_provider_sources(AppSettings()) == {module_id: str(wheel)} + assert get_effective_provider_sources(AppSettings()) == {module_id: wheel_uri} providers = ProviderManager(settings)._discover_providers_from_sources() if source_kind == "local": local_install.assert_called_once() @@ -467,7 +502,7 @@ def get_info(provider_id, *, source_path=None): with ( patch("amplifier_app_cli.provider_manager.get_effective_provider_sources", - return_value={"provider-origin-fixture": str(selected)}), + return_value={"provider-origin-fixture": selected_uri}), patch("amplifier_app_cli.provider_manager.get_provider_info", side_effect=get_info), patch("amplifier_app_cli.provider_manager.uv_target_args", return_value=[]),