668 lines
24 KiB
Python
668 lines
24 KiB
Python
"""Global strategy-agent role CRUD, permissions and visibility tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
import yaml
|
|
from fastapi import HTTPException
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
from app.gateway.routers import _strategy_selection_seed as seed
|
|
from app.gateway.routers import agents as agents_router
|
|
from app.gateway.routers import parallel_agents, strategy_agents
|
|
from deerflow.config import agents_config
|
|
from deerflow.config.paths import Paths
|
|
from deerflow.persistence.agents import MemoryAgentStore
|
|
from deerflow.persistence.skills.memory import MemorySkillStore
|
|
from deerflow.persistence.strategy_agent_roles import MemoryStrategyAgentRoleStore, StrategyAgentRoleRepository
|
|
from deerflow.persistence.strategy_agent_roles.model import StrategyAgentRoleRow
|
|
from deerflow.skills.types import Skill, SkillCategory
|
|
|
|
|
|
class _SkillStorage:
|
|
def __init__(self, skills: list[Skill]) -> None:
|
|
self._skills = skills
|
|
|
|
def load_skills(self, *, enabled_only: bool) -> list[Skill]:
|
|
if enabled_only:
|
|
return [skill for skill in self._skills if skill.enabled]
|
|
return list(self._skills)
|
|
|
|
|
|
def _skill(name: str, category: SkillCategory = SkillCategory.PUBLIC) -> Skill:
|
|
directory = Path("skills") / category.value / name
|
|
return Skill(
|
|
name=name,
|
|
description=name,
|
|
license=None,
|
|
skill_dir=directory,
|
|
skill_file=directory / "SKILL.md",
|
|
relative_path=Path(name),
|
|
category=category,
|
|
enabled=True,
|
|
)
|
|
|
|
|
|
def test_generic_agent_namespace_guard_is_case_insensitive_and_canonical() -> None:
|
|
assert agents_router._is_strategy_agent_id("STRATEGY-SELECTION-INTENT")
|
|
assert strategy_agents._process_lock("STRATEGY-SELECTION-INTENT") is strategy_agents._process_lock("strategy-selection-intent")
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
agents_router._require_canonical_record_id(
|
|
"STRATEGY-SELECTION-INTENT",
|
|
{"id": "strategy-selection-intent"},
|
|
)
|
|
assert exc_info.value.status_code == 409
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_generic_mutations_reject_strategy_namespace_before_storage(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr(
|
|
agents_router,
|
|
"get_agents_api_config",
|
|
lambda: SimpleNamespace(enabled=True),
|
|
)
|
|
request = SimpleNamespace()
|
|
with pytest.raises(HTTPException) as update_error:
|
|
await agents_router.update_agent(
|
|
seed.STRATEGY_INTENT_AGENT_ID.upper(),
|
|
agents_router.AgentUpdateRequest(name="bypass"),
|
|
request,
|
|
)
|
|
assert update_error.value.status_code == 409
|
|
assert "/api/strategy-agents/" in str(update_error.value.detail)
|
|
|
|
with pytest.raises(HTTPException) as delete_error:
|
|
await agents_router.delete_agent(
|
|
seed.STRATEGY_INTENT_AGENT_ID.upper(),
|
|
request,
|
|
)
|
|
assert delete_error.value.status_code == 409
|
|
assert "generic API" in str(delete_error.value.detail)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_strategy_mutation_rejects_case_insensitive_database_alias(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
request, agent_store, role_store = await _request(tmp_path, monkeypatch)
|
|
canonical_id = seed.STRATEGY_INTENT_AGENT_ID
|
|
await role_store.ensure_defaults(seed.default_strategy_agent_roles())
|
|
role = await role_store.get_role(canonical_id)
|
|
record = await agent_store.get_any(canonical_id)
|
|
assert role is not None and record is not None
|
|
|
|
monkeypatch.setattr(strategy_agents, "_ensure_registry", lambda *_: _async_none())
|
|
monkeypatch.setattr(role_store, "get_role", lambda _: _async_value(role))
|
|
monkeypatch.setattr(agent_store, "get_any", lambda _: _async_value(record))
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await strategy_agents._get_role_and_agent(request, canonical_id.upper())
|
|
assert exc_info.value.status_code == 409
|
|
|
|
|
|
async def _async_none() -> None:
|
|
return None
|
|
|
|
|
|
async def _async_value(value):
|
|
return value
|
|
|
|
|
|
async def _request(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
*,
|
|
role: str = "admin",
|
|
) -> tuple[SimpleNamespace, MemoryAgentStore, MemoryStrategyAgentRoleStore]:
|
|
paths = Paths(tmp_path)
|
|
monkeypatch.setattr(seed, "get_paths", lambda: paths)
|
|
monkeypatch.setattr(agents_router, "get_paths", lambda: paths)
|
|
monkeypatch.setattr(strategy_agents, "get_paths", lambda: paths)
|
|
monkeypatch.setattr(agents_config, "get_paths", lambda: paths)
|
|
seed.ensure_strategy_selection_agents()
|
|
|
|
agent_store = MemoryAgentStore()
|
|
for agent_id in seed.required_agent_ids():
|
|
config = yaml.safe_load((paths.agent_dir(agent_id) / "config.yaml").read_text(encoding="utf-8"))
|
|
await agent_store.ensure_builtin(
|
|
{
|
|
"id": agent_id,
|
|
"name": config["name"],
|
|
"description": config.get("description") or "",
|
|
"published": True,
|
|
}
|
|
)
|
|
|
|
role_store = MemoryStrategyAgentRoleStore()
|
|
skill_store = MemorySkillStore()
|
|
storage = _SkillStorage(
|
|
[
|
|
_skill("public-skill"),
|
|
_skill("published-custom", SkillCategory.CUSTOM),
|
|
_skill("private-custom", SkillCategory.CUSTOM),
|
|
]
|
|
)
|
|
await skill_store.create(
|
|
{
|
|
"name": "published-custom",
|
|
"owner_user_id": "publisher",
|
|
"published": True,
|
|
}
|
|
)
|
|
await skill_store.create(
|
|
{
|
|
"name": "private-custom",
|
|
"owner_user_id": "owner",
|
|
"published": False,
|
|
}
|
|
)
|
|
monkeypatch.setattr(agents_router, "get_or_new_skill_storage", lambda **_: storage)
|
|
monkeypatch.setattr(agents_router, "get_skill_usage_entries", lambda: {})
|
|
|
|
state = SimpleNamespace(
|
|
agent_store=agent_store,
|
|
strategy_agent_role_store=role_store,
|
|
skill_store=skill_store,
|
|
config=SimpleNamespace(models=[SimpleNamespace(name="model-a")]),
|
|
)
|
|
request = SimpleNamespace(
|
|
app=SimpleNamespace(state=state),
|
|
state=SimpleNamespace(
|
|
user=SimpleNamespace(id="actor", system_role=role),
|
|
),
|
|
)
|
|
return request, agent_store, role_store
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_catalog_has_fixed_categories_and_redacts_public_config(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
request, _, role_store = await _request(tmp_path, monkeypatch)
|
|
|
|
public = await strategy_agents.list_strategy_agents(request)
|
|
|
|
assert len(public.agents) == 9
|
|
assert [(item.category, item.fixed) for item in public.agents if item.fixed] == [
|
|
("intent", True),
|
|
("judge", True),
|
|
("solution", True),
|
|
]
|
|
assert all(item.model is None and item.soul is None and item.skills is None for item in public.agents)
|
|
|
|
await role_store.update_role(seed.STRATEGY_SECURITY_AGENT_ID, enabled=False)
|
|
active = await strategy_agents.list_strategy_agents(request)
|
|
assert seed.STRATEGY_SECURITY_AGENT_ID not in {item.id for item in active.agents}
|
|
|
|
admin = await strategy_agents.list_strategy_agents(
|
|
request,
|
|
include_disabled=True,
|
|
include_config=True,
|
|
)
|
|
disabled = next(item for item in admin.agents if item.id == seed.STRATEGY_SECURITY_AGENT_ID)
|
|
assert disabled.enabled is False
|
|
assert disabled.soul
|
|
assert disabled.skills == ["policy-knowledge"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_catalog_rejects_user_owned_reserved_default(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
request, agent_store, _ = await _request(tmp_path, monkeypatch)
|
|
await agent_store.delete_builtin(seed.STRATEGY_INTENT_AGENT_ID)
|
|
await agent_store.create(
|
|
{
|
|
"id": seed.STRATEGY_INTENT_AGENT_ID,
|
|
"name": "collision",
|
|
"user_id": "attacker",
|
|
}
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await strategy_agents.list_strategy_agents(request)
|
|
assert exc_info.value.status_code == 409
|
|
assert "owned by a user" in str(exc_info.value.detail)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sensitive_catalog_and_writes_are_admin_only(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
request, _, _ = await _request(tmp_path, monkeypatch, role="user")
|
|
|
|
# The active catalog is intentionally readable by the workbench.
|
|
assert len((await strategy_agents.list_strategy_agents(request)).agents) == 9
|
|
with pytest.raises(HTTPException) as list_error:
|
|
await strategy_agents.list_strategy_agents(
|
|
request,
|
|
include_disabled=True,
|
|
include_config=True,
|
|
)
|
|
assert list_error.value.status_code == 403
|
|
|
|
with pytest.raises(HTTPException) as create_error:
|
|
await strategy_agents.create_strategy_angle(
|
|
strategy_agents.StrategyAngleCreateRequest(name="非法新增"),
|
|
request,
|
|
)
|
|
assert create_error.value.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_angle_crud_is_global_and_delete_is_soft(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
request, agent_store, role_store = await _request(tmp_path, monkeypatch)
|
|
created = await strategy_agents.create_strategy_angle(
|
|
strategy_agents.StrategyAngleCreateRequest(
|
|
name="科技产业视角",
|
|
description="评估关键技术和产业链韧性",
|
|
soul="你是科技产业策略分析智能体。",
|
|
model="model-a",
|
|
skills=["public-skill", "published-custom"],
|
|
),
|
|
request,
|
|
)
|
|
|
|
assert created.category == "angle"
|
|
assert created.fixed is False
|
|
assert created.enabled is True
|
|
assert created.id.startswith(seed.STRATEGY_ANGLE_AGENT_PREFIX)
|
|
assert created.model == "model-a"
|
|
assert created.skills == ["public-skill", "published-custom"]
|
|
|
|
updated = await strategy_agents.update_strategy_agent(
|
|
created.id,
|
|
strategy_agents.StrategyAgentUpdateRequest(
|
|
name="科技与产业链视角",
|
|
enabled=False,
|
|
sort_order=123,
|
|
skills=["public-skill"],
|
|
),
|
|
request,
|
|
)
|
|
assert updated.name == "科技与产业链视角"
|
|
assert updated.enabled is False
|
|
assert updated.sort_order == 123
|
|
assert updated.skills == ["public-skill"]
|
|
|
|
deleted = await strategy_agents.disable_strategy_angle(created.id, request)
|
|
assert deleted.enabled is False
|
|
# Soft delete protects saved projects and permits an administrator to
|
|
# restore the angle without recreating its ID or files.
|
|
assert await agent_store.get_any(created.id) is not None
|
|
assert await role_store.get_role(created.id) is not None
|
|
restored = await strategy_agents.update_strategy_agent(
|
|
created.id,
|
|
strategy_agents.StrategyAgentUpdateRequest(enabled=True),
|
|
request,
|
|
)
|
|
assert restored.enabled is True
|
|
assert (tmp_path / "agents" / created.id / ".strategy-role.yaml").is_file()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_memory_mode_recovers_dynamic_angle_from_folder_sidecar(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
request, _, _ = await _request(tmp_path, monkeypatch)
|
|
created = await strategy_agents.create_strategy_angle(
|
|
strategy_agents.StrategyAngleCreateRequest(
|
|
name="恢复测试角度",
|
|
sort_order=177,
|
|
),
|
|
request,
|
|
)
|
|
await strategy_agents.disable_strategy_angle(created.id, request)
|
|
|
|
request.app.state.agent_store = MemoryAgentStore()
|
|
request.app.state.strategy_agent_role_store = MemoryStrategyAgentRoleStore()
|
|
recovered = await strategy_agents.list_strategy_agents(
|
|
request,
|
|
include_disabled=True,
|
|
include_config=True,
|
|
)
|
|
item = next(entry for entry in recovered.agents if entry.id == created.id)
|
|
assert item.name == "恢复测试角度"
|
|
assert item.enabled is False
|
|
assert item.sort_order == 177
|
|
assert (await request.app.state.agent_store.get_any(created.id))["user_id"] is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_memory_mode_sidecar_keeps_default_angle_disabled_after_restart(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
request, _, _ = await _request(tmp_path, monkeypatch)
|
|
await strategy_agents.update_strategy_agent(
|
|
seed.STRATEGY_SECURITY_AGENT_ID,
|
|
strategy_agents.StrategyAgentUpdateRequest(enabled=False, sort_order=188),
|
|
request,
|
|
)
|
|
|
|
request.app.state.agent_store = MemoryAgentStore()
|
|
request.app.state.strategy_agent_role_store = MemoryStrategyAgentRoleStore()
|
|
with pytest.raises(HTTPException) as run_error:
|
|
await parallel_agents.run_parallel_agents_stream(
|
|
request,
|
|
{
|
|
"task_name": "restart gate",
|
|
"brief": "disabled default must stay disabled",
|
|
"agent_ids": [seed.STRATEGY_SECURITY_AGENT_ID],
|
|
},
|
|
)
|
|
assert run_error.value.status_code == 409
|
|
|
|
recovered = await strategy_agents.list_strategy_agents(
|
|
request,
|
|
include_disabled=True,
|
|
include_config=True,
|
|
)
|
|
item = next(entry for entry in recovered.agents if entry.id == seed.STRATEGY_SECURITY_AGENT_ID)
|
|
assert item.enabled is False
|
|
assert item.sort_order == 188
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancelled_create_compensates_every_persistence_layer(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
request, agent_store, role_store = await _request(tmp_path, monkeypatch)
|
|
generated_id = f"{seed.STRATEGY_ANGLE_AGENT_PREFIX}cancelled"
|
|
monkeypatch.setattr(
|
|
strategy_agents,
|
|
"uuid4",
|
|
lambda: SimpleNamespace(hex="cancelled"),
|
|
)
|
|
|
|
async def cancel_set_extras(*_args, **_kwargs):
|
|
raise asyncio.CancelledError
|
|
|
|
monkeypatch.setattr(agent_store, "set_extras", cancel_set_extras)
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await strategy_agents.create_strategy_angle(
|
|
strategy_agents.StrategyAngleCreateRequest(name="cancelled create"),
|
|
request,
|
|
)
|
|
|
|
assert await agent_store.get_any(generated_id) is None
|
|
assert await role_store.get_role(generated_id) is None
|
|
assert not (tmp_path / "agents" / generated_id).exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cross_process_file_lock_excludes_second_handle(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
paths = Paths(tmp_path)
|
|
monkeypatch.setattr(strategy_agents, "get_paths", lambda: paths)
|
|
first = strategy_agents._open_cross_process_lock(seed.STRATEGY_INTENT_AGENT_ID)
|
|
second = strategy_agents._open_cross_process_lock(seed.STRATEGY_INTENT_AGENT_ID)
|
|
try:
|
|
assert strategy_agents._try_cross_process_lock(first) is True
|
|
assert strategy_agents._try_cross_process_lock(second) is False
|
|
strategy_agents._release_cross_process_lock(first)
|
|
first = None
|
|
assert strategy_agents._try_cross_process_lock(second) is True
|
|
finally:
|
|
if first is not None:
|
|
strategy_agents._release_cross_process_lock(first)
|
|
if not second.closed:
|
|
strategy_agents._release_cross_process_lock(second)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_failure_restores_files_cache_and_role(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
request, agent_store, role_store = await _request(tmp_path, monkeypatch)
|
|
created = await strategy_agents.create_strategy_angle(
|
|
strategy_agents.StrategyAngleCreateRequest(
|
|
name="事务前名称",
|
|
model="model-a",
|
|
skills=["public-skill"],
|
|
sort_order=166,
|
|
),
|
|
request,
|
|
)
|
|
agent_dir = tmp_path / "agents" / created.id
|
|
before_files = {name: (agent_dir / name).read_text(encoding="utf-8") for name in ("config.yaml", "SOUL.md", ".strategy-role.yaml")}
|
|
real_update = role_store.update_role
|
|
failed = False
|
|
|
|
async def fail_once(agent_id: str, **kwargs):
|
|
nonlocal failed
|
|
if kwargs.get("sort_order") == 188 and not failed:
|
|
failed = True
|
|
raise RuntimeError("injected role-store failure")
|
|
return await real_update(agent_id, **kwargs)
|
|
|
|
monkeypatch.setattr(role_store, "update_role", fail_once)
|
|
with pytest.raises(RuntimeError, match="injected role-store failure"):
|
|
await strategy_agents.update_strategy_agent(
|
|
created.id,
|
|
strategy_agents.StrategyAgentUpdateRequest(
|
|
name="不应留下的名称",
|
|
soul="不应留下的 SOUL",
|
|
skills=[],
|
|
sort_order=188,
|
|
),
|
|
request,
|
|
)
|
|
|
|
record = await agent_store.get_any(created.id)
|
|
role = await role_store.get_role(created.id)
|
|
assert record is not None and record["name"] == "事务前名称"
|
|
assert role is not None and role["sort_order"] == 166
|
|
assert (await agent_store.get_extras_for([created.id]))[created.id]["skills"] == ["public-skill"]
|
|
assert {name: (agent_dir / name).read_text(encoding="utf-8") for name in ("config.yaml", "SOUL.md", ".strategy-role.yaml")} == before_files
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fixed_roles_cannot_be_disabled_or_deleted(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
request, _, _ = await _request(tmp_path, monkeypatch)
|
|
|
|
with pytest.raises(HTTPException) as update_error:
|
|
await strategy_agents.update_strategy_agent(
|
|
seed.STRATEGY_INTENT_AGENT_ID,
|
|
strategy_agents.StrategyAgentUpdateRequest(enabled=False),
|
|
request,
|
|
)
|
|
assert update_error.value.status_code == 409
|
|
|
|
with pytest.raises(HTTPException) as delete_error:
|
|
await strategy_agents.disable_strategy_angle(
|
|
seed.STRATEGY_JUDGE_AGENT_ID,
|
|
request,
|
|
)
|
|
assert delete_error.value.status_code == 409
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_model_and_global_skill_visibility_are_validated(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
request, _, _ = await _request(tmp_path, monkeypatch)
|
|
|
|
with pytest.raises(HTTPException) as model_error:
|
|
await strategy_agents.create_strategy_angle(
|
|
strategy_agents.StrategyAngleCreateRequest(
|
|
name="未知模型",
|
|
model="not-configured",
|
|
),
|
|
request,
|
|
)
|
|
assert model_error.value.status_code == 422
|
|
|
|
with pytest.raises(HTTPException) as skill_error:
|
|
await strategy_agents.create_strategy_angle(
|
|
strategy_agents.StrategyAngleCreateRequest(
|
|
name="私有技能",
|
|
skills=["private-custom"],
|
|
),
|
|
request,
|
|
)
|
|
assert skill_error.value.status_code == 422
|
|
assert "globally available" in str(skill_error.value.detail)
|
|
|
|
defaulted = await strategy_agents.create_strategy_angle(
|
|
strategy_agents.StrategyAngleCreateRequest(
|
|
name="默认角色模板",
|
|
description="关注跨领域风险",
|
|
),
|
|
request,
|
|
)
|
|
assert defaulted.soul
|
|
assert "默认角色模板" in defaulted.soul
|
|
assert "不替代其他角度" in defaulted.soul
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disabled_strategy_agent_cannot_be_started_via_parallel_api(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
request, _, role_store = await _request(tmp_path, monkeypatch)
|
|
await role_store.ensure_defaults(seed.default_strategy_agent_roles())
|
|
await role_store.update_role(seed.STRATEGY_SECURITY_AGENT_ID, enabled=False)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await parallel_agents.run_parallel_agents_stream(
|
|
request,
|
|
{
|
|
"task_name": "测试",
|
|
"brief": "验证停用闸门",
|
|
"agent_ids": [seed.STRATEGY_SECURITY_AGENT_ID],
|
|
},
|
|
)
|
|
assert exc_info.value.status_code == 409
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_parallel_api_rejects_user_owned_strategy_backing_agent(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
request, agent_store, role_store = await _request(tmp_path, monkeypatch)
|
|
await role_store.ensure_defaults(seed.default_strategy_agent_roles())
|
|
await agent_store.delete_builtin(seed.STRATEGY_SECURITY_AGENT_ID)
|
|
await agent_store.create(
|
|
{
|
|
"id": seed.STRATEGY_SECURITY_AGENT_ID,
|
|
"name": "collision",
|
|
"user_id": "attacker",
|
|
}
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await parallel_agents.run_parallel_agents_stream(
|
|
request,
|
|
{
|
|
"task_name": "测试",
|
|
"brief": "验证归属闸门",
|
|
"agent_ids": [seed.STRATEGY_SECURITY_AGENT_ID],
|
|
},
|
|
)
|
|
assert exc_info.value.status_code == 409
|
|
assert "owned by a user" in str(exc_info.value.detail)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_memory_registry_repairs_corrupted_fixed_role_metadata() -> None:
|
|
store = MemoryStrategyAgentRoleStore()
|
|
await store.create_role(
|
|
{
|
|
"agent_id": seed.STRATEGY_INTENT_AGENT_ID,
|
|
"category": "angle",
|
|
"fixed": False,
|
|
"enabled": False,
|
|
"sort_order": 999,
|
|
}
|
|
)
|
|
|
|
await store.ensure_defaults(seed.default_strategy_agent_roles())
|
|
repaired = await store.get_role(seed.STRATEGY_INTENT_AGENT_ID)
|
|
assert repaired is not None
|
|
assert repaired["category"] == "intent"
|
|
assert repaired["fixed"] is True
|
|
assert repaired["enabled"] is True
|
|
assert repaired["sort_order"] == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sql_registry_preserves_disabled_defaults_and_soft_records(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'strategy-agent-roles.db'}")
|
|
async with engine.begin() as connection:
|
|
await connection.run_sync(StrategyAgentRoleRow.__table__.create)
|
|
store = StrategyAgentRoleRepository(async_sessionmaker(engine, expire_on_commit=False))
|
|
try:
|
|
defaults = seed.default_strategy_agent_roles()
|
|
assert await store.ensure_defaults(defaults) == 9
|
|
assert await store.ensure_defaults(defaults) == 0
|
|
await store.update_role(seed.STRATEGY_SECURITY_AGENT_ID, enabled=False)
|
|
|
|
# Re-seeding must never undo an administrator's soft delete.
|
|
assert await store.ensure_defaults(defaults) == 0
|
|
disabled = await store.get_role(seed.STRATEGY_SECURITY_AGENT_ID)
|
|
assert disabled is not None
|
|
assert disabled["enabled"] is False
|
|
|
|
# Canonical fixed-role invariants are repaired, unlike angle lifecycle
|
|
# state which intentionally remains administrator-controlled.
|
|
async with store._sf() as session:
|
|
row = await session.get(
|
|
StrategyAgentRoleRow,
|
|
seed.STRATEGY_INTENT_AGENT_ID,
|
|
)
|
|
row.category = "angle"
|
|
row.fixed = False
|
|
row.enabled = False
|
|
row.sort_order = 999
|
|
await session.commit()
|
|
assert await store.ensure_defaults(defaults) == 0
|
|
repaired = await store.get_role(seed.STRATEGY_INTENT_AGENT_ID)
|
|
assert repaired is not None
|
|
assert repaired["category"] == "intent"
|
|
assert repaired["fixed"] is True
|
|
assert repaired["enabled"] is True
|
|
assert repaired["sort_order"] == 0
|
|
|
|
dynamic = await store.create_role(
|
|
{
|
|
"agent_id": "strategy-selection-angle-sql-test",
|
|
"category": "angle",
|
|
"fixed": False,
|
|
"enabled": True,
|
|
"sort_order": 150,
|
|
}
|
|
)
|
|
assert dynamic["category"] == "angle"
|
|
assert await store.update_role(dynamic["agent_id"], enabled=False)
|
|
assert await store.delete_role(dynamic["agent_id"]) is True
|
|
assert await store.get_role(dynamic["agent_id"]) is None
|
|
finally:
|
|
await engine.dispose()
|