feat(dx): Setup Wizard + doctor command — closes #2030 (#2034)

This commit is contained in:
DanielWalnut
2026-04-10 17:43:39 +08:00
committed by GitHub
parent b107444878
commit eef0a6e2da
25 changed files with 2809 additions and 68 deletions
+1
View File
@@ -10,6 +10,7 @@ from unittest.mock import MagicMock
# Make 'app' and 'deerflow' importable from any working directory
sys.path.insert(0, str(Path(__file__).parent.parent))
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "scripts"))
# Break the circular import chain that exists in production code:
# deerflow.subagents.__init__
+342
View File
@@ -0,0 +1,342 @@
"""Unit tests for scripts/doctor.py.
Run from repo root:
cd backend && uv run pytest tests/test_doctor.py -v
"""
from __future__ import annotations
import sys
import doctor
# ---------------------------------------------------------------------------
# check_python
# ---------------------------------------------------------------------------
class TestCheckPython:
def test_current_python_passes(self):
result = doctor.check_python()
assert sys.version_info >= (3, 12)
assert result.status == "ok"
# ---------------------------------------------------------------------------
# check_config_exists
# ---------------------------------------------------------------------------
class TestCheckConfigExists:
def test_missing_config(self, tmp_path):
result = doctor.check_config_exists(tmp_path / "config.yaml")
assert result.status == "fail"
assert result.fix is not None
def test_present_config(self, tmp_path):
cfg = tmp_path / "config.yaml"
cfg.write_text("config_version: 5\n")
result = doctor.check_config_exists(cfg)
assert result.status == "ok"
# ---------------------------------------------------------------------------
# check_config_version
# ---------------------------------------------------------------------------
class TestCheckConfigVersion:
def test_up_to_date(self, tmp_path):
cfg = tmp_path / "config.yaml"
cfg.write_text("config_version: 5\n")
example = tmp_path / "config.example.yaml"
example.write_text("config_version: 5\n")
result = doctor.check_config_version(cfg, tmp_path)
assert result.status == "ok"
def test_outdated(self, tmp_path):
cfg = tmp_path / "config.yaml"
cfg.write_text("config_version: 3\n")
example = tmp_path / "config.example.yaml"
example.write_text("config_version: 5\n")
result = doctor.check_config_version(cfg, tmp_path)
assert result.status == "warn"
assert result.fix is not None
def test_missing_config_skipped(self, tmp_path):
result = doctor.check_config_version(tmp_path / "config.yaml", tmp_path)
assert result.status == "skip"
# ---------------------------------------------------------------------------
# check_config_loadable
# ---------------------------------------------------------------------------
class TestCheckConfigLoadable:
def test_loadable_config(self, tmp_path, monkeypatch):
cfg = tmp_path / "config.yaml"
cfg.write_text("config_version: 5\n")
monkeypatch.setattr(doctor, "_load_app_config", lambda _path: object())
result = doctor.check_config_loadable(cfg)
assert result.status == "ok"
def test_invalid_config(self, tmp_path, monkeypatch):
cfg = tmp_path / "config.yaml"
cfg.write_text("config_version: 5\n")
def fail(_path):
raise ValueError("bad config")
monkeypatch.setattr(doctor, "_load_app_config", fail)
result = doctor.check_config_loadable(cfg)
assert result.status == "fail"
assert "bad config" in result.detail
# ---------------------------------------------------------------------------
# check_models_configured
# ---------------------------------------------------------------------------
class TestCheckModelsConfigured:
def test_no_models(self, tmp_path):
cfg = tmp_path / "config.yaml"
cfg.write_text("config_version: 5\nmodels: []\n")
result = doctor.check_models_configured(cfg)
assert result.status == "fail"
def test_one_model(self, tmp_path):
cfg = tmp_path / "config.yaml"
cfg.write_text("config_version: 5\nmodels:\n - name: default\n use: langchain_openai:ChatOpenAI\n model: gpt-4o\n api_key: $OPENAI_API_KEY\n")
result = doctor.check_models_configured(cfg)
assert result.status == "ok"
def test_missing_config_skipped(self, tmp_path):
result = doctor.check_models_configured(tmp_path / "config.yaml")
assert result.status == "skip"
# ---------------------------------------------------------------------------
# check_llm_api_key
# ---------------------------------------------------------------------------
class TestCheckLLMApiKey:
def test_key_set(self, tmp_path, monkeypatch):
cfg = tmp_path / "config.yaml"
cfg.write_text("config_version: 5\nmodels:\n - name: default\n use: langchain_openai:ChatOpenAI\n model: gpt-4o\n api_key: $OPENAI_API_KEY\n")
monkeypatch.setenv("OPENAI_API_KEY", "sk-test")
results = doctor.check_llm_api_key(cfg)
assert any(r.status == "ok" for r in results)
assert all(r.status != "fail" for r in results)
def test_key_missing(self, tmp_path, monkeypatch):
cfg = tmp_path / "config.yaml"
cfg.write_text("config_version: 5\nmodels:\n - name: default\n use: langchain_openai:ChatOpenAI\n model: gpt-4o\n api_key: $OPENAI_API_KEY\n")
monkeypatch.delenv("OPENAI_API_KEY", raising=False)
results = doctor.check_llm_api_key(cfg)
assert any(r.status == "fail" for r in results)
failed = [r for r in results if r.status == "fail"]
assert all(r.fix is not None for r in failed)
assert any("OPENAI_API_KEY" in (r.fix or "") for r in failed)
def test_missing_config_returns_empty(self, tmp_path):
results = doctor.check_llm_api_key(tmp_path / "config.yaml")
assert results == []
# ---------------------------------------------------------------------------
# check_llm_auth
# ---------------------------------------------------------------------------
class TestCheckLLMAuth:
def test_codex_auth_file_missing_fails(self, tmp_path, monkeypatch):
cfg = tmp_path / "config.yaml"
cfg.write_text("config_version: 5\nmodels:\n - name: codex\n use: deerflow.models.openai_codex_provider:CodexChatModel\n model: gpt-5.4\n")
monkeypatch.setenv("CODEX_AUTH_PATH", str(tmp_path / "missing-auth.json"))
results = doctor.check_llm_auth(cfg)
assert any(result.status == "fail" and "Codex CLI auth available" in result.label for result in results)
def test_claude_oauth_env_passes(self, tmp_path, monkeypatch):
cfg = tmp_path / "config.yaml"
cfg.write_text("config_version: 5\nmodels:\n - name: claude\n use: deerflow.models.claude_provider:ClaudeChatModel\n model: claude-sonnet-4-6\n")
monkeypatch.setenv("CLAUDE_CODE_OAUTH_TOKEN", "token")
results = doctor.check_llm_auth(cfg)
assert any(result.status == "ok" and "Claude auth available" in result.label for result in results)
# ---------------------------------------------------------------------------
# check_web_search
# ---------------------------------------------------------------------------
class TestCheckWebSearch:
def test_ddg_always_ok(self, tmp_path):
cfg = tmp_path / "config.yaml"
cfg.write_text(
"config_version: 5\nmodels:\n - name: default\n use: langchain_openai:ChatOpenAI\n model: gpt-4o\n api_key: $OPENAI_API_KEY\ntools:\n - name: web_search\n use: deerflow.community.ddg_search.tools:web_search_tool\n"
)
result = doctor.check_web_search(cfg)
assert result.status == "ok"
assert "DuckDuckGo" in result.detail
def test_tavily_with_key_ok(self, tmp_path, monkeypatch):
monkeypatch.setenv("TAVILY_API_KEY", "tvly-test")
cfg = tmp_path / "config.yaml"
cfg.write_text("config_version: 5\ntools:\n - name: web_search\n use: deerflow.community.tavily.tools:web_search_tool\n")
result = doctor.check_web_search(cfg)
assert result.status == "ok"
def test_tavily_without_key_warns(self, tmp_path, monkeypatch):
monkeypatch.delenv("TAVILY_API_KEY", raising=False)
cfg = tmp_path / "config.yaml"
cfg.write_text("config_version: 5\ntools:\n - name: web_search\n use: deerflow.community.tavily.tools:web_search_tool\n")
result = doctor.check_web_search(cfg)
assert result.status == "warn"
assert result.fix is not None
assert "make setup" in result.fix
def test_no_search_tool_warns(self, tmp_path):
cfg = tmp_path / "config.yaml"
cfg.write_text("config_version: 5\ntools: []\n")
result = doctor.check_web_search(cfg)
assert result.status == "warn"
assert result.fix is not None
assert "make setup" in result.fix
def test_missing_config_skipped(self, tmp_path):
result = doctor.check_web_search(tmp_path / "config.yaml")
assert result.status == "skip"
def test_invalid_provider_use_fails(self, tmp_path):
cfg = tmp_path / "config.yaml"
cfg.write_text("config_version: 5\ntools:\n - name: web_search\n use: deerflow.community.not_real.tools:web_search_tool\n")
result = doctor.check_web_search(cfg)
assert result.status == "fail"
# ---------------------------------------------------------------------------
# check_web_fetch
# ---------------------------------------------------------------------------
class TestCheckWebFetch:
def test_jina_always_ok(self, tmp_path):
cfg = tmp_path / "config.yaml"
cfg.write_text("config_version: 5\ntools:\n - name: web_fetch\n use: deerflow.community.jina_ai.tools:web_fetch_tool\n")
result = doctor.check_web_fetch(cfg)
assert result.status == "ok"
assert "Jina AI" in result.detail
def test_firecrawl_without_key_warns(self, tmp_path, monkeypatch):
monkeypatch.delenv("FIRECRAWL_API_KEY", raising=False)
cfg = tmp_path / "config.yaml"
cfg.write_text("config_version: 5\ntools:\n - name: web_fetch\n use: deerflow.community.firecrawl.tools:web_fetch_tool\n")
result = doctor.check_web_fetch(cfg)
assert result.status == "warn"
assert "FIRECRAWL_API_KEY" in (result.fix or "")
def test_no_fetch_tool_warns(self, tmp_path):
cfg = tmp_path / "config.yaml"
cfg.write_text("config_version: 5\ntools: []\n")
result = doctor.check_web_fetch(cfg)
assert result.status == "warn"
assert result.fix is not None
def test_invalid_provider_use_fails(self, tmp_path):
cfg = tmp_path / "config.yaml"
cfg.write_text("config_version: 5\ntools:\n - name: web_fetch\n use: deerflow.community.not_real.tools:web_fetch_tool\n")
result = doctor.check_web_fetch(cfg)
assert result.status == "fail"
# ---------------------------------------------------------------------------
# check_env_file
# ---------------------------------------------------------------------------
class TestCheckEnvFile:
def test_missing(self, tmp_path):
result = doctor.check_env_file(tmp_path)
assert result.status == "warn"
def test_present(self, tmp_path):
(tmp_path / ".env").write_text("KEY=val\n")
result = doctor.check_env_file(tmp_path)
assert result.status == "ok"
# ---------------------------------------------------------------------------
# check_frontend_env
# ---------------------------------------------------------------------------
class TestCheckFrontendEnv:
def test_missing(self, tmp_path):
result = doctor.check_frontend_env(tmp_path)
assert result.status == "warn"
def test_present(self, tmp_path):
frontend_dir = tmp_path / "frontend"
frontend_dir.mkdir()
(frontend_dir / ".env").write_text("KEY=val\n")
result = doctor.check_frontend_env(tmp_path)
assert result.status == "ok"
# ---------------------------------------------------------------------------
# check_sandbox
# ---------------------------------------------------------------------------
class TestCheckSandbox:
def test_missing_sandbox_fails(self, tmp_path):
cfg = tmp_path / "config.yaml"
cfg.write_text("config_version: 5\n")
results = doctor.check_sandbox(cfg)
assert results[0].status == "fail"
def test_local_sandbox_with_disabled_host_bash_warns(self, tmp_path):
cfg = tmp_path / "config.yaml"
cfg.write_text("config_version: 5\nsandbox:\n use: deerflow.sandbox.local:LocalSandboxProvider\n allow_host_bash: false\ntools:\n - name: bash\n use: deerflow.sandbox.tools:bash_tool\n")
results = doctor.check_sandbox(cfg)
assert any(result.status == "warn" for result in results)
def test_container_sandbox_without_runtime_warns(self, tmp_path, monkeypatch):
cfg = tmp_path / "config.yaml"
cfg.write_text("config_version: 5\nsandbox:\n use: deerflow.community.aio_sandbox:AioSandboxProvider\ntools: []\n")
monkeypatch.setattr(doctor.shutil, "which", lambda _name: None)
results = doctor.check_sandbox(cfg)
assert any(result.label == "container runtime available" and result.status == "warn" for result in results)
# ---------------------------------------------------------------------------
# main() exit code
# ---------------------------------------------------------------------------
class TestMainExitCode:
def test_returns_int(self, tmp_path, monkeypatch, capsys):
"""main() should return 0 or 1 without raising."""
repo_root = tmp_path / "repo"
scripts_dir = repo_root / "scripts"
scripts_dir.mkdir(parents=True)
fake_doctor = scripts_dir / "doctor.py"
fake_doctor.write_text("# test-only shim for __file__ resolution\n")
monkeypatch.chdir(repo_root)
monkeypatch.setattr(doctor, "__file__", str(fake_doctor))
monkeypatch.delenv("OPENAI_API_KEY", raising=False)
monkeypatch.delenv("TAVILY_API_KEY", raising=False)
exit_code = doctor.main()
captured = capsys.readouterr()
output = captured.out + captured.err
assert exit_code in (0, 1)
assert output
assert "config.yaml" in output
assert ".env" in output
+66
View File
@@ -0,0 +1,66 @@
"""Unit tests for the Firecrawl community tools."""
import json
from unittest.mock import MagicMock, patch
class TestWebSearchTool:
@patch("deerflow.community.firecrawl.tools.FirecrawlApp")
@patch("deerflow.community.firecrawl.tools.get_app_config")
def test_search_uses_web_search_config(self, mock_get_app_config, mock_firecrawl_cls):
search_config = MagicMock()
search_config.model_extra = {"api_key": "firecrawl-search-key", "max_results": 7}
mock_get_app_config.return_value.get_tool_config.return_value = search_config
mock_result = MagicMock()
mock_result.web = [
MagicMock(title="Result", url="https://example.com", description="Snippet"),
]
mock_firecrawl_cls.return_value.search.return_value = mock_result
from deerflow.community.firecrawl.tools import web_search_tool
result = web_search_tool.invoke({"query": "test query"})
assert json.loads(result) == [
{
"title": "Result",
"url": "https://example.com",
"snippet": "Snippet",
}
]
mock_get_app_config.return_value.get_tool_config.assert_called_with("web_search")
mock_firecrawl_cls.assert_called_once_with(api_key="firecrawl-search-key")
mock_firecrawl_cls.return_value.search.assert_called_once_with("test query", limit=7)
class TestWebFetchTool:
@patch("deerflow.community.firecrawl.tools.FirecrawlApp")
@patch("deerflow.community.firecrawl.tools.get_app_config")
def test_fetch_uses_web_fetch_config(self, mock_get_app_config, mock_firecrawl_cls):
fetch_config = MagicMock()
fetch_config.model_extra = {"api_key": "firecrawl-fetch-key"}
def get_tool_config(name):
if name == "web_fetch":
return fetch_config
return None
mock_get_app_config.return_value.get_tool_config.side_effect = get_tool_config
mock_scrape_result = MagicMock()
mock_scrape_result.markdown = "Fetched markdown"
mock_scrape_result.metadata = MagicMock(title="Fetched Page")
mock_firecrawl_cls.return_value.scrape.return_value = mock_scrape_result
from deerflow.community.firecrawl.tools import web_fetch_tool
result = web_fetch_tool.invoke({"url": "https://example.com"})
assert result == "# Fetched Page\n\nFetched markdown"
mock_get_app_config.return_value.get_tool_config.assert_any_call("web_fetch")
mock_firecrawl_cls.assert_called_once_with(api_key="firecrawl-fetch-key")
mock_firecrawl_cls.return_value.scrape.assert_called_once_with(
"https://example.com",
formats=["markdown"],
)
+431
View File
@@ -0,0 +1,431 @@
"""Unit tests for the Setup Wizard (scripts/wizard/).
Run from repo root:
cd backend && uv run pytest tests/test_setup_wizard.py -v
"""
from __future__ import annotations
import yaml
from wizard.providers import LLM_PROVIDERS, SEARCH_PROVIDERS, WEB_FETCH_PROVIDERS
from wizard.steps import search as search_step
from wizard.writer import (
build_minimal_config,
read_env_file,
write_config_yaml,
write_env_file,
)
class TestProviders:
def test_llm_providers_not_empty(self):
assert len(LLM_PROVIDERS) >= 8
def test_llm_providers_have_required_fields(self):
for p in LLM_PROVIDERS:
assert p.name
assert p.display_name
assert p.use
assert ":" in p.use, f"Provider '{p.name}' use path must contain ':'"
assert p.models
assert p.default_model in p.models
def test_search_providers_have_required_fields(self):
for sp in SEARCH_PROVIDERS:
assert sp.name
assert sp.display_name
assert sp.use
assert ":" in sp.use
def test_search_and_fetch_include_firecrawl(self):
assert any(provider.name == "firecrawl" for provider in SEARCH_PROVIDERS)
assert any(provider.name == "firecrawl" for provider in WEB_FETCH_PROVIDERS)
def test_web_fetch_providers_have_required_fields(self):
for provider in WEB_FETCH_PROVIDERS:
assert provider.name
assert provider.display_name
assert provider.use
assert ":" in provider.use
assert provider.tool_name == "web_fetch"
def test_at_least_one_free_search_provider(self):
"""At least one search provider needs no API key."""
free = [sp for sp in SEARCH_PROVIDERS if sp.env_var is None]
assert free, "Expected at least one free (no-key) search provider"
def test_at_least_one_free_web_fetch_provider(self):
free = [provider for provider in WEB_FETCH_PROVIDERS if provider.env_var is None]
assert free, "Expected at least one free (no-key) web fetch provider"
class TestBuildMinimalConfig:
def test_produces_valid_yaml(self):
content = build_minimal_config(
provider_use="langchain_openai:ChatOpenAI",
model_name="gpt-4o",
display_name="OpenAI / gpt-4o",
api_key_field="api_key",
env_var="OPENAI_API_KEY",
)
data = yaml.safe_load(content)
assert data is not None
assert "models" in data
assert len(data["models"]) == 1
model = data["models"][0]
assert model["name"] == "gpt-4o"
assert model["use"] == "langchain_openai:ChatOpenAI"
assert model["model"] == "gpt-4o"
assert model["api_key"] == "$OPENAI_API_KEY"
def test_gemini_uses_gemini_api_key_field(self):
content = build_minimal_config(
provider_use="langchain_google_genai:ChatGoogleGenerativeAI",
model_name="gemini-2.0-flash",
display_name="Gemini",
api_key_field="gemini_api_key",
env_var="GEMINI_API_KEY",
)
data = yaml.safe_load(content)
model = data["models"][0]
assert "gemini_api_key" in model
assert model["gemini_api_key"] == "$GEMINI_API_KEY"
assert "api_key" not in model
def test_search_tool_included(self):
content = build_minimal_config(
provider_use="langchain_openai:ChatOpenAI",
model_name="gpt-4o",
display_name="OpenAI",
api_key_field="api_key",
env_var="OPENAI_API_KEY",
search_use="deerflow.community.tavily.tools:web_search_tool",
search_extra_config={"max_results": 5},
)
data = yaml.safe_load(content)
search_tool = next(t for t in data.get("tools", []) if t["name"] == "web_search")
assert search_tool["max_results"] == 5
def test_openrouter_defaults_are_preserved(self):
content = build_minimal_config(
provider_use="langchain_openai:ChatOpenAI",
model_name="google/gemini-2.5-flash-preview",
display_name="OpenRouter",
api_key_field="api_key",
env_var="OPENROUTER_API_KEY",
extra_model_config={
"base_url": "https://openrouter.ai/api/v1",
"request_timeout": 600.0,
"max_retries": 2,
"max_tokens": 8192,
"temperature": 0.7,
},
)
data = yaml.safe_load(content)
model = data["models"][0]
assert model["base_url"] == "https://openrouter.ai/api/v1"
assert model["request_timeout"] == 600.0
assert model["max_retries"] == 2
assert model["max_tokens"] == 8192
assert model["temperature"] == 0.7
def test_web_fetch_tool_included(self):
content = build_minimal_config(
provider_use="langchain_openai:ChatOpenAI",
model_name="gpt-4o",
display_name="OpenAI",
api_key_field="api_key",
env_var="OPENAI_API_KEY",
web_fetch_use="deerflow.community.jina_ai.tools:web_fetch_tool",
web_fetch_extra_config={"timeout": 10},
)
data = yaml.safe_load(content)
fetch_tool = next(t for t in data.get("tools", []) if t["name"] == "web_fetch")
assert fetch_tool["timeout"] == 10
def test_no_search_tool_when_not_configured(self):
content = build_minimal_config(
provider_use="langchain_openai:ChatOpenAI",
model_name="gpt-4o",
display_name="OpenAI",
api_key_field="api_key",
env_var="OPENAI_API_KEY",
)
data = yaml.safe_load(content)
tool_names = [t["name"] for t in data.get("tools", [])]
assert "web_search" not in tool_names
assert "web_fetch" not in tool_names
def test_sandbox_included(self):
content = build_minimal_config(
provider_use="langchain_openai:ChatOpenAI",
model_name="gpt-4o",
display_name="OpenAI",
api_key_field="api_key",
env_var="OPENAI_API_KEY",
)
data = yaml.safe_load(content)
assert "sandbox" in data
assert "use" in data["sandbox"]
assert data["sandbox"]["use"] == "deerflow.sandbox.local:LocalSandboxProvider"
assert data["sandbox"]["allow_host_bash"] is False
def test_bash_tool_disabled_by_default(self):
content = build_minimal_config(
provider_use="langchain_openai:ChatOpenAI",
model_name="gpt-4o",
display_name="OpenAI",
api_key_field="api_key",
env_var="OPENAI_API_KEY",
)
data = yaml.safe_load(content)
tool_names = [t["name"] for t in data.get("tools", [])]
assert "bash" not in tool_names
def test_can_enable_container_sandbox_and_bash(self):
content = build_minimal_config(
provider_use="langchain_openai:ChatOpenAI",
model_name="gpt-4o",
display_name="OpenAI",
api_key_field="api_key",
env_var="OPENAI_API_KEY",
sandbox_use="deerflow.community.aio_sandbox:AioSandboxProvider",
include_bash_tool=True,
)
data = yaml.safe_load(content)
assert data["sandbox"]["use"] == "deerflow.community.aio_sandbox:AioSandboxProvider"
assert "allow_host_bash" not in data["sandbox"]
tool_names = [t["name"] for t in data.get("tools", [])]
assert "bash" in tool_names
def test_can_disable_write_tools(self):
content = build_minimal_config(
provider_use="langchain_openai:ChatOpenAI",
model_name="gpt-4o",
display_name="OpenAI",
api_key_field="api_key",
env_var="OPENAI_API_KEY",
include_write_tools=False,
)
data = yaml.safe_load(content)
tool_names = [t["name"] for t in data.get("tools", [])]
assert "write_file" not in tool_names
assert "str_replace" not in tool_names
def test_config_version_present(self):
content = build_minimal_config(
provider_use="langchain_openai:ChatOpenAI",
model_name="gpt-4o",
display_name="OpenAI",
api_key_field="api_key",
env_var="OPENAI_API_KEY",
config_version=5,
)
data = yaml.safe_load(content)
assert data["config_version"] == 5
def test_cli_provider_does_not_emit_fake_api_key(self):
content = build_minimal_config(
provider_use="deerflow.models.openai_codex_provider:CodexChatModel",
model_name="gpt-5.4",
display_name="Codex CLI",
api_key_field="api_key",
env_var=None,
)
data = yaml.safe_load(content)
model = data["models"][0]
assert "api_key" not in model
# ---------------------------------------------------------------------------
# writer.py — env file helpers
# ---------------------------------------------------------------------------
class TestEnvFileHelpers:
def test_write_and_read_new_file(self, tmp_path):
env_file = tmp_path / ".env"
write_env_file(env_file, {"OPENAI_API_KEY": "sk-test123"})
pairs = read_env_file(env_file)
assert pairs["OPENAI_API_KEY"] == "sk-test123"
def test_update_existing_key(self, tmp_path):
env_file = tmp_path / ".env"
env_file.write_text("OPENAI_API_KEY=old-key\n")
write_env_file(env_file, {"OPENAI_API_KEY": "new-key"})
pairs = read_env_file(env_file)
assert pairs["OPENAI_API_KEY"] == "new-key"
# Should not duplicate
content = env_file.read_text()
assert content.count("OPENAI_API_KEY") == 1
def test_preserve_existing_keys(self, tmp_path):
env_file = tmp_path / ".env"
env_file.write_text("TAVILY_API_KEY=tavily-val\n")
write_env_file(env_file, {"OPENAI_API_KEY": "sk-new"})
pairs = read_env_file(env_file)
assert pairs["TAVILY_API_KEY"] == "tavily-val"
assert pairs["OPENAI_API_KEY"] == "sk-new"
def test_preserve_comments(self, tmp_path):
env_file = tmp_path / ".env"
env_file.write_text("# My .env file\nOPENAI_API_KEY=old\n")
write_env_file(env_file, {"OPENAI_API_KEY": "new"})
content = env_file.read_text()
assert "# My .env file" in content
def test_read_ignores_comments(self, tmp_path):
env_file = tmp_path / ".env"
env_file.write_text("# comment\nKEY=value\n")
pairs = read_env_file(env_file)
assert "# comment" not in pairs
assert pairs["KEY"] == "value"
# ---------------------------------------------------------------------------
# writer.py — write_config_yaml
# ---------------------------------------------------------------------------
class TestWriteConfigYaml:
def test_generated_config_loadable_by_appconfig(self, tmp_path):
"""The generated config.yaml must be parseable (basic YAML validity)."""
config_path = tmp_path / "config.yaml"
write_config_yaml(
config_path,
provider_use="langchain_openai:ChatOpenAI",
model_name="gpt-4o",
display_name="OpenAI / gpt-4o",
api_key_field="api_key",
env_var="OPENAI_API_KEY",
)
assert config_path.exists()
with open(config_path) as f:
data = yaml.safe_load(f)
assert isinstance(data, dict)
assert "models" in data
def test_copies_example_defaults_for_unconfigured_sections(self, tmp_path):
example_path = tmp_path / "config.example.yaml"
example_path.write_text(
yaml.safe_dump(
{
"config_version": 5,
"log_level": "info",
"token_usage": {"enabled": False},
"tool_groups": [{"name": "web"}, {"name": "file:read"}, {"name": "file:write"}, {"name": "bash"}],
"tools": [
{
"name": "web_search",
"group": "web",
"use": "deerflow.community.ddg_search.tools:web_search_tool",
"max_results": 5,
},
{
"name": "web_fetch",
"group": "web",
"use": "deerflow.community.jina_ai.tools:web_fetch_tool",
"timeout": 10,
},
{
"name": "image_search",
"group": "web",
"use": "deerflow.community.image_search.tools:image_search_tool",
"max_results": 5,
},
{"name": "ls", "group": "file:read", "use": "deerflow.sandbox.tools:ls_tool"},
{"name": "write_file", "group": "file:write", "use": "deerflow.sandbox.tools:write_file_tool"},
{"name": "bash", "group": "bash", "use": "deerflow.sandbox.tools:bash_tool"},
],
"sandbox": {
"use": "deerflow.sandbox.local:LocalSandboxProvider",
"allow_host_bash": False,
},
"summarization": {"max_tokens": 2048},
},
sort_keys=False,
)
)
config_path = tmp_path / "config.yaml"
write_config_yaml(
config_path,
provider_use="langchain_openai:ChatOpenAI",
model_name="gpt-4o",
display_name="OpenAI / gpt-4o",
api_key_field="api_key",
env_var="OPENAI_API_KEY",
)
with open(config_path) as f:
data = yaml.safe_load(f)
assert data["log_level"] == "info"
assert data["token_usage"]["enabled"] is False
assert data["tool_groups"][0]["name"] == "web"
assert data["summarization"]["max_tokens"] == 2048
assert any(tool["name"] == "image_search" and tool["max_results"] == 5 for tool in data["tools"])
def test_config_version_read_from_example(self, tmp_path):
"""write_config_yaml should read config_version from config.example.yaml if present."""
example_path = tmp_path / "config.example.yaml"
example_path.write_text("config_version: 99\n")
config_path = tmp_path / "config.yaml"
write_config_yaml(
config_path,
provider_use="langchain_openai:ChatOpenAI",
model_name="gpt-4o",
display_name="OpenAI",
api_key_field="api_key",
env_var="OPENAI_API_KEY",
)
with open(config_path) as f:
data = yaml.safe_load(f)
assert data["config_version"] == 99
def test_model_base_url_from_extra_config(self, tmp_path):
config_path = tmp_path / "config.yaml"
write_config_yaml(
config_path,
provider_use="langchain_openai:ChatOpenAI",
model_name="google/gemini-2.5-flash-preview",
display_name="OpenRouter",
api_key_field="api_key",
env_var="OPENROUTER_API_KEY",
extra_model_config={"base_url": "https://openrouter.ai/api/v1"},
)
with open(config_path) as f:
data = yaml.safe_load(f)
assert data["models"][0]["base_url"] == "https://openrouter.ai/api/v1"
class TestSearchStep:
def test_reuses_api_key_for_same_provider(self, monkeypatch):
monkeypatch.setattr(search_step, "print_header", lambda *_args, **_kwargs: None)
monkeypatch.setattr(search_step, "print_success", lambda *_args, **_kwargs: None)
monkeypatch.setattr(search_step, "print_info", lambda *_args, **_kwargs: None)
choices = iter([3, 1])
prompts: list[str] = []
def fake_choice(_prompt, _options, default=0):
return next(choices)
def fake_secret(prompt):
prompts.append(prompt)
return "shared-api-key"
monkeypatch.setattr(search_step, "ask_choice", fake_choice)
monkeypatch.setattr(search_step, "ask_secret", fake_secret)
result = search_step.run_search_step()
assert result.search_provider is not None
assert result.fetch_provider is not None
assert result.search_provider.name == "exa"
assert result.fetch_provider.name == "exa"
assert result.search_api_key == "shared-api-key"
assert result.fetch_api_key == "shared-api-key"
assert prompts == ["EXA_API_KEY"]