test(plugins): tests/plugins/web/ — coverage for the 7-plugin migration

Adds 44 focused tests under tests/plugins/web/ covering the surface that
the PR #25182 web-provider migration introduced. Complements the
existing tests/tools/ coverage which is dispatcher-centric; this file is
plugin-centric and tests each plugin + the registry directly.

Test classes (44 tests, ~1.1s on 4 workers)
-------------------------------------------

TestBundledPluginsRegister (16 tests)
  - All seven plugins present in the registry after
    _ensure_plugins_discovered()
  - Per-plugin parametrized capability-flag assertions
    (brave-free / ddgs / searxng: search-only;
     exa / parallel / firecrawl: search + extract;
     tavily: search + extract + crawl)
  - Every plugin exposes name + display_name properties
  - Every plugin returns a picker-compatible get_setup_schema() dict

TestIsAvailable (7 tests)
  - Each premium plugin reports is_available()==False when its env var is
    absent and True once set (brave-free / searxng / tavily / exa /
    parallel)
  - firecrawl recognizes either FIRECRAWL_API_KEY or FIRECRAWL_API_URL
    as a "configured" signal
  - ddgs is the always-on fallback and must not raise from is_available()

TestRegistryResolution (4 tests)
  - Option B semantics validated end-to-end:
    1. Explicit configured provider wins even when is_available()==False
       (dispatcher surfaces typed credential errors, no silent switch)
    2. Unknown/typo name falls back to first available legacy-preference
       provider
    3. Asking for extract via a search-only backend falls back to an
       extract-capable available provider (capability-incompatible
       branch in _resolve())
    4. No config + no credentials → None (or ddgs if installed)

TestAsyncExtractDispatch (4 tests)
  - parallel + firecrawl extract() are coroutine functions (async path
    in dispatcher uses await)
  - exa + tavily extract() are sync (dispatcher wraps in
    asyncio.to_thread)

TestErrorResponseShapes (7 tests)
  - Plugins return typed error dicts (success=False + "error" key) when
    credentials are missing, never raise
  - async extract() returns list of per-URL error dicts
  - tavily crawl() returns {"results": [{"error": ...}]} on missing
    credentials

Design notes
------------
- All tests use real imports of plugin modules — no mocking of provider
  classes themselves — so they catch drift in the ABC, registry, and
  glue layer simultaneously. Per the hermes-agent-dev skill's E2E
  testing guidance.
- The autouse _isolate_env fixture clears every web-provider env var
  before each test so is_available() reflects the test's setup.
- Resolution tests use the lower-level _resolve() directly rather than
  rebuilding the HERMES_HOME config dance — same observable behavior,
  no sys.modules.pop side-effects that would break the ABC isinstance
  check inside ctx.register_web_search_provider().

tests/plugins/web/test_web_search_provider_plugins.py

