from __future__ import annotations import json import os from pathlib import Path import pytest import yaml from pydantic import ValidationError import deerflow.config.app_config as app_config_module from deerflow.config.acp_config import load_acp_config_from_dict from deerflow.config.agents_api_config import get_agents_api_config, load_agents_api_config_from_dict from deerflow.config.app_config import AppConfig, get_app_config, reset_app_config from deerflow.config.checkpointer_config import get_checkpointer_config, load_checkpointer_config_from_dict from deerflow.config.database_config import DatabaseConfig from deerflow.config.guardrails_config import get_guardrails_config, load_guardrails_config_from_dict from deerflow.config.memory_config import get_memory_config, load_memory_config_from_dict from deerflow.config.stream_bridge_config import get_stream_bridge_config, load_stream_bridge_config_from_dict from deerflow.config.subagents_config import get_subagents_app_config, load_subagents_config_from_dict from deerflow.config.summarization_config import get_summarization_config, load_summarization_config_from_dict from deerflow.config.title_config import get_title_config, load_title_config_from_dict from deerflow.config.tool_search_config import get_tool_search_config, load_tool_search_config_from_dict from deerflow.runtime.checkpointer import get_checkpointer, reset_checkpointer from deerflow.runtime.store import get_store, reset_store def _reset_config_singletons() -> None: load_title_config_from_dict({}) load_summarization_config_from_dict({}) load_memory_config_from_dict({}) load_agents_api_config_from_dict({}) load_subagents_config_from_dict({}) load_tool_search_config_from_dict({}) load_guardrails_config_from_dict({}) load_checkpointer_config_from_dict(None) load_stream_bridge_config_from_dict(None) load_acp_config_from_dict({}) reset_checkpointer() reset_store() reset_app_config() def _write_config(path: Path, *, model_name: str, supports_thinking: bool) -> None: path.write_text( yaml.safe_dump( { "sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"}, "models": [ { "name": model_name, "use": "langchain_openai:ChatOpenAI", "model": "gpt-test", "supports_thinking": supports_thinking, } ], } ), encoding="utf-8", ) def _write_config_with_agents_api( path: Path, *, model_name: str, supports_thinking: bool, agents_api: dict | None = None, ) -> None: config = { "sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"}, "models": [ { "name": model_name, "use": "langchain_openai:ChatOpenAI", "model": "gpt-test", "supports_thinking": supports_thinking, } ], } if agents_api is not None: config["agents_api"] = agents_api path.write_text(yaml.safe_dump(config), encoding="utf-8") def _write_config_with_sections(path: Path, sections: dict | None = None) -> None: config = { "sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"}, "models": [ { "name": "first-model", "use": "langchain_openai:ChatOpenAI", "model": "gpt-test", } ], } if sections: config.update(sections) path.write_text(yaml.safe_dump(config), encoding="utf-8") def _write_extensions_config(path: Path) -> None: path.write_text(json.dumps({"mcpServers": {}, "skills": {}}), encoding="utf-8") def test_checkpoint_channel_mode_defaults_to_full() -> None: assert DatabaseConfig().checkpoint_channel_mode == "full" @pytest.mark.parametrize("mode", ["full", "delta"]) def test_checkpoint_channel_mode_accepts_supported_values(mode: str) -> None: assert DatabaseConfig(checkpoint_channel_mode=mode).checkpoint_channel_mode == mode def test_checkpoint_channel_mode_rejects_unknown_value() -> None: with pytest.raises(ValidationError): DatabaseConfig(checkpoint_channel_mode="auto") def test_config_example_does_not_enable_empty_extensions_block_by_default(): config_example_path = Path(__file__).resolve().parents[2] / "config.example.yaml" config_data = yaml.safe_load(config_example_path.read_text(encoding="utf-8")) assert "extensions" not in config_data def test_app_config_defaults_missing_database_to_sqlite(tmp_path, monkeypatch): config_path = tmp_path / "config.yaml" extensions_path = tmp_path / "extensions_config.json" _write_extensions_config(extensions_path) _write_config(config_path, model_name="first-model", supports_thinking=False) monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(extensions_path)) config = AppConfig.from_file(str(config_path)) assert config.database.backend == "sqlite" assert config.database.sqlite_dir == ".deer-flow/data" def test_app_config_preserves_config_yaml_extension_middlewares(tmp_path, monkeypatch): config_path = tmp_path / "config.yaml" extensions_path = tmp_path / "extensions_config.json" extensions_path.write_text( json.dumps({"mcpServers": {}, "skills": {}, "middlewares": ["pkg.from_file:FileMiddleware"]}), encoding="utf-8", ) _write_config_with_sections( config_path, { "extensions": { "middlewares": ["pkg.from_yaml:YamlMiddleware"], } }, ) monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(extensions_path)) config = AppConfig.from_file(str(config_path)) assert config.extensions.middlewares == ["pkg.from_yaml:YamlMiddleware"] def test_app_config_normalizes_config_yaml_extension_aliases_before_override(tmp_path, monkeypatch): config_path = tmp_path / "config.yaml" extensions_path = tmp_path / "extensions_config.json" extensions_path.write_text( json.dumps( { "mcpServers": { "from-file": { "command": "file-mcp", } }, "skills": {}, } ), encoding="utf-8", ) _write_config_with_sections( config_path, { "extensions": { "mcp_servers": { "from-yaml": { "command": "yaml-mcp", } }, } }, ) monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(extensions_path)) config = AppConfig.from_file(str(config_path)) assert set(config.extensions.mcp_servers) == {"from-yaml"} assert config.extensions.mcp_servers["from-yaml"].command == "yaml-mcp" def test_app_config_loads_extension_middlewares_from_extensions_config(tmp_path, monkeypatch): config_path = tmp_path / "config.yaml" extensions_path = tmp_path / "extensions_config.json" extensions_path.write_text( json.dumps({"mcpServers": {}, "skills": {}, "middlewares": ["pkg.from_file:FileMiddleware"]}), encoding="utf-8", ) _write_config_with_sections(config_path) monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(extensions_path)) config = AppConfig.from_file(str(config_path)) assert config.extensions.middlewares == ["pkg.from_file:FileMiddleware"] def test_app_config_defaults_empty_database_to_sqlite(tmp_path, monkeypatch): config_path = tmp_path / "config.yaml" extensions_path = tmp_path / "extensions_config.json" _write_extensions_config(extensions_path) config_path.write_text( yaml.safe_dump( { "database": {}, "sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"}, } ), encoding="utf-8", ) monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(extensions_path)) config = AppConfig.from_file(str(config_path)) assert config.database.backend == "sqlite" assert config.database.sqlite_dir == ".deer-flow/data" def test_app_config_coerces_commented_out_list_sections(tmp_path, monkeypatch): """Commenting out every entry under a list key makes PyYAML parse it as None. Regression for the documented ``cp config.example.yaml config.yaml`` flow (issue #1444): such a config must load with empty lists instead of raising ``Input should be a valid list``. """ config_path = tmp_path / "config.yaml" extensions_path = tmp_path / "extensions_config.json" _write_extensions_config(extensions_path) config_path.write_text( yaml.safe_dump( { "sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"}, "models": None, "tools": None, "tool_groups": None, } ), encoding="utf-8", ) monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(extensions_path)) config = AppConfig.from_file(str(config_path)) assert config.models == [] assert config.tools == [] assert config.tool_groups == [] def test_app_config_coerces_commented_out_object_sections(tmp_path, monkeypatch): """Commenting out every entry under an object key makes PyYAML parse it as None. Same documented ``cp config.example.yaml config.yaml`` flow as the list sections: object sections (memory, summarization, ...) must fall back to their defaults instead of raising ``Input should be a valid dictionary``. """ config_path = tmp_path / "config.yaml" extensions_path = tmp_path / "extensions_config.json" _write_extensions_config(extensions_path) config_path.write_text( yaml.safe_dump( { "sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"}, "memory": None, "summarization": None, "guardrails": None, "tool_output": None, "run_events": None, } ), encoding="utf-8", ) monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(extensions_path)) config = AppConfig.from_file(str(config_path)) # Each present-but-null object section falls back to a real default config # object of the expected type (not merely non-None). assert type(config.memory).__name__ == "MemoryConfig" assert type(config.summarization).__name__ == "SummarizationConfig" assert type(config.guardrails).__name__ == "GuardrailsConfig" assert type(config.tool_output).__name__ == "ToolOutputConfig" assert type(config.run_events).__name__ == "RunEventsConfig" def test_app_config_null_required_section_still_errors(tmp_path, monkeypatch): """A present-but-null *required* section still errors. ``sandbox`` has no default, so dropping a ``sandbox: null`` key leaves the required field absent — there is nothing to fall back to (per ``_drop_null_config_sections``), unlike the optional object sections above. """ config_path = tmp_path / "config.yaml" extensions_path = tmp_path / "extensions_config.json" _write_extensions_config(extensions_path) config_path.write_text(yaml.safe_dump({"sandbox": None}), encoding="utf-8") monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(extensions_path)) with pytest.raises(ValidationError): AppConfig.from_file(str(config_path)) def test_app_config_warns_when_no_models_configured(tmp_path, monkeypatch, caplog): config_path = tmp_path / "config.yaml" extensions_path = tmp_path / "extensions_config.json" _write_extensions_config(extensions_path) config_path.write_text( yaml.safe_dump( { "sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"}, "models": None, } ), encoding="utf-8", ) monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(extensions_path)) with caplog.at_level("WARNING", logger="deerflow.config.app_config"): AppConfig.from_file(str(config_path)) assert "No models are configured" in caplog.text def test_get_app_config_reloads_when_file_changes(tmp_path, monkeypatch): config_path = tmp_path / "config.yaml" extensions_path = tmp_path / "extensions_config.json" _write_extensions_config(extensions_path) _write_config(config_path, model_name="first-model", supports_thinking=False) monkeypatch.setenv("DEER_FLOW_CONFIG_PATH", str(config_path)) monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(extensions_path)) reset_app_config() try: initial = get_app_config() assert initial.models[0].supports_thinking is False _write_config(config_path, model_name="first-model", supports_thinking=True) next_mtime = config_path.stat().st_mtime + 5 os.utime(config_path, (next_mtime, next_mtime)) reloaded = get_app_config() assert reloaded.models[0].supports_thinking is True assert reloaded is not initial finally: reset_app_config() def test_get_app_config_reloads_when_content_digest_changes_without_metadata(tmp_path, monkeypatch): config_path = tmp_path / "config.yaml" extensions_path = tmp_path / "extensions_config.json" _write_extensions_config(extensions_path) _write_config(config_path, model_name="model-a", supports_thinking=False) monkeypatch.setenv("DEER_FLOW_CONFIG_PATH", str(config_path)) monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(extensions_path)) _reset_config_singletons() try: initial = get_app_config() initial_mtime = app_config_module._app_config_mtime initial_signature = app_config_module._app_config_signature assert initial.models[0].name == "model-a" assert initial_signature is not None _write_config(config_path, model_name="model-b", supports_thinking=False) real_get_config_signature = app_config_module._get_config_signature def stale_metadata_signature(path: Path): current_signature = real_get_config_signature(path) assert current_signature is not None return (initial_signature[0], initial_signature[1], current_signature[2]) monkeypatch.setattr(app_config_module, "_get_config_mtime", lambda _path: initial_mtime) monkeypatch.setattr(app_config_module, "_get_config_signature", stale_metadata_signature) reloaded = get_app_config() assert reloaded.models[0].name == "model-b" assert reloaded is not initial assert app_config_module._app_config_signature is not None assert app_config_module._app_config_signature[:2] == initial_signature[:2] assert app_config_module._app_config_signature[2] != initial_signature[2] finally: _reset_config_singletons() def test_get_app_config_reloads_when_config_path_changes(tmp_path, monkeypatch): config_a = tmp_path / "config-a.yaml" config_b = tmp_path / "config-b.yaml" extensions_path = tmp_path / "extensions_config.json" _write_extensions_config(extensions_path) _write_config(config_a, model_name="model-a", supports_thinking=False) _write_config(config_b, model_name="model-b", supports_thinking=True) monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(extensions_path)) monkeypatch.setenv("DEER_FLOW_CONFIG_PATH", str(config_a)) reset_app_config() try: first = get_app_config() assert first.models[0].name == "model-a" monkeypatch.setenv("DEER_FLOW_CONFIG_PATH", str(config_b)) second = get_app_config() assert second.models[0].name == "model-b" assert second is not first finally: reset_app_config() def test_get_app_config_resets_agents_api_config_when_section_removed(tmp_path, monkeypatch): config_path = tmp_path / "config.yaml" extensions_path = tmp_path / "extensions_config.json" _write_extensions_config(extensions_path) _write_config_with_agents_api( config_path, model_name="first-model", supports_thinking=False, agents_api={"enabled": True}, ) monkeypatch.setenv("DEER_FLOW_CONFIG_PATH", str(config_path)) monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(extensions_path)) reset_app_config() try: initial = get_app_config() assert initial.models[0].name == "first-model" assert get_agents_api_config().enabled is True _write_config_with_agents_api( config_path, model_name="first-model", supports_thinking=False, ) next_mtime = config_path.stat().st_mtime + 5 os.utime(config_path, (next_mtime, next_mtime)) reloaded = get_app_config() assert reloaded is not initial assert get_agents_api_config().enabled is False finally: reset_app_config() def test_get_app_config_resets_singleton_configs_when_sections_removed(tmp_path, monkeypatch): config_path = tmp_path / "config.yaml" extensions_path = tmp_path / "extensions_config.json" _write_extensions_config(extensions_path) _write_config_with_sections( config_path, { "title": {"enabled": False, "max_words": 3}, "summarization": {"enabled": True}, "memory": {"enabled": False, "max_facts": 50}, "subagents": {"timeout_seconds": 42, "agents": {"reviewer": {"max_turns": 2}}}, "tool_search": {"enabled": True}, "guardrails": {"enabled": True, "fail_closed": False}, "checkpointer": {"type": "memory"}, "stream_bridge": {"type": "memory", "queue_maxsize": 12}, }, ) monkeypatch.setenv("DEER_FLOW_CONFIG_PATH", str(config_path)) monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(extensions_path)) reset_app_config() try: get_app_config() assert get_title_config().enabled is False assert get_summarization_config().enabled is True assert get_memory_config().enabled is False assert get_subagents_app_config().timeout_seconds == 42 assert get_tool_search_config().enabled is True assert get_guardrails_config().enabled is True assert get_checkpointer_config() is not None assert get_stream_bridge_config() is not None _write_config_with_sections(config_path) next_mtime = config_path.stat().st_mtime + 5 os.utime(config_path, (next_mtime, next_mtime)) get_app_config() assert get_title_config().enabled is True assert get_summarization_config().enabled is False assert get_memory_config().enabled is True assert get_subagents_app_config().timeout_seconds == 1800 assert get_tool_search_config().enabled is False assert get_guardrails_config().enabled is False assert get_checkpointer_config() is None assert get_stream_bridge_config() is None finally: _reset_config_singletons() def test_get_app_config_resets_persistence_runtime_singletons_when_checkpointer_removed(tmp_path, monkeypatch): config_path = tmp_path / "config.yaml" extensions_path = tmp_path / "extensions_config.json" _write_extensions_config(extensions_path) _write_config_with_sections(config_path, {"checkpointer": {"type": "memory"}}) monkeypatch.setenv("DEER_FLOW_CONFIG_PATH", str(config_path)) monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(extensions_path)) reset_checkpointer() reset_store() reset_app_config() try: get_app_config() initial_checkpointer = get_checkpointer() initial_store = get_store() _write_config_with_sections(config_path) next_mtime = config_path.stat().st_mtime + 5 os.utime(config_path, (next_mtime, next_mtime)) get_app_config() assert get_checkpointer_config() is None assert get_checkpointer() is not initial_checkpointer assert get_store() is not initial_store finally: _reset_config_singletons() def test_get_app_config_keeps_persistence_runtime_singletons_when_checkpointer_unchanged(tmp_path, monkeypatch): config_path = tmp_path / "config.yaml" extensions_path = tmp_path / "extensions_config.json" _write_extensions_config(extensions_path) _write_config_with_sections( config_path, { "title": {"enabled": False}, "checkpointer": {"type": "memory"}, }, ) monkeypatch.setenv("DEER_FLOW_CONFIG_PATH", str(config_path)) monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(extensions_path)) _reset_config_singletons() try: get_app_config() initial_checkpointer = get_checkpointer() initial_store = get_store() _write_config_with_sections( config_path, { "title": {"enabled": True}, "checkpointer": {"type": "memory"}, }, ) next_mtime = config_path.stat().st_mtime + 5 os.utime(config_path, (next_mtime, next_mtime)) get_app_config() assert get_checkpointer() is initial_checkpointer assert get_store() is initial_store finally: _reset_config_singletons() def test_get_app_config_does_not_mutate_singletons_when_reload_validation_fails(tmp_path, monkeypatch): config_path = tmp_path / "config.yaml" extensions_path = tmp_path / "extensions_config.json" _write_extensions_config(extensions_path) _write_config_with_sections( config_path, { "title": {"enabled": False}, "tool_search": {"enabled": True}, "checkpointer": {"type": "memory"}, }, ) monkeypatch.setenv("DEER_FLOW_CONFIG_PATH", str(config_path)) monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(extensions_path)) _reset_config_singletons() try: previous_app_config = get_app_config() initial_checkpointer = get_checkpointer() initial_store = get_store() _write_config_with_sections( config_path, { "title": False, "tool_search": False, "checkpointer": {"type": "memory"}, }, ) next_mtime = config_path.stat().st_mtime + 5 os.utime(config_path, (next_mtime, next_mtime)) with pytest.raises(ValidationError): get_app_config() assert app_config_module._app_config is previous_app_config assert get_title_config().enabled is False assert get_tool_search_config().enabled is True assert get_checkpointer_config() is not None assert get_checkpointer() is initial_checkpointer assert get_store() is initial_store finally: _reset_config_singletons() def test_get_memory_config_self_syncs_without_prior_get_app_config(tmp_path, monkeypatch): """get_memory_config() triggers reload when file changes without prior get_app_config(). Background memory paths (updater/queue/storage) never call get_app_config() directly. This test pins the fix: calling get_memory_config() in isolation — after mutating the file, with no intervening get_app_config() — reflects the new value. """ config_path = tmp_path / "config.yaml" extensions_path = tmp_path / "extensions_config.json" _write_extensions_config(extensions_path) _write_config_with_sections(config_path, {"memory": {"enabled": False}}) monkeypatch.setenv("DEER_FLOW_CONFIG_PATH", str(config_path)) monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(extensions_path)) reset_app_config() try: get_app_config() assert get_memory_config().enabled is False _write_config_with_sections(config_path, {"memory": {"enabled": True}}) next_mtime = config_path.stat().st_mtime + 5 os.utime(config_path, (next_mtime, next_mtime)) assert get_memory_config().enabled is True finally: _reset_config_singletons() def test_get_memory_config_falls_back_on_broken_config(tmp_path, monkeypatch): """get_memory_config() does not crash on transiently broken config.yaml. get_app_config() can raise yaml.YAMLError, ValidationError, or ValueError. get_memory_config() catches them and returns the last-good singleton. """ config_path = tmp_path / "config.yaml" extensions_path = tmp_path / "extensions_config.json" _write_extensions_config(extensions_path) _write_config_with_sections(config_path, {"memory": {"enabled": False}}) monkeypatch.setenv("DEER_FLOW_CONFIG_PATH", str(config_path)) monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(extensions_path)) reset_app_config() try: get_app_config() assert get_memory_config().enabled is False config_path.write_text("memory: {enabled: true\n") next_mtime = config_path.stat().st_mtime + 5 os.utime(config_path, (next_mtime, next_mtime)) assert get_memory_config().enabled is False finally: _reset_config_singletons()