diff --git a/backend/tests/test_extension_execution_fence.py b/backend/tests/test_extension_execution_fence.py index 28c497db71..5f07edb62a 100644 --- a/backend/tests/test_extension_execution_fence.py +++ b/backend/tests/test_extension_execution_fence.py @@ -60,6 +60,11 @@ async def test_postgres_takeover_fences_admission_after_lock_wait(): import os import uuid + dsn = os.getenv("DEERFLOW_TEST_WORKFLOW_POSTGRES") + if not dsn: + pytest.skip("Requires dedicated PostgreSQL") + + # The default install does not need optional drivers without this opt-in. import psycopg from psycopg import sql from sqlalchemy import update @@ -69,9 +74,6 @@ async def test_postgres_takeover_fences_admission_after_lock_wait(): from deerflow.persistence.run.model import RunChangeClockRow, RunRow from deerflow.runtime.runs.manager import ConflictError - dsn = os.getenv("DEERFLOW_TEST_WORKFLOW_POSTGRES") - if not dsn: - pytest.skip("Requires dedicated PostgreSQL") schema = "fence_test_" + uuid.uuid4().hex with psycopg.connect(dsn, autocommit=True) as db: db.execute(sql.SQL("CREATE SCHEMA {}").format(sql.Identifier(schema))) diff --git a/backend/tests/test_local_grep_io_errors.py b/backend/tests/test_local_grep_io_errors.py index c3a889c4ea..4e5c0c2d5d 100644 --- a/backend/tests/test_local_grep_io_errors.py +++ b/backend/tests/test_local_grep_io_errors.py @@ -32,6 +32,12 @@ def mounted_search(tmp_path, monkeypatch): sandbox=SandboxConfig(use="deerflow.sandbox.local:LocalSandboxProvider", mounts=[VolumeMountConfig(host_path=str(source), container_path="/mnt/mounts/data")]), ) monkeypatch.setattr("deerflow.config.get_app_config", lambda: config) + from deerflow.sandbox.tools import _get_custom_mounts + + # Tests often run after app initialization has cached mounts from the host + # config. This sandbox uses a different config, so resolve its mount from + # the fixture config instead of reusing that process-wide snapshot. + monkeypatch.setattr(_get_custom_mounts, "_cached", None, raising=False) provider = LocalSandboxProvider() sandbox = provider.get(provider.acquire(None)) monkeypatch.setattr("deerflow.sandbox.tools.ensure_sandbox_initialized", lambda _: sandbox) @@ -55,8 +61,12 @@ def test_grep_custom_mount_errors_use_virtual_paths(mounted_search, monkeypatch, raise OSError(error_number, "read failed", str(path)) return original_open(path, mode, *args, **kwargs) + scan_errors: list[OSError] = [] + def fail_walk(root, *, onerror): - onerror(OSError(error_number, "scan failed", str(root))) + error = OSError(error_number, "scan failed", str(root)) + scan_errors.append(error) + onerror(error) return iter(()) if stage == "scan": @@ -66,6 +76,8 @@ def test_grep_custom_mount_errors_use_virtual_paths(mounted_search, monkeypatch, expected = "/mnt/mounts/data" + ("" if stage == "scan" else "/secret.txt") result = invoke() assert result.startswith("Error:"), result + if stage == "scan": + assert scan_errors, "mounted grep must reach the injected filesystem scan error" assert expected in result assert "private-host-mount" not in result with pytest.raises(OSError) as caught: diff --git a/backend/tests/test_make_test_parallel.py b/backend/tests/test_make_test_parallel.py index 25534394fe..9abb69ff80 100644 --- a/backend/tests/test_make_test_parallel.py +++ b/backend/tests/test_make_test_parallel.py @@ -63,14 +63,16 @@ def run_make(project: Path, *args: str, fail_group: str = "", parallel: bool = T "EXPECT_PARALLEL": "1" if parallel else "0", "FAIL_GROUP": fail_group, } - env.pop("MAKEFLAGS", None) - env.pop("MFLAGS", None) - env.pop("MAKELEVEL", None) + # Make exports command-line variables to pytest. This copied project must + # exercise its own defaults, not the outer test runner's shard/job settings. + for name in ("MAKEFLAGS", "MFLAGS", "MAKELEVEL", "MAKEOVERRIDES", "TEST_JOBS", "SPLITS", "GROUP", "DURATIONS_FILE"): + env.pop(name, None) return subprocess.run([MAKE, *args], cwd=project, env=env, capture_output=True, text=True, encoding="utf-8", check=False, timeout=30) @pytest.mark.parametrize("fail_group", ["", "1"]) -def test_make_test_runs_all_four_shards_and_reports_any_failure(test_project: Path, fail_group: str) -> None: +def test_make_test_runs_all_four_shards_and_reports_any_failure(test_project: Path, fail_group: str, monkeypatch) -> None: + monkeypatch.setenv("TEST_JOBS", "2") result = run_make(test_project, "test", fail_group=fail_group) assert (result.returncode == 0) == (not fail_group), result.stdout + result.stderr diff --git a/backend/tests/test_memory_pii_redaction.py b/backend/tests/test_memory_pii_redaction.py index 0f5496cb01..613669e3ea 100644 --- a/backend/tests/test_memory_pii_redaction.py +++ b/backend/tests/test_memory_pii_redaction.py @@ -105,6 +105,7 @@ def _flush_hook_call(monkeypatch, pii_config, messages): from deerflow.agents.middlewares.summarization_middleware import SummarizationEvent manager = MagicMock() + monkeypatch.setattr(hook_module, "get_memory_config", lambda: MemoryConfig(enabled=True)) monkeypatch.setattr(hook_module, "get_memory_manager", lambda: manager) monkeypatch.setattr(hook_module, "resolve_runtime_user_id", lambda runtime: "u1") event = SummarizationEvent( @@ -141,6 +142,26 @@ def test_compaction_flush_hook_untouched_when_disabled(monkeypatch): assert "alice@example.com" in queued[0].content +def test_compaction_flush_hook_does_not_capture_when_memory_is_disabled(monkeypatch): + import deerflow.agents.memory.summarization_hook as hook_module + from deerflow.agents.middlewares.summarization_middleware import SummarizationEvent + + manager_factory = MagicMock() + monkeypatch.setattr(hook_module, "get_memory_config", lambda: MemoryConfig(enabled=False)) + monkeypatch.setattr(hook_module, "get_memory_manager", manager_factory) + event = SummarizationEvent( + messages_to_summarize=(HumanMessage("reach alice@example.com"),), + preserved_messages=(), + thread_id="thread-123", + agent_name="researcher", + runtime=None, + ) + + hook_module.memory_flush_hook(event, pii_redaction_config=PiiRedactionConfig(enabled=True, token_secret=_TOKEN_SECRET)) + + manager_factory.assert_not_called() + + def test_tool_call_args_redacted(monkeypatch): # OpenViking-style retention keeps the full message object, including # parsed tool_calls and provider-format arguments in additional_kwargs. diff --git a/backend/tests/test_openviking_memory_backend.py b/backend/tests/test_openviking_memory_backend.py index 3a7547cb52..abc9c50c14 100644 --- a/backend/tests/test_openviking_memory_backend.py +++ b/backend/tests/test_openviking_memory_backend.py @@ -11,6 +11,7 @@ import weakref from collections.abc import Iterator from contextlib import contextmanager from contextvars import ContextVar +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any @@ -33,6 +34,65 @@ from deerflow.agents.memory.manager import ( from deerflow.agents.middlewares.dynamic_context_middleware import DynamicContextMiddleware +@pytest.fixture(autouse=True) +def isolated_cli_config(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + config_path = tmp_path / "ovcli.conf" + config_path.write_text("{}", encoding="utf-8") + monkeypatch.setenv("OPENVIKING_CLI_CONFIG_FILE", str(config_path)) + for name in ("OPENVIKING_URL", "OPENVIKING_API_KEY", "OPENVIKING_ACCOUNT", "OPENVIKING_USER", "OPENVIKING_ACTOR_PEER_ID", "OPENVIKING_TIMEOUT"): + monkeypatch.delenv(name, raising=False) + return config_path + + +@pytest.fixture +def openviking_http_server(monkeypatch: pytest.MonkeyPatch) -> Iterator[tuple[str, list[dict[str, Any]]]]: + # This fixture exercises a direct loopback transport, not host proxy routing. + for name in ("NO_PROXY", "no_proxy"): + monkeypatch.setenv(name, "127.0.0.1,localhost,::1") + requests: list[dict[str, Any]] = [] + + class Handler(BaseHTTPRequestHandler): + def do_GET(self) -> None: + requests.append({"path": self.path, "headers": {key.lower(): value for key, value in self.headers.items()}}) + self._respond({"status": "ok"}) + + def do_POST(self) -> None: + payload = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) + requests.append({"path": self.path, "headers": {key.lower(): value for key, value in self.headers.items()}, "payload": payload}) + if self.path == "/api/v1/search/find": + result = {"memories": [{"uri": "viking://user/alice/memories/style.md", "category": "preferences", "overview": "Prefers concise answers.", "score": 0.9}]} + elif self.path.endswith("/messages/batch"): + result = {"pending_tokens": 2} + elif self.path.endswith("/commit"): + result = {"status": "committed"} + else: + self.send_error(404) + return + self._respond({"status": "success", "result": result}) + + def _respond(self, body: dict[str, Any]) -> None: + encoded = json.dumps(body).encode("utf-8") + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(encoded))) + self.end_headers() + self.wfile.write(encoded) + + def log_message(self, format: str, *args: Any) -> None: + pass + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + thread = threading.Thread(target=lambda: server.serve_forever(poll_interval=0.01), daemon=True) + thread.start() + try: + yield f"http://127.0.0.1:{server.server_port}", requests + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5) + assert not thread.is_alive() + + class _CommitPolicy: def __init__(self, *, mode: str, pending_token_threshold: int = 8_000): self.mode = mode @@ -394,6 +454,61 @@ def test_official_loader_uses_standalone_package() -> None: assert integration["OpenVikingRetriever"].__module__.startswith("langchain_openviking") +@pytest.mark.parametrize("cli_profile", ["missing", "empty", "foreign", "oidc-env", "ldap-env"]) +def test_user_key_connection_uses_owner_key_without_ambient_authorization( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + isolated_cli_config: Path, + openviking_http_server: tuple[str, list[dict[str, Any]]], + cli_profile: str, +) -> None: + if cli_profile == "missing": + isolated_cli_config.unlink() + elif cli_profile == "foreign": + isolated_cli_config.write_text(json.dumps({"url": "http://unused.invalid", "api_key": "cli-key", "extra_headers": {"X-CLI-Only": "must-not-leak", "X-API-Key": "cli-header-key"}}), encoding="utf-8") + if cli_profile in {"oidc-env", "ldap-env"}: + monkeypatch.setenv("OPENVIKING_AUTH_MODE", cli_profile.removesuffix("-env")) + monkeypatch.setenv("OPENVIKING_OIDC_TOKEN", "foreign-token") + monkeypatch.setenv("OPENVIKING_USERNAME", "foreign-user") + monkeypatch.setenv("OPENVIKING_PASSWORD", "foreign-password") + base_url, requests = openviking_http_server + manager = _manager(tmp_path, monkeypatch, base_url=base_url, timeout_seconds=1, failure_policy={"read": "raise"}) + try: + assert manager.warm() is True + facts = manager.search("response style", top_k=2, user_id="alice") + assert facts == [{"id": "viking://user/alice/memories/style.md", "content": "Prefers concise answers.", "category": "preferences", "confidence": 0.9, "source": "viking://user/alice/memories/style.md", "score": 0.9}] + assert [request["path"] for request in requests] == ["/health", "/api/v1/search/find"] + assert requests[-1]["payload"]["query"] == "response style" + assert requests[-1]["payload"]["limit"] == 2 + + messages = [HumanMessage("remember short replies"), AIMessage("noted")] + manager.add("thread-1", messages, user_id="alice") + batch, commit = requests[2:] + assert batch["path"].endswith("/messages/batch") + assert commit["path"] == batch["path"].removesuffix("/messages/batch") + "/commit" + assert [message["role"] for message in batch["payload"]["messages"]] == ["user", "assistant"] + assert [message["parts"] for message in batch["payload"]["messages"]] == [[{"type": "text", "text": "remember short replies"}], [{"type": "text", "text": "noted"}]] + assert all(request["headers"]["x-api-key"] == "user-key" for request in requests) + assert all("x-cli-only" not in request["headers"] for request in requests) + assert all("authorization" not in request["headers"] for request in requests) + before = len(requests) + manager.add("thread-1", messages, user_id="alice") + assert len(requests) == before + with pytest.raises(MemoryReadError, match="owner_user_id"): + manager.search("private data", user_id="another-owner") + with pytest.raises(MemoryManagerError, match="owner_user_id"): + manager.add("thread-1", messages, user_id="another-owner") + assert len(requests) == before + finally: + manager.close() + + +def test_malformed_cli_configuration_is_not_silently_ignored(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, isolated_cli_config: Path) -> None: + isolated_cli_config.write_text("{invalid", encoding="utf-8") + with pytest.raises(ValueError, match="Invalid CLI config"): + _manager(tmp_path, monkeypatch) + + def test_manager_uses_official_recorder_retriever_and_commit_always( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, @@ -471,6 +586,18 @@ def test_context_without_thread_uses_existing_find_path( ] +def _injection_middleware(manager: OpenVikingMemoryManager) -> DynamicContextMiddleware: + from deerflow.config.app_config import AppConfig + from deerflow.config.memory_config import MemoryConfig + from deerflow.config.sandbox_config import SandboxConfig + + config = AppConfig( + sandbox=SandboxConfig(use="deerflow.sandbox.local:LocalSandboxProvider"), + memory=MemoryConfig(enabled=True, injection_enabled=True, manager_class="openviking", backend_config=manager.backend_config), + ) + return DynamicContextMiddleware(app_config=config) + + def test_unreachable_context_read_raise_aborts_dynamic_context_injection( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, @@ -491,7 +618,7 @@ def test_unreachable_context_read_raise_aborts_dynamic_context_injection( "deerflow.agents.middlewares.dynamic_context_middleware.resolve_runtime_user_id", lambda runtime: "alice", ) - middleware = DynamicContextMiddleware() + middleware = _injection_middleware(manager) state = {"messages": [HumanMessage("answer this", id="message-1")]} with pytest.raises(MemoryReadError) as exc_info: @@ -518,7 +645,7 @@ def test_strict_scope_mismatch_uses_required_read_error( "deerflow.agents.middlewares.dynamic_context_middleware.resolve_runtime_user_id", lambda runtime: "bob", ) - middleware = DynamicContextMiddleware() + middleware = _injection_middleware(manager) state = {"messages": [HumanMessage("answer this", id="message-1")]} with pytest.raises(MemoryReadError, match="owner_user_id 'alice'") as exc_info: @@ -552,7 +679,7 @@ def test_unreachable_context_read_fail_open_returns_no_injected_context( "deerflow.agents.middlewares.dynamic_context_middleware.resolve_runtime_user_id", lambda runtime: "alice", ) - middleware = DynamicContextMiddleware() + middleware = _injection_middleware(manager) state = {"messages": [HumanMessage("answer this", id="message-1")]} assert manager.get_context("alice", agent_name="research") == "" @@ -606,7 +733,7 @@ async def test_unreachable_context_read_raise_aborts_async_dynamic_context_injec "deerflow.agents.middlewares.dynamic_context_middleware.resolve_runtime_user_id", lambda runtime: "alice", ) - middleware = DynamicContextMiddleware() + middleware = _injection_middleware(manager) state = {"messages": [HumanMessage("answer this", id="message-1")]} with pytest.raises(MemoryReadError) as exc_info: diff --git a/backend/tests/test_subagent_prompt_security.py b/backend/tests/test_subagent_prompt_security.py index 770c64c007..88420fc48c 100644 --- a/backend/tests/test_subagent_prompt_security.py +++ b/backend/tests/test_subagent_prompt_security.py @@ -2,22 +2,34 @@ from types import SimpleNamespace +import pytest + from deerflow.agents.lead_agent import prompt as prompt_module +from deerflow.config.agent_storage_config import AgentStorageConfig +from deerflow.config.paths import Paths +from deerflow.config.subagents_config import SubagentsAppConfig +from deerflow.persistence.managed_subagents import file as managed_subagents_file from deerflow.subagents import registry as registry_module -def test_get_available_subagent_names_hides_bash_when_host_bash_disabled(monkeypatch) -> None: +@pytest.fixture +def builtin_subagents_config(tmp_path, monkeypatch): + monkeypatch.setattr(managed_subagents_file, "get_paths", lambda: Paths(base_dir=tmp_path)) + return SimpleNamespace(subagents=SubagentsAppConfig(), agent_storage=AgentStorageConfig(backend="file")) + + +def test_get_available_subagent_names_hides_bash_when_host_bash_disabled(monkeypatch, builtin_subagents_config) -> None: monkeypatch.setattr(registry_module, "is_host_bash_allowed", lambda: False) - names = registry_module.get_available_subagent_names() + names = registry_module.get_available_subagent_names(app_config=builtin_subagents_config) assert names == ["general-purpose"] -def test_get_available_subagent_names_keeps_bash_when_allowed(monkeypatch) -> None: +def test_get_available_subagent_names_keeps_bash_when_allowed(monkeypatch, builtin_subagents_config) -> None: monkeypatch.setattr(registry_module, "is_host_bash_allowed", lambda: True) - names = registry_module.get_available_subagent_names() + names = registry_module.get_available_subagent_names(app_config=builtin_subagents_config) assert names == ["general-purpose", "bash"]