mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 21:26:39 +00:00
f61372764e
* fix(agents): anchor agent-name validation against a trailing newline
Two copies of the shared ^[A-Za-z0-9-]+$ grammar still used re.match,
whose $ also matches before a final newline, while every other copy uses
fullmatch. POST /api/agents {"name": "reviewer\n"} therefore passed the
router guard and died inside the strict file store as an HTTP 500, and
DeerMem accepted the same name as an agents/{name}/facts directory.
* test(agents): state which trailing-newline params the regression actually covered
.match only accepted "reviewer\n" on main; "reviewer\n\n" and "reviewer \n"
were already rejected, so the docstrings should not read as if every param in
these parametrizations was broken before the fix.
989 lines
45 KiB
Python
989 lines
45 KiB
Python
"""Tests for custom agent support."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
from unittest.mock import patch
|
||
|
||
import pytest
|
||
import yaml
|
||
from fastapi.testclient import TestClient
|
||
|
||
from app.gateway.routers.agents import AGENT_NAME_PATTERN as GATEWAY_AGENT_NAME_PATTERN
|
||
from deerflow.agents.memory.backends.deermem.deermem.core.paths import AGENT_NAME_PATTERN as DEERMEM_AGENT_NAME_PATTERN
|
||
from deerflow.agents.memory.backends.deermem.deermem.core.paths import DEFAULT_AGENT_BUCKET, agent_facts_directory, validate_agent_name
|
||
from deerflow.config.agents_api_config import AgentsApiConfig, get_agents_api_config, set_agents_api_config
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_reserved_memory_bucket_stays_outside_both_public_agent_patterns() -> None:
|
||
assert GATEWAY_AGENT_NAME_PATTERN.fullmatch(DEFAULT_AGENT_BUCKET) is None
|
||
assert DEERMEM_AGENT_NAME_PATTERN.fullmatch(DEFAULT_AGENT_BUCKET) is None
|
||
validate_agent_name(DEFAULT_AGENT_BUCKET) # Internal storage sentinel remains usable.
|
||
|
||
|
||
@pytest.mark.parametrize("name", ["reviewer\n", "reviewer \n"])
|
||
def test_agent_name_validation_rejects_trailing_newline(name: str) -> None:
|
||
"""``$`` in ``^[A-Za-z0-9-]+$`` also matches before a final newline.
|
||
|
||
Only ``fullmatch`` anchors it, so DeerMem's inlined copy of the host's
|
||
agent-name grammar accepted ``"reviewer\\n"`` and used it as a directory
|
||
name — which the host's own strict validator then refuses forever.
|
||
|
||
``"reviewer \\n"`` was already rejected by ``.match`` (the space falls
|
||
outside the class, so the match never reaches ``$``); it is parametrized
|
||
here to pin the grammar, not because it regressed.
|
||
"""
|
||
with pytest.raises(ValueError, match="Invalid agent name"):
|
||
validate_agent_name(name)
|
||
with pytest.raises(ValueError, match="Invalid agent name"):
|
||
agent_facts_directory(Path("memory.json"), name)
|
||
|
||
|
||
def _make_paths(base_dir: Path):
|
||
"""Return a Paths instance pointing to base_dir."""
|
||
from deerflow.config.paths import Paths
|
||
|
||
return Paths(base_dir=base_dir)
|
||
|
||
|
||
def _write_agent(base_dir: Path, name: str, config: dict, soul: str = "You are helpful.") -> None:
|
||
"""Write an agent directory with config.yaml and SOUL.md."""
|
||
agent_dir = base_dir / "agents" / name
|
||
agent_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
config_copy = dict(config)
|
||
if "name" not in config_copy:
|
||
config_copy["name"] = name
|
||
|
||
with open(agent_dir / "config.yaml", "w") as f:
|
||
yaml.dump(config_copy, f)
|
||
|
||
(agent_dir / "SOUL.md").write_text(soul, encoding="utf-8")
|
||
|
||
|
||
# ===========================================================================
|
||
# 1. Paths class – agent path methods
|
||
# ===========================================================================
|
||
|
||
|
||
class TestPaths:
|
||
def test_agents_dir(self, tmp_path):
|
||
paths = _make_paths(tmp_path)
|
||
assert paths.agents_dir == tmp_path / "agents"
|
||
|
||
def test_agent_dir(self, tmp_path):
|
||
paths = _make_paths(tmp_path)
|
||
assert paths.agent_dir("code-reviewer") == tmp_path / "agents" / "code-reviewer"
|
||
|
||
def test_agent_memory_file(self, tmp_path):
|
||
paths = _make_paths(tmp_path)
|
||
assert paths.agent_memory_file("code-reviewer") == tmp_path / "agents" / "code-reviewer" / "memory.json"
|
||
|
||
def test_user_md_file(self, tmp_path):
|
||
paths = _make_paths(tmp_path)
|
||
assert paths.user_md_file("alice") == tmp_path / "users" / "alice" / "USER.md"
|
||
assert paths.user_md_file("bob") != paths.user_md_file("alice")
|
||
|
||
def test_paths_are_different_from_global(self, tmp_path):
|
||
paths = _make_paths(tmp_path)
|
||
assert paths.memory_file != paths.agent_memory_file("my-agent")
|
||
assert paths.memory_file == tmp_path / "memory.json"
|
||
assert paths.agent_memory_file("my-agent") == tmp_path / "agents" / "my-agent" / "memory.json"
|
||
|
||
|
||
# ===========================================================================
|
||
# 2. AgentConfig – Pydantic parsing
|
||
# ===========================================================================
|
||
|
||
|
||
class TestAgentConfig:
|
||
def test_minimal_config(self):
|
||
from deerflow.config.agents_config import AgentConfig
|
||
|
||
cfg = AgentConfig(name="my-agent")
|
||
assert cfg.name == "my-agent"
|
||
assert cfg.description == ""
|
||
assert cfg.model is None
|
||
assert cfg.tool_groups is None
|
||
|
||
def test_full_config(self):
|
||
from deerflow.config.agents_config import AgentConfig
|
||
|
||
cfg = AgentConfig(
|
||
name="code-reviewer",
|
||
description="Specialized for code review",
|
||
model="deepseek-v3",
|
||
tool_groups=["file:read", "bash"],
|
||
)
|
||
assert cfg.name == "code-reviewer"
|
||
assert cfg.model == "deepseek-v3"
|
||
assert cfg.tool_groups == ["file:read", "bash"]
|
||
|
||
def test_config_from_dict(self):
|
||
from deerflow.config.agents_config import AgentConfig
|
||
|
||
data = {"name": "test-agent", "description": "A test", "model": "gpt-4"}
|
||
cfg = AgentConfig(**data)
|
||
assert cfg.name == "test-agent"
|
||
assert cfg.model == "gpt-4"
|
||
assert cfg.tool_groups is None
|
||
|
||
|
||
# ===========================================================================
|
||
# 3. load_agent_config
|
||
# ===========================================================================
|
||
|
||
|
||
class TestLoadAgentConfig:
|
||
def test_load_valid_config(self, tmp_path):
|
||
config_dict = {"name": "code-reviewer", "description": "Code review agent", "model": "deepseek-v3"}
|
||
_write_agent(tmp_path, "code-reviewer", config_dict)
|
||
|
||
with patch("deerflow.config.agents_config.get_paths", return_value=_make_paths(tmp_path)):
|
||
from deerflow.config.agents_config import load_agent_config
|
||
|
||
cfg = load_agent_config("code-reviewer")
|
||
|
||
assert cfg.name == "code-reviewer"
|
||
assert cfg.description == "Code review agent"
|
||
assert cfg.model == "deepseek-v3"
|
||
|
||
def test_load_missing_agent_raises(self, tmp_path):
|
||
with patch("deerflow.config.agents_config.get_paths", return_value=_make_paths(tmp_path)):
|
||
from deerflow.config.agents_config import load_agent_config
|
||
|
||
with pytest.raises(FileNotFoundError):
|
||
load_agent_config("nonexistent-agent")
|
||
|
||
def test_load_missing_config_yaml_raises(self, tmp_path):
|
||
# Create directory without config.yaml
|
||
(tmp_path / "agents" / "broken-agent").mkdir(parents=True)
|
||
|
||
with patch("deerflow.config.agents_config.get_paths", return_value=_make_paths(tmp_path)):
|
||
from deerflow.config.agents_config import load_agent_config
|
||
|
||
with pytest.raises(FileNotFoundError):
|
||
load_agent_config("broken-agent")
|
||
|
||
def test_load_config_infers_name_from_dir(self, tmp_path):
|
||
"""Config without 'name' field should use directory name."""
|
||
agent_dir = tmp_path / "agents" / "inferred-name"
|
||
agent_dir.mkdir(parents=True)
|
||
(agent_dir / "config.yaml").write_text("description: My agent\n")
|
||
(agent_dir / "SOUL.md").write_text("Hello")
|
||
|
||
with patch("deerflow.config.agents_config.get_paths", return_value=_make_paths(tmp_path)):
|
||
from deerflow.config.agents_config import load_agent_config
|
||
|
||
cfg = load_agent_config("inferred-name")
|
||
|
||
assert cfg.name == "inferred-name"
|
||
|
||
def test_load_config_with_tool_groups(self, tmp_path):
|
||
config_dict = {"name": "restricted", "tool_groups": ["file:read", "file:write"]}
|
||
_write_agent(tmp_path, "restricted", config_dict)
|
||
|
||
with patch("deerflow.config.agents_config.get_paths", return_value=_make_paths(tmp_path)):
|
||
from deerflow.config.agents_config import load_agent_config
|
||
|
||
cfg = load_agent_config("restricted")
|
||
|
||
assert cfg.tool_groups == ["file:read", "file:write"]
|
||
|
||
def test_load_config_with_skills_empty_list(self, tmp_path):
|
||
config_dict = {"name": "no-skills-agent", "skills": []}
|
||
_write_agent(tmp_path, "no-skills-agent", config_dict)
|
||
|
||
with patch("deerflow.config.agents_config.get_paths", return_value=_make_paths(tmp_path)):
|
||
from deerflow.config.agents_config import load_agent_config
|
||
|
||
cfg = load_agent_config("no-skills-agent")
|
||
|
||
assert cfg.skills == []
|
||
|
||
def test_load_config_with_skills_omitted(self, tmp_path):
|
||
config_dict = {"name": "default-skills-agent"}
|
||
_write_agent(tmp_path, "default-skills-agent", config_dict)
|
||
|
||
with patch("deerflow.config.agents_config.get_paths", return_value=_make_paths(tmp_path)):
|
||
from deerflow.config.agents_config import load_agent_config
|
||
|
||
cfg = load_agent_config("default-skills-agent")
|
||
|
||
assert cfg.skills is None
|
||
|
||
def test_legacy_prompt_file_field_ignored(self, tmp_path):
|
||
"""Unknown fields like the old prompt_file should be silently ignored."""
|
||
agent_dir = tmp_path / "agents" / "legacy-agent"
|
||
agent_dir.mkdir(parents=True)
|
||
(agent_dir / "config.yaml").write_text("name: legacy-agent\nprompt_file: system.md\n")
|
||
(agent_dir / "SOUL.md").write_text("Soul content")
|
||
|
||
with patch("deerflow.config.agents_config.get_paths", return_value=_make_paths(tmp_path)):
|
||
from deerflow.config.agents_config import load_agent_config
|
||
|
||
cfg = load_agent_config("legacy-agent")
|
||
|
||
assert cfg.name == "legacy-agent"
|
||
|
||
|
||
# ===========================================================================
|
||
# 3b. resolve_agent_dir — memory-only directory fallback (#3390)
|
||
# ===========================================================================
|
||
|
||
|
||
class TestResolveAgentDirMemoryOnlyFallback:
|
||
"""Regression tests for #3390.
|
||
|
||
When memory is enabled, the first conversation creates a user-isolated
|
||
agent directory containing only ``memory.json`` (no ``config.yaml``).
|
||
On the next turn ``resolve_agent_dir`` must fall through to the legacy
|
||
shared layout instead of returning the incomplete user directory.
|
||
"""
|
||
|
||
def test_user_dir_with_only_memory_falls_back_to_legacy(self, tmp_path):
|
||
"""User dir has memory.json but no config.yaml → use legacy dir."""
|
||
from deerflow.config.agents_config import resolve_agent_dir
|
||
|
||
# Legacy agent with full config
|
||
legacy_dir = tmp_path / "agents" / "my-agent"
|
||
legacy_dir.mkdir(parents=True)
|
||
(legacy_dir / "config.yaml").write_text("name: my-agent\n", encoding="utf-8")
|
||
(legacy_dir / "SOUL.md").write_text("legacy soul", encoding="utf-8")
|
||
|
||
# User dir created by memory write — no config.yaml
|
||
user_dir = tmp_path / "users" / "u1" / "agents" / "my-agent"
|
||
user_dir.mkdir(parents=True)
|
||
(user_dir / "memory.json").write_text("{}", encoding="utf-8")
|
||
|
||
with patch("deerflow.config.agents_config.get_paths", return_value=_make_paths(tmp_path)), patch("deerflow.config.agents_config.get_effective_user_id", return_value="u1"):
|
||
result = resolve_agent_dir("my-agent", user_id="u1")
|
||
|
||
assert result == legacy_dir
|
||
|
||
def test_user_dir_with_config_takes_priority(self, tmp_path):
|
||
"""User dir with config.yaml should still win over legacy."""
|
||
from deerflow.config.agents_config import resolve_agent_dir
|
||
|
||
# Legacy
|
||
legacy_dir = tmp_path / "agents" / "my-agent"
|
||
legacy_dir.mkdir(parents=True)
|
||
(legacy_dir / "config.yaml").write_text("name: my-agent\n", encoding="utf-8")
|
||
|
||
# User dir with full config (migrated)
|
||
user_dir = tmp_path / "users" / "u1" / "agents" / "my-agent"
|
||
user_dir.mkdir(parents=True)
|
||
(user_dir / "config.yaml").write_text("name: my-agent\nmodel: gpt-4\n", encoding="utf-8")
|
||
(user_dir / "memory.json").write_text("{}", encoding="utf-8")
|
||
|
||
with patch("deerflow.config.agents_config.get_paths", return_value=_make_paths(tmp_path)), patch("deerflow.config.agents_config.get_effective_user_id", return_value="u1"):
|
||
result = resolve_agent_dir("my-agent", user_id="u1")
|
||
|
||
assert result == user_dir
|
||
|
||
def test_load_config_falls_back_when_user_dir_is_memory_only(self, tmp_path):
|
||
"""End-to-end: load_agent_config works when user dir only has memory.json."""
|
||
config_dict = {"name": "my-agent", "description": "Legacy agent", "model": "deepseek-v3"}
|
||
_write_agent(tmp_path, "my-agent", config_dict)
|
||
|
||
# Simulate memory write creating user dir without config
|
||
user_dir = tmp_path / "users" / "u1" / "agents" / "my-agent"
|
||
user_dir.mkdir(parents=True)
|
||
(user_dir / "memory.json").write_text("{}", encoding="utf-8")
|
||
|
||
with patch("deerflow.config.agents_config.get_paths", return_value=_make_paths(tmp_path)), patch("deerflow.config.agents_config.get_effective_user_id", return_value="u1"):
|
||
from deerflow.config.agents_config import load_agent_config
|
||
|
||
cfg = load_agent_config("my-agent", user_id="u1")
|
||
|
||
assert cfg.name == "my-agent"
|
||
assert cfg.model == "deepseek-v3"
|
||
|
||
|
||
# ===========================================================================
|
||
# 4. load_agent_soul
|
||
# ===========================================================================
|
||
|
||
|
||
class TestLoadAgentSoul:
|
||
def test_reads_soul_file(self, tmp_path):
|
||
expected_soul = "You are a specialized code review expert."
|
||
_write_agent(tmp_path, "code-reviewer", {"name": "code-reviewer"}, soul=expected_soul)
|
||
|
||
with patch("deerflow.config.agents_config.get_paths", return_value=_make_paths(tmp_path)):
|
||
from deerflow.config.agents_config import AgentConfig, load_agent_soul
|
||
|
||
cfg = AgentConfig(name="code-reviewer")
|
||
soul = load_agent_soul(cfg.name)
|
||
|
||
assert soul == expected_soul
|
||
|
||
def test_missing_soul_file_returns_none(self, tmp_path):
|
||
agent_dir = tmp_path / "agents" / "no-soul"
|
||
agent_dir.mkdir(parents=True)
|
||
(agent_dir / "config.yaml").write_text("name: no-soul\n")
|
||
# No SOUL.md created
|
||
|
||
with patch("deerflow.config.agents_config.get_paths", return_value=_make_paths(tmp_path)):
|
||
from deerflow.config.agents_config import AgentConfig, load_agent_soul
|
||
|
||
cfg = AgentConfig(name="no-soul")
|
||
soul = load_agent_soul(cfg.name)
|
||
|
||
assert soul is None
|
||
|
||
def test_empty_soul_file_returns_none(self, tmp_path):
|
||
agent_dir = tmp_path / "agents" / "empty-soul"
|
||
agent_dir.mkdir(parents=True)
|
||
(agent_dir / "config.yaml").write_text("name: empty-soul\n")
|
||
(agent_dir / "SOUL.md").write_text(" \n ")
|
||
|
||
with patch("deerflow.config.agents_config.get_paths", return_value=_make_paths(tmp_path)):
|
||
from deerflow.config.agents_config import AgentConfig, load_agent_soul
|
||
|
||
cfg = AgentConfig(name="empty-soul")
|
||
soul = load_agent_soul(cfg.name)
|
||
|
||
assert soul is None
|
||
|
||
def test_loads_soul_without_config_yaml(self, tmp_path):
|
||
"""SOUL.md should load even when the agent dir has no config.yaml (#4135)."""
|
||
agent_dir = tmp_path / "agents" / "soul-only"
|
||
agent_dir.mkdir(parents=True)
|
||
# Deliberately no config.yaml – the agent is configured externally
|
||
(agent_dir / "SOUL.md").write_text("You are a brave agent.", encoding="utf-8")
|
||
|
||
with patch("deerflow.config.agents_config.get_paths", return_value=_make_paths(tmp_path)), patch("deerflow.config.agents_config.get_effective_user_id", return_value="default"):
|
||
from deerflow.config.agents_config import load_agent_soul
|
||
|
||
soul = load_agent_soul("soul-only")
|
||
|
||
assert soul == "You are a brave agent."
|
||
|
||
def test_loads_soul_from_user_dir_without_config_yaml(self, tmp_path):
|
||
"""Fallback should find SOUL.md when resolver returns a default dir without it (#4135).
|
||
|
||
Setup: per-user agent 'foo' exists as a memory-only directory
|
||
(no config.yaml, no SOUL.md). Legacy agent 'foo' has SOUL.md but
|
||
no config.yaml. resolve_agent_dir returns the per-user path as
|
||
default (neither dir has config.yaml). The fallback then finds
|
||
SOUL.md in the legacy directory.
|
||
"""
|
||
# Per-user dir: memory-only (no config.yaml, no SOUL.md)
|
||
user_dir = tmp_path / "users" / "test-user" / "agents" / "foo"
|
||
user_dir.mkdir(parents=True)
|
||
(user_dir / "memory.json").write_text("{}", encoding="utf-8")
|
||
|
||
# Legacy dir: has SOUL.md but no config.yaml
|
||
legacy_dir = tmp_path / "agents" / "foo"
|
||
legacy_dir.mkdir(parents=True)
|
||
(legacy_dir / "SOUL.md").write_text("You are a legacy agent.", encoding="utf-8")
|
||
|
||
with patch("deerflow.config.agents_config.get_paths", return_value=_make_paths(tmp_path)), patch("deerflow.config.agents_config.get_effective_user_id", return_value="test-user"):
|
||
from deerflow.config.agents_config import load_agent_soul
|
||
|
||
soul = load_agent_soul("foo")
|
||
|
||
assert soul == "You are a legacy agent."
|
||
|
||
def test_soul_not_leaked_from_legacy_when_per_user_has_config(self, tmp_path):
|
||
"""Per-user agent with config.yaml but no SOUL.md should NOT fall back to legacy SOUL.md.
|
||
|
||
This verifies the gated condition: fallback only fires when the
|
||
resolved dir lacks config.yaml. A properly-resolved per-user agent
|
||
that simply has no SOUL.md returns None, preserving the
|
||
"per-user entries fully shadow legacy entries" invariant.
|
||
"""
|
||
# Legacy dir: has SOUL.md
|
||
legacy_dir = tmp_path / "agents" / "foo"
|
||
legacy_dir.mkdir(parents=True)
|
||
(legacy_dir / "config.yaml").write_text("name: foo\n")
|
||
(legacy_dir / "SOUL.md").write_text("You are a legacy agent.", encoding="utf-8")
|
||
|
||
# Per-user dir: has config.yaml (resolver returns this) but no SOUL.md
|
||
user_dir = tmp_path / "users" / "test-user" / "agents" / "foo"
|
||
user_dir.mkdir(parents=True)
|
||
(user_dir / "config.yaml").write_text("name: foo\n")
|
||
# No SOUL.md in per-user dir
|
||
|
||
with patch("deerflow.config.agents_config.get_paths", return_value=_make_paths(tmp_path)), patch("deerflow.config.agents_config.get_effective_user_id", return_value="test-user"):
|
||
from deerflow.config.agents_config import load_agent_soul
|
||
|
||
soul = load_agent_soul("foo")
|
||
|
||
assert soul is None
|
||
|
||
|
||
# ===========================================================================
|
||
# 5. list_custom_agents
|
||
# ===========================================================================
|
||
|
||
|
||
class TestListCustomAgents:
|
||
def test_empty_when_no_agents_dir(self, tmp_path):
|
||
with patch("deerflow.config.agents_config.get_paths", return_value=_make_paths(tmp_path)):
|
||
from deerflow.config.agents_config import list_custom_agents
|
||
|
||
agents = list_custom_agents()
|
||
|
||
assert agents == []
|
||
|
||
def test_discovers_multiple_agents(self, tmp_path):
|
||
_write_agent(tmp_path, "agent-a", {"name": "agent-a"})
|
||
_write_agent(tmp_path, "agent-b", {"name": "agent-b", "description": "B"})
|
||
|
||
with patch("deerflow.config.agents_config.get_paths", return_value=_make_paths(tmp_path)):
|
||
from deerflow.config.agents_config import list_custom_agents
|
||
|
||
agents = list_custom_agents()
|
||
|
||
names = [a.name for a in agents]
|
||
assert "agent-a" in names
|
||
assert "agent-b" in names
|
||
|
||
def test_skips_dirs_without_config_yaml(self, tmp_path):
|
||
# Valid agent
|
||
_write_agent(tmp_path, "valid-agent", {"name": "valid-agent"})
|
||
# Invalid dir (no config.yaml)
|
||
(tmp_path / "agents" / "invalid-dir").mkdir(parents=True)
|
||
|
||
with patch("deerflow.config.agents_config.get_paths", return_value=_make_paths(tmp_path)):
|
||
from deerflow.config.agents_config import list_custom_agents
|
||
|
||
agents = list_custom_agents()
|
||
|
||
assert len(agents) == 1
|
||
assert agents[0].name == "valid-agent"
|
||
|
||
def test_skips_non_directory_entries(self, tmp_path):
|
||
# Create the agents dir with a file (not a dir)
|
||
agents_dir = tmp_path / "agents"
|
||
agents_dir.mkdir(parents=True)
|
||
(agents_dir / "not-a-dir.txt").write_text("hello")
|
||
_write_agent(tmp_path, "real-agent", {"name": "real-agent"})
|
||
|
||
with patch("deerflow.config.agents_config.get_paths", return_value=_make_paths(tmp_path)):
|
||
from deerflow.config.agents_config import list_custom_agents
|
||
|
||
agents = list_custom_agents()
|
||
|
||
assert len(agents) == 1
|
||
assert agents[0].name == "real-agent"
|
||
|
||
def test_returns_sorted_by_name(self, tmp_path):
|
||
_write_agent(tmp_path, "z-agent", {"name": "z-agent"})
|
||
_write_agent(tmp_path, "a-agent", {"name": "a-agent"})
|
||
_write_agent(tmp_path, "m-agent", {"name": "m-agent"})
|
||
|
||
with patch("deerflow.config.agents_config.get_paths", return_value=_make_paths(tmp_path)):
|
||
from deerflow.config.agents_config import list_custom_agents
|
||
|
||
agents = list_custom_agents()
|
||
|
||
names = [a.name for a in agents]
|
||
assert names == sorted(names)
|
||
|
||
|
||
# ===========================================================================
|
||
# 7. Memory isolation: _get_memory_file_path
|
||
# ===========================================================================
|
||
|
||
|
||
class TestMemoryFilePath:
|
||
def test_global_memory_path(self, tmp_path, monkeypatch):
|
||
"""None agent_name should return global memory file."""
|
||
from deerflow.agents.memory.backends.deermem.deermem.config import DeerMemConfig
|
||
from deerflow.agents.memory.backends.deermem.deermem.core.storage import FileMemoryStorage
|
||
|
||
monkeypatch.setenv("DEERMEM_DATA_DIR", str(tmp_path))
|
||
storage = FileMemoryStorage(DeerMemConfig())
|
||
path = storage._get_memory_file_path(None)
|
||
assert path == tmp_path / "memory.json"
|
||
|
||
def test_agent_memory_path(self, tmp_path, monkeypatch):
|
||
"""All agents share the user-global summary JSON path."""
|
||
from deerflow.agents.memory.backends.deermem.deermem.config import DeerMemConfig
|
||
from deerflow.agents.memory.backends.deermem.deermem.core.storage import FileMemoryStorage
|
||
|
||
monkeypatch.setenv("DEERMEM_DATA_DIR", str(tmp_path))
|
||
storage = FileMemoryStorage(DeerMemConfig())
|
||
path = storage._get_memory_file_path("code-reviewer")
|
||
assert path == tmp_path / "memory.json"
|
||
|
||
def test_agents_share_summary_path(self, tmp_path, monkeypatch):
|
||
from deerflow.agents.memory.backends.deermem.deermem.config import DeerMemConfig
|
||
from deerflow.agents.memory.backends.deermem.deermem.core.storage import FileMemoryStorage
|
||
|
||
monkeypatch.setenv("DEERMEM_DATA_DIR", str(tmp_path))
|
||
storage = FileMemoryStorage(DeerMemConfig())
|
||
path_global = storage._get_memory_file_path(None)
|
||
path_a = storage._get_memory_file_path("agent-a")
|
||
path_b = storage._get_memory_file_path("agent-b")
|
||
|
||
assert path_global == path_a == path_b
|
||
|
||
|
||
# ===========================================================================
|
||
# 8. Gateway API – Agents endpoints
|
||
# ===========================================================================
|
||
|
||
|
||
# Model names the agents API tests may send in a create/update payload. The
|
||
# router validates `model` against the app config, so the fixture pins a stub
|
||
# config exposing exactly these instead of leaving the assertion at the mercy
|
||
# of whatever `config.yaml` happens to sit in the repo root: CI has none (the
|
||
# validation is skipped and the request passes), a real dev checkout does (an
|
||
# unlisted model name yields 422 and the test fails).
|
||
_KNOWN_TEST_MODELS = frozenset({"deepseek-v3"})
|
||
|
||
|
||
def _stub_app_config():
|
||
"""App config that knows only ``_KNOWN_TEST_MODELS``."""
|
||
return SimpleNamespace(get_model_config=lambda name: SimpleNamespace(name=name) if name in _KNOWN_TEST_MODELS else None)
|
||
|
||
|
||
def _make_test_app(tmp_path: Path):
|
||
"""Create a FastAPI app with the agents router, patching paths to tmp_path.
|
||
|
||
Uses the stub-auth helper so the ``@require_permission`` decorators on the
|
||
agents routes see an authenticated user with all permissions (mirroring
|
||
what ``AuthMiddleware`` does in the real gateway).
|
||
"""
|
||
from _router_auth_helpers import make_authed_test_app
|
||
|
||
from app.gateway.routers.agents import router
|
||
|
||
app = make_authed_test_app()
|
||
app.include_router(router)
|
||
return app
|
||
|
||
|
||
@pytest.fixture()
|
||
def agent_client(tmp_path):
|
||
"""TestClient with agents router, using tmp_path as base_dir."""
|
||
import app.gateway.routers.agents as agents_router
|
||
|
||
paths_instance = _make_paths(tmp_path)
|
||
previous_config = AgentsApiConfig(**get_agents_api_config().model_dump())
|
||
|
||
with (
|
||
patch("deerflow.config.agents_config.get_paths", return_value=paths_instance),
|
||
patch.object(agents_router, "get_paths", return_value=paths_instance),
|
||
patch.object(agents_router, "get_app_config", _stub_app_config),
|
||
):
|
||
set_agents_api_config(AgentsApiConfig(enabled=True))
|
||
try:
|
||
app = _make_test_app(tmp_path)
|
||
with TestClient(app) as client:
|
||
client._tmp_path = tmp_path # type: ignore[attr-defined]
|
||
yield client
|
||
finally:
|
||
set_agents_api_config(previous_config)
|
||
|
||
|
||
@pytest.fixture()
|
||
def disabled_agent_client(tmp_path):
|
||
"""TestClient with agents router while the management API is disabled."""
|
||
import app.gateway.routers.agents as agents_router
|
||
|
||
paths_instance = _make_paths(tmp_path)
|
||
previous_config = AgentsApiConfig(**get_agents_api_config().model_dump())
|
||
|
||
with patch("deerflow.config.agents_config.get_paths", return_value=paths_instance), patch.object(agents_router, "get_paths", return_value=paths_instance):
|
||
set_agents_api_config(AgentsApiConfig(enabled=False))
|
||
try:
|
||
app = _make_test_app(tmp_path)
|
||
with TestClient(app) as client:
|
||
yield client
|
||
finally:
|
||
set_agents_api_config(previous_config)
|
||
|
||
|
||
class TestAgentsAPI:
|
||
@pytest.mark.parametrize("display_name", ["x" * 150, 123, "\u200b" * 3])
|
||
def test_invalid_stored_display_name_falls_back_in_api(self, agent_client, display_name):
|
||
from deerflow.persistence.agents.file import FileAgentStore
|
||
|
||
FileAgentStore().create("reviewer", {"display_name": display_name, "description": "healthy"}, "Soul")
|
||
response = agent_client.get("/api/agents/reviewer")
|
||
assert response.status_code == 200
|
||
assert response.json()["display_name"] is None
|
||
assert response.json()["description"] == "healthy"
|
||
assert agent_client.get("/api/agents").json()["agents"][0]["name"] == "reviewer"
|
||
response = agent_client.put("/api/agents/reviewer", json={"display_name": "已修复"})
|
||
assert response.status_code == 200
|
||
assert response.json()["display_name"] == "已修复"
|
||
|
||
@pytest.mark.parametrize("display_name", ["a\u202eb", "line1\nline2", "z\x00ero", "\u200b" * 3, "a\u200fb", "a\u2028b", "\u200c\u200d"])
|
||
def test_invalid_display_name_cannot_be_persisted(self, agent_client, display_name):
|
||
assert agent_client.post("/api/agents", json={"name": "reviewer", "display_name": display_name}).status_code == 422
|
||
assert agent_client.get("/api/agents").json()["agents"] == []
|
||
assert agent_client.post("/api/agents", json={"name": "reviewer", "display_name": "🦌" * 100}).status_code == 201
|
||
assert agent_client.put("/api/agents/reviewer", json={"display_name": display_name}).status_code == 422
|
||
assert agent_client.get("/api/agents/reviewer").json()["display_name"] == "🦌" * 100
|
||
|
||
@pytest.mark.parametrize("name", ["reviewer\n", "reviewer\n\n"])
|
||
def test_trailing_newline_in_agent_name_is_rejected(self, agent_client, name):
|
||
"""The router's ``AGENT_NAME_PATTERN.match`` accepted ``"reviewer\\n"`` and the store 500'd.
|
||
|
||
``$`` matches before a single trailing newline, so that one param reached
|
||
the file store, which validates the same grammar with ``fullmatch``.
|
||
``"reviewer\\n\\n"`` was already rejected by ``.match`` on main; it is
|
||
parametrized to pin the grammar, not because it regressed.
|
||
"""
|
||
assert agent_client.post("/api/agents", json={"name": name}).status_code == 422
|
||
assert agent_client.get("/api/agents").json()["agents"] == []
|
||
|
||
def test_display_name_round_trip_keeps_stable_identity(self, agent_client):
|
||
response = agent_client.post("/api/agents", json={"name": "code-reviewer", "display_name": " 代码审查助手 "})
|
||
assert response.status_code == 201
|
||
assert response.json()["display_name"] == "代码审查助手"
|
||
assert response.json()["name"] == "code-reviewer"
|
||
assert agent_client.get("/api/agents").json()["agents"][0]["display_name"] == "代码审查助手"
|
||
response = agent_client.put("/api/agents/code-reviewer", json={"description": "Updated"})
|
||
assert response.json()["display_name"] == "代码审查助手"
|
||
response = agent_client.put("/api/agents/code-reviewer", json={"display_name": "审查员 🦌"})
|
||
assert response.json()["display_name"] == "审查员 🦌"
|
||
assert agent_client.get("/api/agents/code-reviewer").json()["display_name"] == "审查员 🦌"
|
||
response = agent_client.put("/api/agents/code-reviewer", json={"display_name": None})
|
||
assert response.json()["display_name"] is None
|
||
assert response.json()["name"] == "code-reviewer"
|
||
|
||
def test_display_name_validation_does_not_relax_agent_identifier(self, agent_client):
|
||
assert agent_client.post("/api/agents", json={"name": "中文"}).status_code == 422
|
||
assert agent_client.post("/api/agents", json={"name": "reviewer", "display_name": "名" * 101}).status_code == 422
|
||
assert agent_client.post("/api/agents", json={"name": "reviewer", "display_name": "名" * 100}).status_code == 201
|
||
assert agent_client.put("/api/agents/reviewer", json={"display_name": "名" * 101}).status_code == 422
|
||
assert agent_client.get("/api/agents/reviewer").json()["display_name"] == "名" * 100
|
||
|
||
def test_list_agents_empty(self, agent_client):
|
||
response = agent_client.get("/api/agents")
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert data["agents"] == []
|
||
|
||
def test_create_agent(self, agent_client):
|
||
payload = {
|
||
"name": "code-reviewer",
|
||
"description": "Reviews code",
|
||
"soul": "You are a code reviewer.",
|
||
}
|
||
response = agent_client.post("/api/agents", json=payload)
|
||
assert response.status_code == 201
|
||
data = response.json()
|
||
assert data["name"] == "code-reviewer"
|
||
assert data["description"] == "Reviews code"
|
||
assert data["soul"] == "You are a code reviewer."
|
||
|
||
def test_create_agent_invalid_name(self, agent_client):
|
||
payload = {"name": "Code Reviewer!", "soul": "test"}
|
||
response = agent_client.post("/api/agents", json=payload)
|
||
assert response.status_code == 422
|
||
|
||
def test_create_duplicate_agent_409(self, agent_client):
|
||
payload = {"name": "my-agent", "soul": "test"}
|
||
agent_client.post("/api/agents", json=payload)
|
||
|
||
# Second create should fail
|
||
response = agent_client.post("/api/agents", json=payload)
|
||
assert response.status_code == 409
|
||
|
||
def test_list_agents_after_create(self, agent_client):
|
||
agent_client.post("/api/agents", json={"name": "agent-one", "soul": "p1"})
|
||
agent_client.post("/api/agents", json={"name": "agent-two", "soul": "p2"})
|
||
|
||
response = agent_client.get("/api/agents")
|
||
assert response.status_code == 200
|
||
names = [a["name"] for a in response.json()["agents"]]
|
||
assert "agent-one" in names
|
||
assert "agent-two" in names
|
||
|
||
def test_list_agents_includes_soul(self, agent_client):
|
||
agent_client.post("/api/agents", json={"name": "soul-agent", "soul": "My soul content"})
|
||
|
||
response = agent_client.get("/api/agents")
|
||
assert response.status_code == 200
|
||
agents = response.json()["agents"]
|
||
soul_agent = next(a for a in agents if a["name"] == "soul-agent")
|
||
assert soul_agent["soul"] == "My soul content"
|
||
|
||
def test_get_agent(self, agent_client):
|
||
agent_client.post("/api/agents", json={"name": "test-agent", "soul": "Hello world"})
|
||
|
||
response = agent_client.get("/api/agents/test-agent")
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert data["name"] == "test-agent"
|
||
assert data["soul"] == "Hello world"
|
||
|
||
def test_get_missing_agent_404(self, agent_client):
|
||
response = agent_client.get("/api/agents/nonexistent")
|
||
assert response.status_code == 404
|
||
|
||
def test_update_agent_soul(self, agent_client):
|
||
agent_client.post("/api/agents", json={"name": "update-me", "soul": "original"})
|
||
|
||
response = agent_client.put("/api/agents/update-me", json={"soul": "updated"})
|
||
assert response.status_code == 200
|
||
assert response.json()["soul"] == "updated"
|
||
|
||
def test_update_agent_description(self, agent_client):
|
||
agent_client.post("/api/agents", json={"name": "desc-agent", "description": "old desc", "soul": "p"})
|
||
|
||
response = agent_client.put("/api/agents/desc-agent", json={"description": "new desc"})
|
||
assert response.status_code == 200
|
||
assert response.json()["description"] == "new desc"
|
||
|
||
def test_update_agent_preserves_hand_authored_non_managed_fields(self, agent_client):
|
||
"""Hand-authored ``github:`` and ``memory_enabled`` fields must survive PATCH.
|
||
|
||
The HTTP route does not expose ``github`` as an editable field
|
||
(and rightly so — the GitHub App credentials and binding triggers
|
||
are operator-authored, not end-user-editable). But it MUST carry
|
||
the block forward when rewriting ``config.yaml`` for a description /
|
||
model / tool_groups / skills change, otherwise an operator who
|
||
edits the agent's description from the Web UI silently strips the
|
||
binding and the next webhook delivery silently no-ops.
|
||
|
||
Mirrors the same property the harness ``update_agent`` tool enforces
|
||
via ``preserve_non_managed_fields``; both surfaces share the helper.
|
||
"""
|
||
# Create an agent through the API, then hand-author a github: block
|
||
# into its config.yaml — exactly the workflow an operator would
|
||
# follow when wiring a new repo binding.
|
||
agent_client.post("/api/agents", json={"name": "github-agent", "description": "old desc", "soul": "p"})
|
||
|
||
tmp_path: Path = agent_client._tmp_path # type: ignore[attr-defined]
|
||
agent_dir = tmp_path / "users" / "test-user-autouse" / "agents" / "github-agent"
|
||
config_file = agent_dir / "config.yaml"
|
||
config_data = yaml.safe_load(config_file.read_text())
|
||
config_data["github"] = {
|
||
"installation_id": 99999,
|
||
"bot_login": "github-agent-bot",
|
||
"bindings": [
|
||
{
|
||
"repo": "acme/widget",
|
||
"triggers": {"pull_request": {"actions": ["opened"]}},
|
||
}
|
||
],
|
||
}
|
||
config_data["memory_enabled"] = False
|
||
config_file.write_text(yaml.safe_dump(config_data, sort_keys=False), encoding="utf-8")
|
||
|
||
# PATCH only the description.
|
||
response = agent_client.put("/api/agents/github-agent", json={"description": "new desc"})
|
||
assert response.status_code == 200
|
||
|
||
# github: block must survive verbatim.
|
||
reloaded = yaml.safe_load(config_file.read_text())
|
||
assert reloaded["description"] == "new desc"
|
||
assert reloaded["memory_enabled"] is False
|
||
assert reloaded["github"] == {
|
||
"installation_id": 99999,
|
||
"bot_login": "github-agent-bot",
|
||
"bindings": [
|
||
{
|
||
"repo": "acme/widget",
|
||
"triggers": {"pull_request": {"actions": ["opened"]}},
|
||
}
|
||
],
|
||
}
|
||
|
||
def test_update_memory_only_user_dir_with_legacy_agent_returns_409(self, agent_client, tmp_path):
|
||
"""Regression for #3390's PUT /api/agents/{name} guard.
|
||
|
||
A per-user agent directory can exist containing only memory.json
|
||
(written the first time this user chats with a legacy shared
|
||
agent). The stale guard checked bare directory existence and
|
||
missed this case, letting the route silently fork a brand-new
|
||
config.yaml/SOUL.md into the memory-only directory instead of
|
||
blocking with the migration-script guidance.
|
||
"""
|
||
legacy_dir = tmp_path / "agents" / "legacy-agent"
|
||
legacy_dir.mkdir(parents=True)
|
||
(legacy_dir / "config.yaml").write_text("name: legacy-agent\ndescription: legacy\n", encoding="utf-8")
|
||
(legacy_dir / "SOUL.md").write_text("legacy soul", encoding="utf-8")
|
||
|
||
user_agent_dir = tmp_path / "users" / "test-user-autouse" / "agents" / "legacy-agent"
|
||
user_agent_dir.mkdir(parents=True)
|
||
(user_agent_dir / "memory.json").write_text("{}", encoding="utf-8")
|
||
|
||
response = agent_client.put("/api/agents/legacy-agent", json={"soul": "should not write"})
|
||
|
||
assert response.status_code == 409
|
||
assert not (user_agent_dir / "config.yaml").exists()
|
||
assert not (user_agent_dir / "SOUL.md").exists()
|
||
assert (user_agent_dir / "memory.json").exists(), "the user's existing memory must be left untouched"
|
||
|
||
def test_update_missing_agent_404(self, agent_client):
|
||
response = agent_client.put("/api/agents/ghost-agent", json={"soul": "new"})
|
||
assert response.status_code == 404
|
||
|
||
def test_delete_agent(self, agent_client):
|
||
agent_client.post("/api/agents", json={"name": "del-me", "soul": "bye"})
|
||
|
||
response = agent_client.delete("/api/agents/del-me")
|
||
assert response.status_code == 204
|
||
|
||
# Verify it's gone
|
||
response = agent_client.get("/api/agents/del-me")
|
||
assert response.status_code == 404
|
||
|
||
def test_delete_missing_agent_404(self, agent_client):
|
||
response = agent_client.delete("/api/agents/does-not-exist")
|
||
assert response.status_code == 404
|
||
|
||
def test_delete_rejects_memory_only_directory_without_removing_facts(self, agent_client, tmp_path):
|
||
agent_dir = tmp_path / "users" / "test-user-autouse" / "agents" / "lead-agent"
|
||
facts_dir = agent_dir / "facts"
|
||
facts_dir.mkdir(parents=True)
|
||
fact_path = facts_dir / "fact_keep.md"
|
||
fact_path.write_text("memory data", encoding="utf-8")
|
||
|
||
response = agent_client.delete("/api/agents/lead-agent")
|
||
|
||
assert response.status_code == 409
|
||
assert fact_path.read_text(encoding="utf-8") == "memory data"
|
||
|
||
def test_reserved_default_bucket_cannot_be_created_as_custom_agent(self, agent_client):
|
||
response = agent_client.post("/api/agents", json={"name": "__default__", "soul": "must fail"})
|
||
assert response.status_code == 422
|
||
|
||
def test_create_agent_with_model_and_tool_groups(self, agent_client):
|
||
payload = {
|
||
"name": "specialized",
|
||
"description": "Specialized agent",
|
||
"model": "deepseek-v3",
|
||
"tool_groups": ["file:read", "bash"],
|
||
"soul": "You are specialized.",
|
||
}
|
||
response = agent_client.post("/api/agents", json=payload)
|
||
assert response.status_code == 201
|
||
data = response.json()
|
||
assert data["model"] == "deepseek-v3"
|
||
assert data["tool_groups"] == ["file:read", "bash"]
|
||
|
||
def test_create_persists_files_on_disk(self, agent_client, tmp_path):
|
||
agent_client.post("/api/agents", json={"name": "disk-check", "soul": "disk soul"})
|
||
|
||
# tests/conftest.py installs an autouse fixture that sets the
|
||
# contextvar to "test-user-autouse", so the agent is persisted under
|
||
# users/test-user-autouse/agents/ rather than the legacy shared dir.
|
||
agent_dir = tmp_path / "users" / "test-user-autouse" / "agents" / "disk-check"
|
||
assert agent_dir.exists()
|
||
assert (agent_dir / "config.yaml").exists()
|
||
assert (agent_dir / "SOUL.md").exists()
|
||
assert (agent_dir / "SOUL.md").read_text() == "disk soul"
|
||
|
||
def test_delete_removes_files_from_disk(self, agent_client, tmp_path):
|
||
agent_client.post("/api/agents", json={"name": "remove-me", "soul": "bye"})
|
||
agent_dir = tmp_path / "users" / "test-user-autouse" / "agents" / "remove-me"
|
||
assert agent_dir.exists()
|
||
|
||
agent_client.delete("/api/agents/remove-me")
|
||
assert not agent_dir.exists()
|
||
|
||
def test_create_rejects_legacy_name_collision(self, agent_client, tmp_path):
|
||
"""An unmigrated legacy agent must still block name collision so that
|
||
running the migration script later won't shadow the legacy entry."""
|
||
legacy_dir = tmp_path / "agents" / "legacy-agent"
|
||
legacy_dir.mkdir(parents=True)
|
||
(legacy_dir / "config.yaml").write_text("name: legacy-agent\n", encoding="utf-8")
|
||
(legacy_dir / "SOUL.md").write_text("legacy soul", encoding="utf-8")
|
||
|
||
response = agent_client.post("/api/agents", json={"name": "legacy-agent", "soul": "x"})
|
||
assert response.status_code == 409
|
||
|
||
|
||
# ===========================================================================
|
||
# 9. Gateway API – User Profile endpoints
|
||
# ===========================================================================
|
||
|
||
|
||
class TestUserProfileAPI:
|
||
def test_get_user_profile_empty(self, agent_client):
|
||
response = agent_client.get("/api/user-profile")
|
||
assert response.status_code == 200
|
||
assert response.json()["content"] is None
|
||
|
||
def test_put_user_profile(self, agent_client, tmp_path):
|
||
content = "# User Profile\n\nI am a developer."
|
||
response = agent_client.put("/api/user-profile", json={"content": content})
|
||
assert response.status_code == 200
|
||
assert response.json()["content"] == content
|
||
|
||
# File should be written to the caller's per-user bucket. The autouse
|
||
# _auto_user_context fixture in conftest.py sets user
|
||
# "test-user-autouse", so that is the effective id here.
|
||
user_md = tmp_path / "users" / "test-user-autouse" / "USER.md"
|
||
assert user_md.exists()
|
||
assert user_md.read_text(encoding="utf-8") == content
|
||
|
||
def test_user_profile_is_isolated_per_user(self, agent_client, tmp_path):
|
||
"""A legacy global USER.md must never leak into a user's profile read.
|
||
|
||
Pre-fix behavior: GET/PUT /api/user-profile read and wrote the shared
|
||
``{base_dir}/USER.md`` singleton, so any authenticated user could
|
||
overwrite the prompt context injected for every other user.
|
||
"""
|
||
legacy_global = tmp_path / "USER.md"
|
||
legacy_global.write_text("# injected by another user", encoding="utf-8")
|
||
|
||
got = agent_client.get("/api/user-profile")
|
||
assert got.status_code == 200
|
||
# Per-user file does not exist yet and the legacy global file is not
|
||
# consulted as a fallback.
|
||
assert got.json()["content"] is None
|
||
|
||
def test_get_user_profile_after_put(self, agent_client):
|
||
content = "# Profile\n\nI work on data science."
|
||
agent_client.put("/api/user-profile", json={"content": content})
|
||
|
||
response = agent_client.get("/api/user-profile")
|
||
assert response.status_code == 200
|
||
assert response.json()["content"] == content
|
||
|
||
def test_put_empty_user_profile_returns_none(self, agent_client):
|
||
response = agent_client.put("/api/user-profile", json={"content": ""})
|
||
assert response.status_code == 200
|
||
assert response.json()["content"] is None
|
||
|
||
|
||
class TestAgentsApiDisabled:
|
||
def test_agents_list_returns_403(self, disabled_agent_client):
|
||
response = disabled_agent_client.get("/api/agents")
|
||
assert response.status_code == 403
|
||
assert "agents_api.enabled=true" in response.json()["detail"]
|
||
|
||
def test_agent_get_returns_403(self, disabled_agent_client):
|
||
response = disabled_agent_client.get("/api/agents/example-agent")
|
||
assert response.status_code == 403
|
||
|
||
def test_agent_name_check_returns_403(self, disabled_agent_client):
|
||
response = disabled_agent_client.get("/api/agents/check", params={"name": "example-agent"})
|
||
assert response.status_code == 403
|
||
|
||
def test_agent_create_returns_403(self, disabled_agent_client):
|
||
response = disabled_agent_client.post("/api/agents", json={"name": "example-agent", "soul": "blocked"})
|
||
assert response.status_code == 403
|
||
|
||
def test_agent_update_returns_403(self, disabled_agent_client):
|
||
response = disabled_agent_client.put("/api/agents/example-agent", json={"description": "blocked"})
|
||
assert response.status_code == 403
|
||
|
||
def test_agent_delete_returns_403(self, disabled_agent_client):
|
||
response = disabled_agent_client.delete("/api/agents/example-agent")
|
||
assert response.status_code == 403
|
||
|
||
def test_user_profile_routes_return_403(self, disabled_agent_client):
|
||
get_response = disabled_agent_client.get("/api/user-profile")
|
||
put_response = disabled_agent_client.put("/api/user-profile", json={"content": "blocked"})
|
||
|
||
assert get_response.status_code == 403
|
||
assert put_response.status_code == 403
|