@@ -0,0 +1,453 @@
+"""Plugin-side tests for the web search provider migration (PR #25182).
+
+Covers:
+
+- All seven bundled plugins (brave-free, ddgs, searxng, exa, parallel,
+  tavily, firecrawl) instantiate and self-report the expected
+  capabilities + ABC-derived defaults.
+- Each plugin's ``is_available()`` correctly reflects env-var presence.
+- The web_search_registry resolves an active provider in the documented
+  scenarios (explicit config wins ignoring availability, fallback walks
+  legacy preference filtered by availability, unknown name falls back).
+- Plugin response shapes match the legacy bit-for-bit contract.
+
+Per the dev skill: these tests use *real* imports from the plugin
+modules — no mocking of provider classes themselves — so the test
+catches drift in the ABC interface, the registry, and the plugin
+glue layer simultaneously.
+"""
+from __future__ import annotations
+
+import asyncio
+import inspect
+import os
+import sys
+from typing import Any, Dict, List
+
+import pytest
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _clear_web_env(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Strip every web-provider env var so is_available() returns False."""
+    for k in (
+        "BRAVE_SEARCH_API_KEY",
+        "SEARXNG_URL",
+        "TAVILY_API_KEY",
+        "TAVILY_BASE_URL",
+        "EXA_API_KEY",
+        "PARALLEL_API_KEY",
+        "PARALLEL_SEARCH_MODE",
+        "FIRECRAWL_API_KEY",
+        "FIRECRAWL_API_URL",
+        "FIRECRAWL_GATEWAY_URL",
+        "TOOL_GATEWAY_DOMAIN",
+        "TOOL_GATEWAY_USER_TOKEN",
+    ):
+        monkeypatch.delenv(k, raising=False)
+
+
+def _ensure_plugins_loaded() -> None:
+    """Idempotently load plugins so the registry is populated."""
+    from hermes_cli.plugins import _ensure_plugins_discovered
+
+    _ensure_plugins_discovered()
+
+
+# ---------------------------------------------------------------------------
+# Per-plugin discovery + capability flags
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture(autouse=True)
+def _isolate_env(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Each test starts with a clean web-provider env."""
+    _clear_web_env(monkeypatch)
+
+
+class TestBundledPluginsRegister:
+    """All seven bundled web plugins discover and register correctly."""
+
+    def test_all_seven_plugins_present_in_registry(self) -> None:
+        _ensure_plugins_loaded()
+        from agent.web_search_registry import list_providers
+
+        names = sorted(p.name for p in list_providers())
+        assert names == [
+            "brave-free",
+            "ddgs",
+            "exa",
+            "firecrawl",
+            "parallel",
+            "searxng",
+            "tavily",
+        ]
+
+    @pytest.mark.parametrize(
+        "plugin_name,expected_search,expected_extract,expected_crawl",
+        [
+            ("brave-free", True, False, False),
+            ("ddgs", True, False, False),
+            ("searxng", True, False, False),
+            ("exa", True, True, False),
+            ("parallel", True, True, False),
+            ("tavily", True, True, True),
+            ("firecrawl", True, True, False),
+        ],
+    )
+    def test_capability_flags_match_spec(
+        self,
+        plugin_name: str,
+        expected_search: bool,
+        expected_extract: bool,
+        expected_crawl: bool,
+    ) -> None:
+        _ensure_plugins_loaded()
+        from agent.web_search_registry import get_provider
+
+        provider = get_provider(plugin_name)
+        assert provider is not None, f"plugin {plugin_name!r} not registered"
+        assert provider.supports_search() is expected_search
+        assert provider.supports_extract() is expected_extract
+        assert provider.supports_crawl() is expected_crawl
+
+    @pytest.mark.parametrize(
+        "plugin_name",
+        ["brave-free", "ddgs", "searxng", "exa", "parallel", "tavily", "firecrawl"],
+    )
+    def test_each_plugin_has_name_and_display_name(self, plugin_name: str) -> None:
+        _ensure_plugins_loaded()
+        from agent.web_search_registry import get_provider
+
+        provider = get_provider(plugin_name)
+        assert provider is not None
+        assert provider.name == plugin_name
+        assert provider.display_name  # any non-empty string
+
+    @pytest.mark.parametrize(
+        "plugin_name",
+        ["brave-free", "ddgs", "searxng", "exa", "parallel", "tavily", "firecrawl"],
+    )
+    def test_each_plugin_has_setup_schema(self, plugin_name: str) -> None:
+        """``get_setup_schema()`` returns a dict the picker can consume."""
+        _ensure_plugins_loaded()
+        from agent.web_search_registry import get_provider
+
+        provider = get_provider(plugin_name)
+        assert provider is not None
+        schema = provider.get_setup_schema()
+        assert isinstance(schema, dict)
+        assert "name" in schema
+        assert "env_vars" in schema
+
+
+# ---------------------------------------------------------------------------
+# is_available() behavior
+# ---------------------------------------------------------------------------
+
+
+class TestIsAvailable:
+    """Each plugin's ``is_available()`` returns False without env config."""
+
+    def test_brave_free_requires_api_key(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        _ensure_plugins_loaded()
+        from agent.web_search_registry import get_provider
+
+        p = get_provider("brave-free")
+        assert p is not None
+        assert p.is_available() is False  # no BRAVE_SEARCH_API_KEY
+        monkeypatch.setenv("BRAVE_SEARCH_API_KEY", "real")
+        assert p.is_available() is True
+
+    def test_searxng_requires_url(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        _ensure_plugins_loaded()
+        from agent.web_search_registry import get_provider
+
+        p = get_provider("searxng")
+        assert p is not None
+        assert p.is_available() is False
+        monkeypatch.setenv("SEARXNG_URL", "http://localhost:8080")
+        assert p.is_available() is True
+
+    def test_tavily_requires_api_key(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        _ensure_plugins_loaded()
+        from agent.web_search_registry import get_provider
+
+        p = get_provider("tavily")
+        assert p is not None
+        assert p.is_available() is False
+        monkeypatch.setenv("TAVILY_API_KEY", "real")
+        assert p.is_available() is True
+
+    def test_exa_requires_api_key(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        _ensure_plugins_loaded()
+        from agent.web_search_registry import get_provider
+
+        p = get_provider("exa")
+        assert p is not None
+        assert p.is_available() is False
+        monkeypatch.setenv("EXA_API_KEY", "real")
+        assert p.is_available() is True
+
+    def test_parallel_requires_api_key(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        _ensure_plugins_loaded()
+        from agent.web_search_registry import get_provider
+
+        p = get_provider("parallel")
+        assert p is not None
+        assert p.is_available() is False
+        monkeypatch.setenv("PARALLEL_API_KEY", "real")
+        assert p.is_available() is True
+
+    def test_firecrawl_requires_either_key_or_url(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        _ensure_plugins_loaded()
+        from agent.web_search_registry import get_provider
+
+        p = get_provider("firecrawl")
+        assert p is not None
+        assert p.is_available() is False
+
+        # Either FIRECRAWL_API_KEY or FIRECRAWL_API_URL lights it up.
+        monkeypatch.setenv("FIRECRAWL_API_KEY", "real")
+        assert p.is_available() is True
+        monkeypatch.delenv("FIRECRAWL_API_KEY", raising=False)
+        monkeypatch.setenv("FIRECRAWL_API_URL", "http://localhost:3002")
+        assert p.is_available() is True
+
+    def test_ddgs_always_available_when_package_importable(self) -> None:
+        """DDGS is the always-on fallback — no API key required.
+
+        It may report unavailable if the ``ddgs`` package itself isn't
+        installed in the env (legitimate — the plugin's post_setup hook
+        triggers pip install on first selection). We only assert that
+        is_available() doesn't raise.
+        """
+        _ensure_plugins_loaded()
+        from agent.web_search_registry import get_provider
+
+        p = get_provider("ddgs")
+        assert p is not None
+        # Truthy or falsy, just must not raise.
+        _ = bool(p.is_available())
+
+
+# ---------------------------------------------------------------------------
+# Registry resolution semantics (Option B — conservative smart fallback)
+# ---------------------------------------------------------------------------
+
+
+class TestRegistryResolution:
+    """``_resolve()`` follows explicit-config + availability-filtered fallback."""
+
+    def test_explicit_configured_provider_returned_even_when_unavailable(
+        self,
+    ) -> None:
+        """Explicit ``web.search_backend`` wins regardless of is_available().
+
+        Without availability filtering on the explicit path, the dispatcher
+        would silently switch backends; with this check the dispatcher
+        surfaces a precise "FOO_API_KEY is not set" error instead.
+        """
+        _ensure_plugins_loaded()
+        from agent.web_search_registry import _resolve, get_provider
+
+        # No BRAVE_SEARCH_API_KEY (fixture cleared it).
+        result = _resolve("brave-free", capability="search")
+        assert result is not None
+        assert result.name == "brave-free"
+        # Confirm it's the unavailable one — dispatcher will surface
+        # a typed credential-missing error to the caller.
+        assert result.is_available() is False
+
+    def test_unknown_configured_name_falls_back_to_available_provider(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """Typo / uninstalled plugin → walk legacy preference, pick available."""
+        _ensure_plugins_loaded()
+        from agent.web_search_registry import _resolve
+
+        monkeypatch.setenv("EXA_API_KEY", "real")
+        result = _resolve("not-a-real-provider", capability="search")
+        # Either ddgs (no-key fallback) or exa (the only available
+        # premium provider) — both are valid. The point is the unknown
+        # name shouldn't return None when SOMETHING is available.
+        assert result is not None
+        assert result.is_available() is True
+
+    def test_explicit_search_only_provider_for_extract_falls_back(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """Asking for extract via a search-only backend → fall back.
+
+        ``brave-free`` is search-only (``supports_extract() is False``).
+        When the registry resolves it for an extract capability, the
+        explicit-config branch rejects it as capability-incompatible
+        and the fallback walk picks an extract-capable provider.
+        """
+        _ensure_plugins_loaded()
+        from agent.web_search_registry import _resolve
+
+        monkeypatch.setenv("EXA_API_KEY", "real")
+        result = _resolve("brave-free", capability="extract")
+        # Should land on exa (only extract-capable available provider).
+        assert result is not None
+        assert result.supports_extract() is True
+        assert result.is_available() is True
+
+    def test_no_config_no_credentials_returns_none(
+        self,
+    ) -> None:
+        """No backend configured AND no available providers → typically None.
+
+        ``ddgs`` is the no-credential fallback; if its ``ddgs`` Python
+        package is installed in the test env, ddgs will be picked.
+        Otherwise the resolver returns None. Either outcome is correct.
+        """
+        _ensure_plugins_loaded()
+        from agent.web_search_registry import _resolve
+
+        result = _resolve(None, capability="search")
+        if result is not None:
+            # The only no-credential provider is ddgs; anything else
+            # means an env var leaked in.
+            assert result.is_available() is True
+
+
+# ---------------------------------------------------------------------------
+# Sync-vs-async extract detection
+# ---------------------------------------------------------------------------
+
+
+class TestAsyncExtractDispatch:
+    """The dispatcher detects async vs sync extract methods correctly."""
+
+    def test_parallel_extract_is_async(self) -> None:
+        _ensure_plugins_loaded()
+        from agent.web_search_registry import get_provider
+
+        p = get_provider("parallel")
+        assert p is not None
+        assert inspect.iscoroutinefunction(p.extract) is True
+
+    def test_firecrawl_extract_is_async(self) -> None:
+        _ensure_plugins_loaded()
+        from agent.web_search_registry import get_provider
+
+        p = get_provider("firecrawl")
+        assert p is not None
+        assert inspect.iscoroutinefunction(p.extract) is True
+
+    def test_exa_extract_is_sync(self) -> None:
+        _ensure_plugins_loaded()
+        from agent.web_search_registry import get_provider
+
+        p = get_provider("exa")
+        assert p is not None
+        assert inspect.iscoroutinefunction(p.extract) is False
+
+    def test_tavily_extract_is_sync(self) -> None:
+        _ensure_plugins_loaded()
+        from agent.web_search_registry import get_provider
+
+        p = get_provider("tavily")
+        assert p is not None
+        assert inspect.iscoroutinefunction(p.extract) is False
+
+
+# ---------------------------------------------------------------------------
+# Error response shape (preserved bit-for-bit from legacy)
+# ---------------------------------------------------------------------------
+
+
+class TestErrorResponseShapes:
+    """When credentials are missing, plugins return typed errors, not raises."""
+
+    def test_brave_free_returns_error_dict_when_unconfigured(self) -> None:
+        _ensure_plugins_loaded()
+        from agent.web_search_registry import get_provider
+
+        p = get_provider("brave-free")
+        assert p is not None
+        result = p.search("test", limit=5)
+        assert isinstance(result, dict)
+        assert result.get("success") is False
+        assert "error" in result
+
+    def test_searxng_returns_error_dict_when_unconfigured(self) -> None:
+        _ensure_plugins_loaded()
+        from agent.web_search_registry import get_provider
+
+        p = get_provider("searxng")
+        assert p is not None
+        result = p.search("test", limit=5)
+        assert isinstance(result, dict)
+        assert result.get("success") is False
+        assert "error" in result
+
+    def test_exa_returns_error_dict_when_unconfigured(self) -> None:
+        _ensure_plugins_loaded()
+        from agent.web_search_registry import get_provider
+
+        p = get_provider("exa")
+        assert p is not None
+        result = p.search("test", limit=5)
+        assert isinstance(result, dict)
+        assert result.get("success") is False
+        assert "error" in result
+
+    def test_tavily_returns_error_dict_when_unconfigured(self) -> None:
+        _ensure_plugins_loaded()
+        from agent.web_search_registry import get_provider
+
+        p = get_provider("tavily")
+        assert p is not None
+        result = p.search("test", limit=5)
+        assert isinstance(result, dict)
+        assert result.get("success") is False
+        assert "error" in result
+
+    def test_parallel_extract_returns_per_url_errors_when_unconfigured(self) -> None:
+        _ensure_plugins_loaded()
+        from agent.web_search_registry import get_provider
+
+        p = get_provider("parallel")
+        assert p is not None
+        result = asyncio.run(p.extract(["https://example.com"]))
+        assert isinstance(result, list)
+        assert len(result) == 1
+        assert "error" in result[0]
+        assert result[0]["url"] == "https://example.com"
+
+    def test_firecrawl_extract_returns_per_url_errors_when_unconfigured(self) -> None:
+        _ensure_plugins_loaded()
+        from agent.web_search_registry import get_provider
+
+        p = get_provider("firecrawl")
+        assert p is not None
+        # firecrawl extract returns [] when the website-policy gate rejects
+        # the URL, or a per-URL error dict when the gate passes but the
+        # firecrawl client fails. Use a URL the policy allows to make sure
+        # we hit the credential-missing path.
+        result = asyncio.run(p.extract(["https://example.com"]))
+        assert isinstance(result, list)
+        if result:  # if anything came back, it should be an error entry
+            assert "error" in result[0]
+
+    def test_tavily_crawl_returns_error_dict_when_unconfigured(self) -> None:
+        _ensure_plugins_loaded()
+        from agent.web_search_registry import get_provider
+
+        p = get_provider("tavily")
+        assert p is not None
+        result = p.crawl("https://example.com")
+        assert isinstance(result, dict)
+        assert "results" in result
+        assert isinstance(result["results"], list)
+        if result["results"]:
+            assert "error" in result["results"][0]