hermes-agent/tests/hermes_cli/test_tools_config_post_setup.py

"""Provider setup delegates package policy to PM and external server owners."""

import os
import sys
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from threading import Thread
from types import ModuleType
from unittest.mock import patch

import pm
import pytest

from hermes_cli.tools_config_post_setup import _run_post_setup


@pytest.mark.platforms('linux')
@pytest.mark.parametrize('provider,docker,playwright', [
    ('browserbase', False, False), ('agent_browser', False, True),
    ('agent_browser', False, False), ('agent_browser', True, False),
    ('browser_use_cli', False, False),
])
def test_browser_setup_respects_provider_and_native_owner(monkeypatch, capsys, provider, docker, playwright):
    with (
        patch('tools.browser_use_cli._find_cli', return_value=['harness']) as install,
        patch('tools.browser_tool_install._running_in_docker', return_value=docker),
        patch('tools.browser_tool_install._find_agent_browser', return_value='/image/agent-browser'),
        patch('tools.browser_tool_install._chromium_installed', return_value=False) as chromium,
        patch('importlib.util.find_spec', return_value=object() if playwright else None),
        patch('shutil.which', return_value='/usr/bin/apt-get'),
        patch('pm.ensure') as ensure, patch('subprocess.run') as run,
    ):
        before = dict(os.environ)
        _run_post_setup(provider)
        assert dict(os.environ) == before
    install.assert_called_once()
    run.assert_not_called()
    if provider == 'browserbase':
        chromium.assert_not_called()
    if provider == 'browser_use_cli' or docker:
        ensure.assert_not_called()
    else:
        ensure.assert_called_once_with('agent-browser', explicit=True)
    output = capsys.readouterr().out
    if docker:
        assert 'Docker' in output
    elif provider == 'agent_browser':
        assert 'system' in output and 'npx' not in output
        if playwright:
            assert sys.executable in output and '-m playwright install-deps chromium' in output
        else:
            assert 'https://playwright.dev/python/docs/browsers#install-system-dependencies' in output


@pytest.mark.parametrize("status", [200, 503])
def test_camofox_setup_leaves_external_server_and_config_owned_by_user(monkeypatch, capsys, status):
    from hermes_cli.config import get_hermes_home
    from hermes_cli.tools_config_post_setup import _POST_SETUP_READY

    class Health(BaseHTTPRequestHandler):
        def do_GET(self):
            assert self.path == "/health"
            self.send_response(status)
            self.end_headers()
            self.wfile.write(b"{}")

        def log_message(self, format, *args):
            pass

    config_path = get_hermes_home() / "config.yaml"
    config_path.write_text("browser:\n  cloud_provider: camofox\n", encoding="utf-8")
    before = config_path.read_bytes()
    with ThreadingHTTPServer(("127.0.0.1", 0), Health) as server:
        thread = Thread(target=server.serve_forever, daemon=True)
        thread.start()
        monkeypatch.setenv("CAMOFOX_URL", f"http://127.0.0.1:{server.server_port}")
        try:
            with (
                patch("hermes_constants.find_node_executable", return_value="/external/npm"),
                patch("subprocess.run") as run,
                patch("pm.ensure") as ensure,
            ):
                _run_post_setup("camofox")
                run.assert_not_called()
                ensure.assert_not_called()
                assert _POST_SETUP_READY["camofox"]() is (status == 200)
        finally:
            server.shutdown()
            thread.join(timeout=5)

    assert config_path.read_bytes() == before
    output = capsys.readouterr().out
    assert "externally managed" in output
    assert ("reachable" if status == 200 else "not reachable") in output


@pytest.mark.parametrize("key,extra", [
    ("ddgs", "ddgs"), ("faster_whisper", "stt-whisper"),
    ("kittentts", "kittentts"), ("piper", "piper"),
])
@pytest.mark.parametrize("refused", [False, True])
def test_importable_sdk_does_not_bypass_pm_constraints(monkeypatch, capsys, key, extra, refused):
    sdk = ModuleType(key)
    sdk.__version__ = "0.0.0"
    monkeypatch.setitem(sys.modules, key, sdk)
    error = pm.InstallError("venv", "outside frozen feature set") if refused else None
    with patch("pm.sync_venv", side_effect=error) as sync:
        _run_post_setup(key)

    sync.assert_called_once_with([extra], explicit=True)
    output = capsys.readouterr().out
    if refused:
        assert "outside frozen feature set" in output
        assert "Retry with: hermes tools" in output
        assert "Restart Hermes" not in output
    else:
        assert "Restart Hermes" in output


@pytest.mark.parametrize("failure", [None, "sdk", "admission"])
def test_langfuse_setup_uses_plugin_admission_and_preserves_config_on_refusal(
    monkeypatch, tmp_path, capsys, failure,
):
    from hermes_cli.config import get_hermes_home, read_raw_config

    monkeypatch.setenv("HERMES_RUNTIME_DIR", str(tmp_path / "runtime"))
    monkeypatch.setitem(sys.modules, "langfuse", ModuleType("langfuse"))
    config_path = get_hermes_home() / "config.yaml"
    config_path.write_text(
        "plugins:\n  enabled: [other]\n  disabled: [langfuse, observability/langfuse]\n",
        encoding="utf-8",
    )
    before = config_path.read_bytes()

    def resolve_candidate(**kwargs):
        assert kwargs["explicit"] is True
        assert kwargs["plugins"].data["enabled"]
        if failure == "admission":
            raise pm.InstallError("venv", "candidate refused")
        # The real admission publisher must commit both lists, not a second UI writer.
        from pm.publication import PluginSelection
        from pm.paths import repo_root
        from hermes_cli.runtime_state import runtime_lock, finish_publication
        with runtime_lock(repo_root()):
            PluginSelection(dict(kwargs["plugins"].data)).publish(repo_root())
            finish_publication(repo_root())

    with (
        patch("pm.sync_venv", side_effect=pm.InstallError("venv", "SDK refused") if failure == "sdk" else None) as sdk,
        patch("pm.client.sync_venv", side_effect=resolve_candidate) as admission,
    ):
        _run_post_setup("langfuse")

    sdk.assert_called_once_with(["langfuse"], explicit=True)
    if failure == "sdk":
        admission.assert_not_called()
    else:
        admission.assert_called_once()
    if failure:
        assert config_path.read_bytes() == before
        output = capsys.readouterr().out
        assert "refused" in output
        if failure == "sdk":
            assert "Retry with: hermes tools" in output
    else:
        plugin_config = read_raw_config()["plugins"]
        assert set(plugin_config["enabled"]) == {"other", "observability/langfuse"}
        assert plugin_config["disabled"] == []