mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 21:26:39 +00:00
505c33c793
* fix(scheduler): reject nonexistent local times at admission * fix(scheduler): label run_at errors and clarify cron DST behavior
1171 lines
60 KiB
Python
1171 lines
60 KiB
Python
"""Gateway schedule grants retain the owner's permissions and origin scope."""
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock
|
|
from zoneinfo import ZoneInfo
|
|
|
|
import pytest
|
|
from _router_auth_helpers import call_unwrapped
|
|
|
|
from app.gateway.auth_disabled import AUTH_SOURCE_AUTH_DISABLED, AUTH_SOURCE_INTERNAL, AUTH_SOURCE_PAT, AUTH_SOURCE_SESSION
|
|
from app.gateway.authz import AuthContext
|
|
from app.gateway.routers import scheduled_tasks
|
|
from deerflow.agents.interaction_policy import RunInteractionMode, RunInteractionPolicy
|
|
from deerflow.config.app_config import AppConfig
|
|
from deerflow.persistence.scheduled_tasks import ActiveScheduledTaskMutationConflict
|
|
|
|
_PERMISSIONS = ("threads:read", "threads:write", "runs:create")
|
|
|
|
|
|
def _user(*, user_id="alice", token_version=0, role="user"):
|
|
return SimpleNamespace(id=user_id, token_version=token_version, system_role=role, needs_setup=False)
|
|
|
|
|
|
_FUTURE = (datetime.now(UTC) + timedelta(days=1)).replace(microsecond=0).isoformat()
|
|
|
|
|
|
def _task(*, user_id="alice", origin="origin", **updates):
|
|
return {
|
|
"id": "task-1",
|
|
"user_id": user_id,
|
|
"origin_thread_id": origin,
|
|
"title": "Report",
|
|
"prompt": "Write the report",
|
|
"stop_condition": None,
|
|
"status": "enabled",
|
|
"schedule_type": "interval",
|
|
"schedule_spec": {"every_seconds": 3600},
|
|
"timezone": "UTC",
|
|
"next_run_at": _FUTURE,
|
|
"context_mode": "fresh_thread_per_run",
|
|
"goal_objective": None,
|
|
"max_runs": None,
|
|
"end_at": None,
|
|
"standing_notes": [],
|
|
**updates,
|
|
}
|
|
|
|
|
|
def _setup(monkeypatch, *, source=AUTH_SOURCE_SESSION, permissions=_PERMISSIONS, original_text="Run this task now", mode="interactive", task=None, enabled=True, tool_enabled=True):
|
|
from app.gateway import scheduled_task_access as access
|
|
|
|
user = _user()
|
|
repo = SimpleNamespace(
|
|
get=AsyncMock(return_value=task or _task()),
|
|
list_by_origin_thread=AsyncMock(return_value=[task or _task()]),
|
|
list_manageable_from_thread=AsyncMock(return_value=[task or _task()]),
|
|
create=AsyncMock(side_effect=lambda **values: {"id": values["task_id"], "status": "enabled", **values}),
|
|
pause_with_queue_cancellation=AsyncMock(return_value="paused"),
|
|
delete_with_queue_cancellation=AsyncMock(return_value="deleted"),
|
|
append_standing_note=AsyncMock(return_value=task or _task(standing_notes=["use develop"])),
|
|
get_active_run_status=AsyncMock(return_value=None),
|
|
active_run_status_for=AsyncMock(return_value={}),
|
|
automatic_runs_used_for=AsyncMock(return_value={}),
|
|
)
|
|
|
|
async def update(task_id, *, user_id, updates, require_mutable=False):
|
|
return {**(await repo.get(task_id, user_id=user_id)), **updates}
|
|
|
|
repo.update = AsyncMock(side_effect=update)
|
|
occurrences = SimpleNamespace(request_stop=AsyncMock(return_value=True), task_ids_for_thread=AsyncMock(return_value=[]), list_by_task=AsyncMock(return_value=[]))
|
|
service = SimpleNamespace(is_running=True, dispatch_task=AsyncMock(return_value={"outcome": "launched", "run_id": "trial-run", "thread_id": "trial-thread", "existing": False}))
|
|
threads = SimpleNamespace(get=AsyncMock(return_value={"user_id": "alice"}), check_access=AsyncMock(return_value=True))
|
|
provider = SimpleNamespace(get_user=AsyncMock(return_value=user))
|
|
config = AppConfig.model_validate({"sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"}, "scheduler": {"enabled": enabled, "tool_enabled": tool_enabled}})
|
|
request = SimpleNamespace(state=SimpleNamespace(auth_source=source, user=user, auth=AuthContext(user, list(permissions))), app=SimpleNamespace(state=SimpleNamespace(scheduled_task_service=service)))
|
|
monkeypatch.setattr(access, "get_config", lambda: config)
|
|
monkeypatch.setattr(access, "get_scheduled_task_repo", lambda _request: repo)
|
|
monkeypatch.setattr(access, "get_scheduled_task_run_repo", lambda _request: occurrences)
|
|
monkeypatch.setattr(access, "get_scheduled_task_service", lambda _request: service)
|
|
monkeypatch.setattr(access, "get_thread_store", lambda _request: threads)
|
|
monkeypatch.setattr(access, "get_local_provider", lambda: provider)
|
|
resolve = AsyncMock(return_value=list(permissions))
|
|
monkeypatch.setattr(access, "resolve_route_permissions", resolve)
|
|
|
|
async def prepare(**updates):
|
|
return await access.prepare_scheduler_capability(
|
|
request,
|
|
user=user,
|
|
thread_id="origin",
|
|
run_id="run-1",
|
|
assistant_id="lead_agent",
|
|
original_user_text=original_text,
|
|
interaction_policy=RunInteractionPolicy(RunInteractionMode(mode)),
|
|
**updates,
|
|
)
|
|
|
|
return prepare, repo, occurrences, service, threads, provider, resolve, request
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("source", [None, AUTH_SOURCE_PAT, "unknown"])
|
|
async def test_untrusted_auth_sources_cannot_receive_a_schedule_grant(monkeypatch, source):
|
|
prepare, *_ = _setup(monkeypatch, source=source)
|
|
assert await prepare() is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(("enabled", "tool_enabled"), [(False, True), (True, False)])
|
|
async def test_operator_flags_withhold_the_gateway_grant(monkeypatch, enabled, tool_enabled):
|
|
prepare, *_ = _setup(monkeypatch, enabled=enabled, tool_enabled=tool_enabled)
|
|
assert await prepare() is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_capability_requires_the_running_scheduler_service(monkeypatch):
|
|
from fastapi import HTTPException
|
|
|
|
from app.gateway import scheduled_task_access as access
|
|
|
|
prepare, _, _, service, *_ = _setup(monkeypatch)
|
|
service.is_running = False
|
|
assert await prepare() is None
|
|
service.is_running = True
|
|
assert await prepare() is not None
|
|
monkeypatch.setattr(access, "get_scheduled_task_service", lambda _request: (_ for _ in ()).throw(HTTPException(503)))
|
|
assert await prepare() is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("mode", ["scheduled", "autonomous", "webhook"])
|
|
async def test_noninteractive_runs_without_private_occurrence_authority_get_no_grant(monkeypatch, mode):
|
|
prepare, *_ = _setup(monkeypatch, mode=mode)
|
|
assert await prepare() is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_principal_and_permission_ceiling_cannot_be_widened(monkeypatch):
|
|
prepare, repo, _, _, _, _, resolve, request = _setup(monkeypatch, permissions=("threads:read",))
|
|
capability = await prepare()
|
|
resolve.return_value = list(_PERMISSIONS)
|
|
result = await capability.manage(action="create", request={})
|
|
assert result["status_code"] == 403
|
|
repo.create.assert_not_awaited()
|
|
request.state.auth.user = _user(user_id="bob")
|
|
assert await prepare() is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("owner", [None, _user(user_id="bob"), _user(token_version=1)])
|
|
async def test_deleted_changed_or_revoked_owner_cannot_use_an_existing_grant(monkeypatch, owner):
|
|
prepare, repo, _, _, _, provider, _, _ = _setup(monkeypatch)
|
|
capability = await prepare()
|
|
provider.get_user.return_value = owner
|
|
result = await capability.manage(action="list", request={})
|
|
assert result["status_code"] == 401
|
|
repo.list_manageable_from_thread.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_live_permission_revocation_is_honored_each_action(monkeypatch):
|
|
prepare, repo, _, _, _, _, resolve, _ = _setup(monkeypatch)
|
|
capability = await prepare()
|
|
resolve.return_value = []
|
|
result = await capability.manage(action="list", request={})
|
|
assert result["status_code"] == 403
|
|
repo.list_manageable_from_thread.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(("action", "permission"), [("list", "threads:read"), ("create", "runs:create"), ("pause", "threads:write"), ("delete", "threads:write"), ("note", "threads:write"), ("trial", "runs:create")])
|
|
async def test_each_action_checks_its_route_permissions(monkeypatch, action, permission):
|
|
prepare, repo, _, service, *_ = _setup(monkeypatch, permissions=tuple(value for value in _PERMISSIONS if value != permission))
|
|
capability = await prepare()
|
|
result = await capability.manage(action=action, request={"task_id": "task-1"})
|
|
assert result["status_code"] == 403
|
|
repo.get.assert_not_awaited()
|
|
repo.create.assert_not_awaited()
|
|
service.dispatch_task.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("source_owner", [None, "bob"])
|
|
async def test_deleted_shared_or_foreign_origin_is_not_a_management_grant(monkeypatch, source_owner):
|
|
prepare, _, _, _, threads, *_ = _setup(monkeypatch)
|
|
threads.get.return_value = None if source_owner is None else {"user_id": source_owner}
|
|
assert await prepare() is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("task", [_task(user_id="bob"), _task(origin="other-origin"), _task(origin=None)])
|
|
@pytest.mark.parametrize("action", ["pause", "delete", "note", "trial"])
|
|
async def test_management_targets_require_both_owner_and_origin(monkeypatch, task, action):
|
|
prepare, repo, _, service, *_ = _setup(monkeypatch, task=task, original_text="use develop")
|
|
capability = await prepare()
|
|
payload = {"task_id": "task-1", **({"note": "use develop"} if action == "note" else {})}
|
|
result = await capability.manage(action=action, request=payload)
|
|
assert result["status_code"] == 404
|
|
repo.pause_with_queue_cancellation.assert_not_awaited()
|
|
repo.delete_with_queue_cancellation.assert_not_awaited()
|
|
repo.append_standing_note.assert_not_awaited()
|
|
service.dispatch_task.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_projects_only_safe_fields_from_the_bound_origin(monkeypatch):
|
|
prepare, repo, *_ = _setup(monkeypatch, task=_task(secret="must-not-leak", lease_owner="private-worker", standing_notes=["use develop"]))
|
|
capability = await prepare()
|
|
result = await capability.manage(action="list", request={})
|
|
repo.list_manageable_from_thread.assert_awaited_once_with("alice", "origin")
|
|
assert result["tasks"][0]["standing_notes"] == ["use develop"]
|
|
assert "secret" not in result["tasks"][0]
|
|
assert "lease_owner" not in result["tasks"][0]
|
|
assert "next_run_at" not in result["tasks"][0]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_uses_bound_origin_and_echoes_goal_and_stop_details(monkeypatch):
|
|
prepare, repo, *_ = _setup(monkeypatch)
|
|
capability = await prepare()
|
|
result = await capability.manage(
|
|
action="create",
|
|
request={"title": "Report", "prompt": "Write the report", "schedule_type": "interval", "schedule_spec": {"every_seconds": 3600}, "timezone": "UTC", "goal_objective": "Write a report from the source documents", "max_runs": 3},
|
|
)
|
|
values = repo.create.await_args.kwargs
|
|
assert values["user_id"] == "alice"
|
|
assert values["origin_thread_id"] == "origin"
|
|
assert values["thread_id"] is None
|
|
assert values["assistant_id"] == "lead_agent"
|
|
task = result["task"]
|
|
assert task["goal_objective"] == "Write a report from the source documents"
|
|
assert task["max_runs"] == 3
|
|
assert task["next_run_local"].endswith("(UTC)")
|
|
assert "next_run_at" not in task
|
|
assert "stop_instructions" not in result
|
|
assert "stop_instructions" not in task
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(("action", "field"), [("create", "run_at"), ("update", "run_at"), ("create", "end_at"), ("update", "end_at"), ("resume", "end_at")])
|
|
async def test_conversation_rejects_nonexistent_local_time_before_persistence(monkeypatch, action, field):
|
|
from app.gateway import scheduled_task_access as access
|
|
|
|
class FixedDatetime(datetime):
|
|
@classmethod
|
|
def now(cls, tz=None):
|
|
return cls(2026, 10, 22, tzinfo=UTC)
|
|
|
|
monkeypatch.setattr(access, "datetime", FixedDatetime)
|
|
task = _task(timezone="America/New_York", status="paused" if action == "resume" else "enabled", next_run_at="2026-10-23T00:00:00+00:00")
|
|
prepare, repo, *_ = _setup(monkeypatch, task=task)
|
|
capability = await prepare(client_timezone="America/New_York")
|
|
wall_time = "2027-03-14T02:30:00"
|
|
fields = {"schedule_type": "once", "schedule_spec": {"run_at": wall_time}} if field == "run_at" else {"end_at": wall_time}
|
|
payload = {"title": "Report", "prompt": "Write the report", "schedule_type": "interval", "schedule_spec": {"every_seconds": 3600}} if action == "create" else {"task_id": "task-1"}
|
|
result = await capability.manage(action=action, request={**payload, **fields})
|
|
assert result.get("status_code") == 422, result
|
|
assert result["code"] == ("invalid_schedule" if field == "run_at" else "invalid_request")
|
|
assert "does not exist" in result["error"]
|
|
repo.create.assert_not_awaited()
|
|
repo.update.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("objective", ["", " \t\n ", "g" * 4001])
|
|
async def test_create_rejects_invalid_goal_before_persisting_schedule(monkeypatch, objective):
|
|
prepare, repo, *_ = _setup(monkeypatch)
|
|
capability = await prepare()
|
|
result = await capability.manage(action="create", request={"title": "Report", "prompt": "Write report", "schedule_type": "interval", "schedule_spec": {"every_seconds": 3600}, "timezone": "UTC", "goal_objective": objective})
|
|
assert result["status_code"] == 422
|
|
repo.create.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("objective", [None, " " + "g" * 4000 + "\t\n"])
|
|
async def test_create_preserves_optional_and_normalized_boundary_goal(monkeypatch, objective):
|
|
prepare, repo, *_ = _setup(monkeypatch)
|
|
capability = await prepare()
|
|
result = await capability.manage(action="create", request={"title": "Report", "prompt": "Write report", "schedule_type": "interval", "schedule_spec": {"every_seconds": 3600}, "timezone": "UTC", "goal_objective": objective})
|
|
assert "error" not in result
|
|
assert result["task"]["goal_objective"] == objective
|
|
assert repo.create.await_args.kwargs["goal_objective"] == objective
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("payload", [{"user_id": "bob"}, {"thread_id": "other-origin"}, {"assistant_id": "other-agent"}, {"origin_thread_id": "other-origin"}])
|
|
async def test_model_identity_fields_are_rejected_even_if_directly_called(monkeypatch, payload):
|
|
prepare, repo, *_ = _setup(monkeypatch)
|
|
capability = await prepare()
|
|
result = await capability.manage(action="create", request=payload)
|
|
assert result["status_code"] == 422
|
|
repo.create.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("schedule_type,schedule_spec", [("interval", {"every_seconds": 60}), ("cron", {"cron": "*/5 * * * *"}), ("cron", {"cron": "0,30 * * * *"})])
|
|
async def test_frequent_conversation_schedules_require_a_user_end_condition(monkeypatch, schedule_type, schedule_spec):
|
|
prepare, repo, *_ = _setup(monkeypatch)
|
|
capability = await prepare()
|
|
result = await capability.manage(action="create", request={"title": "Check", "prompt": "Check sources", "schedule_type": schedule_type, "schedule_spec": schedule_spec, "timezone": "UTC"})
|
|
assert result["status_code"] == 422
|
|
repo.create.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reused_thread_goal_is_rejected(monkeypatch):
|
|
prepare, repo, *_ = _setup(monkeypatch)
|
|
capability = await prepare()
|
|
result = await capability.manage(
|
|
action="create", request={"title": "Report", "prompt": "Write report", "schedule_type": "interval", "schedule_spec": {"every_seconds": 3600}, "timezone": "UTC", "context_mode": "reuse_thread", "goal_objective": "Write report"}
|
|
)
|
|
assert result["status_code"] == 422
|
|
repo.create.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("bounds", [{"goal_objective": " "}, {"max_runs": True}, {"end_at": "2025-01-01T00:00:00+00:00"}, {"end_at": "2027-01-01T00:00:00"}, {"end_at": "2027-01-01T00:00:00+00:00"}])
|
|
async def test_create_rejects_unusable_goal_and_end_bounds(monkeypatch, bounds):
|
|
prepare, repo, *_ = _setup(monkeypatch)
|
|
capability = await prepare()
|
|
definition = {"title": "Report", "prompt": "Write report", "schedule_type": "once", "schedule_spec": {"run_at": "2027-01-01T00:00:00+00:00"}, "timezone": "UTC", **bounds}
|
|
result = await capability.manage(action="create", request=definition)
|
|
assert result["status_code"] == 422
|
|
repo.create.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("note", ["inferred correction", "", "x" * 501])
|
|
async def test_notes_require_bounded_verbatim_current_user_words(monkeypatch, note):
|
|
prepare, repo, *_ = _setup(monkeypatch, original_text="For this task, use develop.")
|
|
capability = await prepare()
|
|
result = await capability.manage(action="note", request={"task_id": "task-1", "note": note})
|
|
assert result["status_code"] == 422
|
|
repo.append_standing_note.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_note_keeps_the_users_exact_text_and_repository_guard(monkeypatch):
|
|
prepare, repo, *_ = _setup(monkeypatch, original_text="For this task: use develop ")
|
|
capability = await prepare()
|
|
await capability.manage(action="note", request={"task_id": "task-1", "note": " use develop "})
|
|
repo.append_standing_note.assert_awaited_once_with("task-1", user_id="alice", note=" use develop ")
|
|
repo.append_standing_note.side_effect = ActiveScheduledTaskMutationConflict("running")
|
|
result = await capability.manage(action="note", request={"task_id": "task-1", "note": "use develop"})
|
|
assert result["status_code"] == 409
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trial_requires_a_current_user_turn_and_uses_manual_dispatch(monkeypatch):
|
|
prepare, _, _, service, *_ = _setup(monkeypatch, original_text="")
|
|
capability = await prepare()
|
|
result = await capability.manage(action="trial", request={"task_id": "task-1"})
|
|
assert result["status_code"] == 403
|
|
service.dispatch_task.assert_not_awaited()
|
|
prepare, _, _, service, *_ = _setup(monkeypatch, original_text="Yes, run it now.")
|
|
capability = await prepare()
|
|
result = await capability.manage(action="trial", request={"task_id": "task-1"})
|
|
assert result["trial"]["outcome"] == "launched"
|
|
assert service.dispatch_task.await_args.kwargs["trigger"] == "manual"
|
|
service.dispatch_task.return_value = {"outcome": "failed", "error": "cannot launch"}
|
|
result = await capability.manage(action="trial", request={"task_id": "task-1"})
|
|
assert result["status_code"] == 502
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"user_text",
|
|
[
|
|
"List my tasks",
|
|
"What is a trial?",
|
|
"task-1",
|
|
"Report",
|
|
"Don't run it now.",
|
|
"Do not run this task now",
|
|
"不要试跑",
|
|
"不要先跑一次",
|
|
"Not now; run it once after I approve",
|
|
"If I approve, run it now",
|
|
'The file says "Run it now"',
|
|
'"Run it now"',
|
|
"What does 先跑一次 mean?",
|
|
"Yes",
|
|
"yes",
|
|
"好",
|
|
"ok",
|
|
"OK!",
|
|
"好的。",
|
|
"Run it now?",
|
|
"run task-2 now",
|
|
"Run Other report now",
|
|
"Run it now and then delete all tasks",
|
|
"先跑一次，但不要真的执行",
|
|
],
|
|
)
|
|
async def test_trial_rejects_unrelated_negative_quoted_or_ambiguous_turns(monkeypatch, user_text):
|
|
prepare, _, _, service, *_ = _setup(monkeypatch, original_text=user_text, task=_task(title="Report"))
|
|
capability = await prepare()
|
|
result = await capability.manage(action="trial", request={"task_id": "task-1"})
|
|
assert result["status_code"] == 403
|
|
service.dispatch_task.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("user_text", ["Run it now", "Yes, run it now.", "Please run this task once", "Run task-1 now", "先跑一次", "请先试跑一次。", "可以，先跑一次", "先试跑task-1", "现在运行task-1一次"])
|
|
async def test_trial_accepts_direct_requests_in_the_current_user_turn(monkeypatch, user_text):
|
|
prepare, _, _, service, *_ = _setup(monkeypatch, original_text=user_text, task=_task(title="Report"))
|
|
capability = await prepare()
|
|
result = await capability.manage(action="trial", request={"task_id": "task-1"})
|
|
assert result["trial"]["outcome"] == "launched"
|
|
assert service.dispatch_task.await_args.kwargs["trigger"] == "manual"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
("title", "user_text"), [("周报，但不要真的执行", "先试跑周报，但不要真的执行"), ("周报如果我以后批准", "现在试跑周报如果我以后批准"), ("it now and then delete all tasks", "Run it now and then delete all tasks once")]
|
|
)
|
|
async def test_trial_never_uses_task_title_as_authorization_text(monkeypatch, title, user_text):
|
|
prepare, _, _, service, *_ = _setup(monkeypatch, original_text=user_text, task=_task(title=title))
|
|
capability = await prepare()
|
|
result = await capability.manage(action="trial", request={"task_id": "task-1"})
|
|
assert result["status_code"] == 403
|
|
service.dispatch_task.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("action", ["create", "pause"])
|
|
async def test_tool_quota_conflicts_use_409_without_dispatch(monkeypatch, action):
|
|
from deerflow.persistence.scheduled_tasks import ScheduledTaskQuotaExceeded
|
|
|
|
prepare, repo, _, service, *_ = _setup(monkeypatch)
|
|
capability = await prepare()
|
|
request = {"task_id": "task-1"}
|
|
if action == "create":
|
|
request = {"title": "Report", "prompt": "Write report", "schedule_type": "interval", "schedule_spec": {"every_seconds": 3600}, "timezone": "UTC"}
|
|
repo.create.side_effect = ScheduledTaskQuotaExceeded()
|
|
else:
|
|
repo.pause_with_queue_cancellation.side_effect = ScheduledTaskQuotaExceeded()
|
|
result = await capability.manage(action=action, request=request)
|
|
assert (result["status_code"], result["code"]) == (409, "task_quota_exceeded")
|
|
service.dispatch_task.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_internal_im_grant_uses_actual_owners_noninternal_permissions(monkeypatch):
|
|
prepare, _, _, _, _, _, resolve, request = _setup(monkeypatch, source=AUTH_SOURCE_INTERNAL)
|
|
request.state.auth = AuthContext(_user(user_id="internal", role="internal"), ["threads:read"])
|
|
capability = await prepare()
|
|
assert capability is not None
|
|
assert all(call.kwargs["is_internal"] is False for call in resolve.await_args_list)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_synthetic_internal_principal_cannot_become_the_schedule_owner(monkeypatch):
|
|
prepare, _, _, _, _, provider, _, request = _setup(monkeypatch, source=AUTH_SOURCE_INTERNAL)
|
|
request.state.user.system_role = "internal"
|
|
provider.get_user.return_value.system_role = "internal"
|
|
assert await prepare() is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scheduled_grant_only_stops_the_privately_bound_occurrence(monkeypatch):
|
|
prepare, _, occurrences, *_ = _setup(monkeypatch, source=AUTH_SOURCE_INTERNAL, mode="scheduled")
|
|
capability = await prepare(scheduled_task_runtime={"user_id": "alice", "task_id": "task-1", "occurrence_id": "occurrence-1"})
|
|
assert capability.mode == "scheduled"
|
|
result = await capability.stop_current_schedule()
|
|
assert result == {"action": "stop", "stop_requested": True}
|
|
occurrences.request_stop.assert_awaited_once_with("occurrence-1", task_id="task-1", run_id="run-1", user_id="alice")
|
|
result = await capability.manage(action="list", request={})
|
|
assert result["status_code"] == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("runtime", [{"user_id": "bob", "task_id": "task-1", "occurrence_id": "occurrence-1"}, {"user_id": "alice", "task_id": "task-1"}])
|
|
async def test_scheduled_runtime_must_bind_the_actual_owner_and_occurrence(monkeypatch, runtime):
|
|
prepare, *_ = _setup(monkeypatch, source=AUTH_SOURCE_INTERNAL, mode="scheduled")
|
|
assert await prepare(scheduled_task_runtime=runtime) is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_self_stop_rechecks_permissions_and_durable_occurrence_binding(monkeypatch):
|
|
prepare, _, occurrences, _, _, _, resolve, _ = _setup(monkeypatch, source=AUTH_SOURCE_INTERNAL, mode="scheduled")
|
|
capability = await prepare(scheduled_task_runtime={"user_id": "alice", "task_id": "task-1", "occurrence_id": "occurrence-1"})
|
|
occurrences.request_stop.return_value = False
|
|
stopped = await capability.stop_current_schedule()
|
|
assert (stopped["status_code"], stopped["code"]) == (409, "occurrence_not_active")
|
|
resolve.return_value = ["threads:read"]
|
|
stopped = await capability.stop_current_schedule()
|
|
assert (stopped["status_code"], stopped["code"]) == (403, "permission_denied")
|
|
assert occurrences.request_stop.await_count == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auth_disabled_default_owner_needs_the_explicit_host_mode(monkeypatch):
|
|
from app.gateway import scheduled_task_access as access
|
|
from app.gateway.auth_disabled import get_auth_disabled_user
|
|
|
|
prepare, _, _, _, threads, provider, _, request = _setup(monkeypatch, source=AUTH_SOURCE_AUTH_DISABLED)
|
|
user = get_auth_disabled_user()
|
|
request.state.user = user
|
|
request.state.auth = AuthContext(user, list(_PERMISSIONS))
|
|
threads.get.return_value = {"user_id": str(user.id)}
|
|
provider.get_user.return_value = None
|
|
monkeypatch.setattr(access, "is_auth_disabled", lambda: True)
|
|
capability = await access.prepare_scheduler_capability(request, user=user, thread_id="origin", run_id="run-1", assistant_id="lead_agent", original_user_text="List tasks", interaction_policy=RunInteractionPolicy.interactive())
|
|
assert capability is not None
|
|
result = await capability.manage(action="list", request={})
|
|
assert "error" not in result
|
|
monkeypatch.setattr(access, "is_auth_disabled", lambda: False)
|
|
result = await capability.manage(action="list", request={})
|
|
assert result["status_code"] == 401
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"updates",
|
|
[
|
|
{"timezone": "missing/zone"},
|
|
{"timezone": "Europe"},
|
|
{"schedule_type": "cron", "schedule_spec": {"cron": "0 9 * * *"}, "timezone": "Europe"},
|
|
{"schedule_type": "unsupported"},
|
|
{"schedule_spec": {"every_seconds": 1}},
|
|
{"schedule_type": "cron", "schedule_spec": {"cron": "not cron"}},
|
|
{"context_mode": "wrong"},
|
|
{"schedule_type": "once", "schedule_spec": {"run_at": (datetime.now(UTC) - timedelta(hours=1)).isoformat()}},
|
|
],
|
|
)
|
|
async def test_rest_and_tool_share_creation_validation_status_and_reason(monkeypatch, updates):
|
|
prepare, repo, _, _, threads, _, _, request = _setup(monkeypatch)
|
|
capability = await prepare()
|
|
body = {"title": "Task", "prompt": "Work", "schedule_type": "interval", "schedule_spec": {"every_seconds": 3600}, "timezone": "UTC", **updates}
|
|
tool_result = await capability.manage(action="create", request=body)
|
|
monkeypatch.setattr(scheduled_tasks, "get_scheduled_task_repo", lambda _request: repo)
|
|
monkeypatch.setattr(scheduled_tasks, "get_thread_store", lambda _request: threads)
|
|
monkeypatch.setattr(scheduled_tasks, "get_config", lambda: AppConfig.model_validate({"sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"}}))
|
|
monkeypatch.setattr(scheduled_tasks, "get_optional_user_from_request", AsyncMock(return_value=_user()))
|
|
from fastapi import HTTPException
|
|
|
|
with pytest.raises(HTTPException) as error:
|
|
await call_unwrapped(scheduled_tasks.create_scheduled_task, request=request, body=scheduled_tasks.ScheduledTaskCreateRequest(**body))
|
|
assert tool_result["status_code"] == error.value.status_code
|
|
assert tool_result["code"] == error.value.detail["code"]
|
|
assert tool_result["error"] == error.value.detail["message"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_capability_maps_real_repository_note_capacity_and_live_task_limit(monkeypatch, tmp_path):
|
|
from app.gateway import scheduled_task_access as access
|
|
from deerflow.config.database_config import DatabaseConfig
|
|
from deerflow.persistence.engine import close_engine, get_session_factory, init_engine_from_config
|
|
from deerflow.persistence.scheduled_tasks import ScheduledTaskRepository
|
|
|
|
await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))
|
|
try:
|
|
prepare, *_ = _setup(monkeypatch, original_text=" ".join(f"note-{index}" for index in range(11)))
|
|
repo = ScheduledTaskRepository(get_session_factory())
|
|
monkeypatch.setattr(access, "get_scheduled_task_repo", lambda _request: repo)
|
|
capability = await prepare()
|
|
definition = {"title": "Report", "prompt": "Write report", "schedule_type": "interval", "schedule_spec": {"every_seconds": 3600}, "timezone": "UTC"}
|
|
created = (await capability.manage(action="create", request=definition))["task"]
|
|
for index in range(10):
|
|
result = await capability.manage(action="note", request={"task_id": created["id"], "note": f"note-{index}"})
|
|
assert "error" not in result
|
|
result = await capability.manage(action="note", request={"task_id": created["id"], "note": "note-10"})
|
|
assert (result["status_code"], result["code"], result["params"]) == (422, "note_limit_reached", {"limit": 10})
|
|
stored = await repo.get(created["id"], user_id="alice")
|
|
assert len(stored["standing_notes"]) == 10
|
|
for _ in range(19):
|
|
assert "error" not in await capability.manage(action="create", request=definition)
|
|
result = await capability.manage(action="create", request=definition)
|
|
assert (result["status_code"], result["code"]) == (409, "task_quota_exceeded")
|
|
assert "20 live" in result["error"]
|
|
assert len(await repo.list_by_origin_thread("alice", "origin")) == 20
|
|
finally:
|
|
await close_engine()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_capability_and_shared_validation_offload_configuration_and_timezone_io(monkeypatch, tmp_path):
|
|
from support.detectors.blocking_io_runtime import detect_blocking_io_strict
|
|
|
|
from app.gateway import scheduled_task_access as access
|
|
from app.gateway import scheduled_task_validation as validation
|
|
|
|
prepare, *_ = _setup(monkeypatch)
|
|
config = access.get_config()
|
|
marker = tmp_path / "configuration-and-timezone-fixture.txt"
|
|
marker.write_text("fixture", encoding="utf-8")
|
|
original_validate_timezone = validation.validate_timezone
|
|
|
|
def read_config():
|
|
marker.read_text(encoding="utf-8")
|
|
return config
|
|
|
|
def read_timezone(timezone):
|
|
marker.read_text(encoding="utf-8")
|
|
return original_validate_timezone(timezone)
|
|
|
|
monkeypatch.setattr(access, "get_config", read_config)
|
|
monkeypatch.setattr(validation, "validate_timezone", read_timezone)
|
|
with detect_blocking_io_strict():
|
|
capability = await prepare()
|
|
result = await capability.manage(action="create", request={"title": "Report", "prompt": "Write report", "schedule_type": "interval", "schedule_spec": {"every_seconds": 3600}, "timezone": "UTC"})
|
|
assert "error" not in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scheduled_grant_for_a_page_created_task_stops_only_its_own_occurrence(monkeypatch):
|
|
prepare, repo, occurrences, *_ = _setup(monkeypatch, source=AUTH_SOURCE_INTERNAL, mode="scheduled", task=_task(origin=None))
|
|
capability = await prepare(scheduled_task_runtime={"user_id": "alice", "task_id": "task-1", "occurrence_id": "occurrence-1"})
|
|
assert capability is not None
|
|
result = await capability.stop_current_schedule()
|
|
assert result["stop_requested"] is True
|
|
occurrences.request_stop.assert_awaited_once_with("occurrence-1", task_id="task-1", run_id="run-1", user_id="alice")
|
|
managed = await capability.manage(action="pause", request={"task_id": "task-1"})
|
|
assert (managed["status_code"], managed["code"]) == (403, "interactive_run_required")
|
|
repo.pause_with_queue_cancellation.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scheduled_grant_cannot_stop_another_task(monkeypatch):
|
|
prepare, repo, occurrences, *_ = _setup(monkeypatch, source=AUTH_SOURCE_INTERNAL, mode="scheduled", task=_task(origin=None))
|
|
capability = await prepare(scheduled_task_runtime={"user_id": "alice", "task_id": "task-1", "occurrence_id": "occurrence-1"})
|
|
repo.get.return_value = {**_task(origin=None), "id": "task-2"}
|
|
result = await capability.stop_current_schedule()
|
|
assert (result["status_code"], result["code"]) == (404, "task_not_found")
|
|
occurrences.request_stop.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_capability_pause_of_a_finished_task_is_coded(monkeypatch):
|
|
prepare, repo, *_ = _setup(monkeypatch)
|
|
repo.pause_with_queue_cancellation.return_value = "finished"
|
|
capability = await prepare()
|
|
result = await capability.manage(action="pause", request={"task_id": "task-1"})
|
|
assert result == {"error": "Scheduled task has finished; resume it to run again", "code": "task_finished", "status_code": 409}
|
|
|
|
|
|
# --- WP2: conversation update/resume, run-conversation authority, results --------------------------
|
|
|
|
_INTERVAL = {"title": "Report", "prompt": "Write the report", "schedule_type": "interval", "schedule_spec": {"every_seconds": 3600}}
|
|
|
|
|
|
async def _real_repos(monkeypatch, tmp_path, **setup):
|
|
"""``_setup`` backed by real SQLite repositories (owner ``alice``, conversation ``origin``)."""
|
|
from app.gateway import scheduled_task_access as access
|
|
from deerflow.config.database_config import DatabaseConfig
|
|
from deerflow.persistence.engine import get_session_factory, init_engine_from_config
|
|
from deerflow.persistence.scheduled_task_runs import ScheduledTaskRunRepository
|
|
from deerflow.persistence.scheduled_tasks import ScheduledTaskRepository
|
|
|
|
await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))
|
|
prepare, _, _, service, threads, *_ = _setup(monkeypatch, **setup)
|
|
sf = get_session_factory()
|
|
repo, runs = ScheduledTaskRepository(sf), ScheduledTaskRunRepository(sf)
|
|
monkeypatch.setattr(access, "get_scheduled_task_repo", lambda _request: repo)
|
|
monkeypatch.setattr(access, "get_scheduled_task_run_repo", lambda _request: runs)
|
|
return prepare, sf, repo, runs, service, threads
|
|
|
|
|
|
async def _seed_task(repo, task_id="task-1", *, origin_thread_id="origin", schedule_spec=None, **extra):
|
|
from _scheduled_rows import BASE
|
|
|
|
return await repo.create(
|
|
task_id=task_id,
|
|
user_id="alice",
|
|
thread_id=None,
|
|
context_mode="fresh_thread_per_run",
|
|
assistant_id="lead_agent",
|
|
title=task_id,
|
|
prompt="Summarize the notifications.",
|
|
schedule_type="interval",
|
|
schedule_spec=schedule_spec or {"every_seconds": 3600},
|
|
timezone="UTC",
|
|
next_run_at=BASE,
|
|
origin_thread_id=origin_thread_id,
|
|
**extra,
|
|
)
|
|
|
|
|
|
async def _seed_runs(sf, *rows):
|
|
async with sf() as session:
|
|
session.add_all(list(rows))
|
|
await session.commit()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_keeps_task_id_and_never_deletes(monkeypatch, tmp_path):
|
|
from _scheduled_rows import occurrence
|
|
|
|
from deerflow.persistence.engine import close_engine
|
|
|
|
prepare, sf, repo, runs, *_ = await _real_repos(monkeypatch, tmp_path)
|
|
try:
|
|
await _seed_task(repo)
|
|
await _seed_runs(sf, occurrence("trial", "task-1", seq=None, trigger="manual", accounted=None))
|
|
capability = await prepare(client_timezone="Asia/Shanghai")
|
|
result = await capability.manage(action="update", request={"task_id": "task-1", "schedule_type": "cron", "schedule_spec": {"cron": "0 10 * * *"}, "timezone": "Asia/Shanghai"})
|
|
assert result["action"] == "update"
|
|
assert result["task"]["id"] == "task-1"
|
|
assert result["task"]["schedule_spec"] == {"cron": "0 10 * * *"}
|
|
assert result["task"]["next_run_local"].endswith("10:00 (Asia/Shanghai)")
|
|
assert [task["id"] for task in await repo.list_by_user("alice")] == ["task-1"]
|
|
assert len(await runs.list_by_task("task-1")) == 1
|
|
finally:
|
|
await close_engine()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_and_update_store_stop_condition_separately(monkeypatch):
|
|
prepare, repo, *_ = _setup(monkeypatch)
|
|
capability = await prepare(client_timezone="Asia/Shanghai")
|
|
created = await capability.manage(action="create", request={**_INTERVAL, "prompt": "Check release-checklist.md", "stop_condition": "every item\n is checked", "max_runs": 5})
|
|
values = repo.create.await_args.kwargs
|
|
assert values["prompt"] == "Check release-checklist.md"
|
|
assert values["stop_condition"] == "every item is checked"
|
|
assert created["task"]["prompt"] == "Check release-checklist.md"
|
|
assert created["task"]["stop_condition"] == "every item is checked"
|
|
repo.get.return_value = _task(stop_condition="every item is checked")
|
|
updated = await capability.manage(action="update", request={"task_id": "task-1", "clear_fields": ["stop_condition"]})
|
|
updates = repo.update.await_args.kwargs["updates"]
|
|
assert updates["stop_condition"] is None
|
|
assert "prompt" not in updates
|
|
assert updated["task"]["stop_condition"] is None
|
|
assert updated["task"]["prompt"] == "Write the report"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
("task_updates", "fields"),
|
|
[
|
|
({}, {"schedule_spec": {"every_seconds": 600}}),
|
|
({"timezone": "Asia/Shanghai"}, {"end_at": "2020-01-01T00:00:00"}),
|
|
({}, {"timezone": "missing/zone"}),
|
|
({}, {"timezone": "Europe"}),
|
|
],
|
|
ids=["sub-hourly-without-cap", "end-at-past", "invalid-timezone", "timezone-directory"],
|
|
)
|
|
async def test_update_shares_rest_validation_codes(monkeypatch, task_updates, fields):
|
|
from fastapi import HTTPException
|
|
|
|
prepare, repo, _, _, threads, _, _, request = _setup(monkeypatch, task=_task(**task_updates))
|
|
capability = await prepare()
|
|
tool_result = await capability.manage(action="update", request={"task_id": "task-1", **fields})
|
|
monkeypatch.setattr(scheduled_tasks, "get_scheduled_task_repo", lambda _request: repo)
|
|
monkeypatch.setattr(scheduled_tasks, "get_thread_store", lambda _request: threads)
|
|
monkeypatch.setattr(scheduled_tasks, "get_config", lambda: AppConfig.model_validate({"sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"}}))
|
|
monkeypatch.setattr(scheduled_tasks, "get_optional_user_from_request", AsyncMock(return_value=_user()))
|
|
with pytest.raises(HTTPException) as error:
|
|
await call_unwrapped(scheduled_tasks.update_scheduled_task, task_id="task-1", request=request, body=scheduled_tasks.ScheduledTaskUpdateRequest(**fields))
|
|
assert tool_result["code"] == error.value.detail["code"]
|
|
assert tool_result["status_code"] == error.value.status_code
|
|
repo.update.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resume_recomputes_and_reports_limits_exhausted_with_renewal(monkeypatch, tmp_path):
|
|
from _scheduled_rows import occurrence
|
|
|
|
from deerflow.persistence.engine import close_engine
|
|
|
|
prepare, sf, repo, *_ = await _real_repos(monkeypatch, tmp_path)
|
|
try:
|
|
await _seed_task(repo, max_runs=2)
|
|
await _seed_runs(sf, occurrence("a1", seq=1), occurrence("a2", seq=2))
|
|
await repo.update("task-1", user_id="alice", updates={"status": "completed", "next_run_at": None})
|
|
capability = await prepare()
|
|
exhausted = await capability.manage(action="resume", request={"task_id": "task-1"})
|
|
assert (exhausted["status_code"], exhausted["code"]) == (409, "limits_exhausted")
|
|
assert exhausted["params"] == {"limit": "max_runs", "used": 2, "max_runs": 2, "end_at": None}
|
|
assert (await repo.get("task-1", user_id="alice"))["status"] == "completed"
|
|
too_low = await capability.manage(action="resume", request={"task_id": "task-1", "max_runs": 2})
|
|
assert (too_low["code"], too_low["params"]) == ("max_runs_not_above_used", {"used": 2})
|
|
resumed = await capability.manage(action="resume", request={"task_id": "task-1", "max_runs": 4})
|
|
assert resumed["action"] == "resume"
|
|
assert (resumed["task"]["status"], resumed["task"]["max_runs"], resumed["task"]["automatic_runs_used"]) == ("enabled", 4, 2)
|
|
stored = await repo.get("task-1", user_id="alice")
|
|
# No catch-up run: the next run is computed from now.
|
|
assert datetime.fromisoformat(stored["next_run_at"]).astimezone(UTC) > datetime.now(UTC) + timedelta(minutes=55)
|
|
assert resumed["task"]["next_run_local"].endswith("(UTC)")
|
|
finally:
|
|
await close_engine()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_conversation_can_manage_its_task(monkeypatch, tmp_path):
|
|
from _scheduled_rows import occurrence
|
|
|
|
from deerflow.persistence.engine import close_engine
|
|
|
|
# This turn runs in "origin", which is a run conversation of a page-created task.
|
|
prepare, sf, repo, *_ = await _real_repos(monkeypatch, tmp_path)
|
|
try:
|
|
await _seed_task(repo, origin_thread_id=None)
|
|
await _seed_runs(sf, occurrence("a1", seq=1, thread_id="origin"))
|
|
capability = await prepare()
|
|
listed = await capability.manage(action="list", request={})
|
|
assert [task["id"] for task in listed["tasks"]] == ["task-1"]
|
|
assert listed["tasks"][0]["last_run"]["status"] == "success"
|
|
paused = await capability.manage(action="pause", request={"task_id": "task-1"})
|
|
assert (paused["action"], paused["task"]["status"]) == ("pause", "paused")
|
|
finally:
|
|
await close_engine()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("case", ["foreign-owner", "unrelated-thread"])
|
|
async def test_foreign_or_unrelated_run_thread_grants_nothing(monkeypatch, tmp_path, case):
|
|
from _scheduled_rows import create_task, occurrence
|
|
|
|
from deerflow.persistence.engine import close_engine
|
|
|
|
prepare, sf, repo, *_ = await _real_repos(monkeypatch, tmp_path)
|
|
try:
|
|
if case == "foreign-owner":
|
|
await create_task(repo, "task-1", user_id="bob")
|
|
await _seed_runs(sf, occurrence("a1", seq=1, thread_id="origin"))
|
|
else:
|
|
await _seed_task(repo, origin_thread_id=None)
|
|
await _seed_runs(sf, occurrence("a1", seq=1, thread_id="other-thread"))
|
|
capability = await prepare()
|
|
assert (await capability.manage(action="list", request={}))["tasks"] == []
|
|
for action in ("pause", "update", "resume", "delete", "trial"):
|
|
result = await capability.manage(action=action, request={"task_id": "task-1"})
|
|
assert (result["status_code"], result["code"]) == (404, "task_not_found"), action
|
|
assert (await repo.get_internal("task-1"))["status"] == "enabled"
|
|
finally:
|
|
await close_engine()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("action", ["list", "update", "resume", "pause", "create"])
|
|
async def test_scheduled_mode_still_cannot_manage(monkeypatch, action):
|
|
prepare, repo, occurrences, *_ = _setup(monkeypatch, source=AUTH_SOURCE_INTERNAL, mode="scheduled")
|
|
occurrences.task_ids_for_thread.return_value = ["task-1"]
|
|
capability = await prepare(scheduled_task_runtime={"user_id": "alice", "task_id": "task-1", "occurrence_id": "occurrence-1"})
|
|
result = await capability.manage(action=action, request={"task_id": "task-1"})
|
|
assert (result["status_code"], result["code"]) == (403, "interactive_run_required")
|
|
repo.update.assert_not_awaited()
|
|
repo.pause_with_queue_cancellation.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_uses_browser_timezone_by_default(monkeypatch):
|
|
prepare, repo, *_ = _setup(monkeypatch)
|
|
capability = await prepare(client_timezone="Asia/Shanghai")
|
|
result = await capability.manage(action="create", request={**_INTERVAL, "schedule_type": "cron", "schedule_spec": {"cron": "0 9 * * 1-5"}})
|
|
assert repo.create.await_args.kwargs["timezone"] == "Asia/Shanghai"
|
|
assert (result["task"]["timezone"], result["task"]["timezone_source"]) == ("Asia/Shanghai", "browser_default")
|
|
assert "browser_timezone" not in result["task"]
|
|
assert result["task"]["next_run_local"].endswith("09:00 (Asia/Shanghai)")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_cron_without_any_timezone_asks(monkeypatch):
|
|
prepare, repo, *_ = _setup(monkeypatch, source=AUTH_SOURCE_INTERNAL)
|
|
capability = await prepare()
|
|
result = await capability.manage(action="create", request={**_INTERVAL, "schedule_type": "cron", "schedule_spec": {"cron": "0 9 * * *"}})
|
|
assert (result["status_code"], result["code"]) == (422, "timezone_required")
|
|
assert result["error"] == "Ask the user which timezone to use."
|
|
assert result["params"] == {"now_local": None}
|
|
repo.create.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_interval_without_any_timezone_needs_none(monkeypatch):
|
|
prepare, repo, *_ = _setup(monkeypatch, source=AUTH_SOURCE_INTERNAL)
|
|
capability = await prepare()
|
|
result = await capability.manage(action="create", request={**_INTERVAL, "schedule_spec": {"every_seconds": 1800}, "max_runs": 5})
|
|
assert "error" not in result
|
|
assert repo.create.await_args.kwargs["timezone"] == "UTC"
|
|
assert (result["task"]["timezone"], result["task"]["timezone_source"]) == ("UTC", "not_needed")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_once_with_offset_needs_no_timezone(monkeypatch):
|
|
prepare, repo, *_ = _setup(monkeypatch, source=AUTH_SOURCE_INTERNAL)
|
|
capability = await prepare()
|
|
run_at = (datetime.now(UTC) + timedelta(days=1)).astimezone(ZoneInfo("Asia/Shanghai")).replace(microsecond=0).isoformat()
|
|
result = await capability.manage(action="create", request={**_INTERVAL, "schedule_type": "once", "schedule_spec": {"run_at": run_at}})
|
|
assert "error" not in result
|
|
assert result["task"]["timezone_source"] == "not_needed"
|
|
assert repo.create.await_args.kwargs["next_run_at"] == datetime.fromisoformat(run_at)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_once_local_time_without_any_timezone_asks(monkeypatch):
|
|
prepare, repo, *_ = _setup(monkeypatch, source=AUTH_SOURCE_INTERNAL)
|
|
capability = await prepare()
|
|
run_at = (datetime.now(UTC) + timedelta(days=1)).replace(microsecond=0, tzinfo=None).isoformat()
|
|
result = await capability.manage(action="create", request={**_INTERVAL, "schedule_type": "once", "schedule_spec": {"run_at": run_at}})
|
|
assert result["code"] == "timezone_required"
|
|
repo.create.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_explicit_timezone_wins(monkeypatch):
|
|
prepare, repo, *_ = _setup(monkeypatch)
|
|
capability = await prepare(client_timezone="Europe/Berlin")
|
|
result = await capability.manage(action="create", request={**_INTERVAL, "schedule_type": "cron", "schedule_spec": {"cron": "0 9 * * *"}, "timezone": "Asia/Shanghai"})
|
|
assert repo.create.await_args.kwargs["timezone"] == "Asia/Shanghai"
|
|
assert (result["task"]["timezone_source"], result["task"]["browser_timezone"]) == ("explicit", "Europe/Berlin")
|
|
same = await capability.manage(action="create", request={**_INTERVAL, "schedule_type": "cron", "schedule_spec": {"cron": "0 9 * * *"}, "timezone": "Europe/Berlin"})
|
|
assert same["task"]["timezone_source"] == "explicit"
|
|
assert "browser_timezone" not in same["task"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_keeps_saved_timezone(monkeypatch):
|
|
prepare, repo, *_ = _setup(monkeypatch, task=_task(schedule_type="cron", schedule_spec={"cron": "0 9 * * *"}, timezone="Asia/Shanghai"))
|
|
capability = await prepare(client_timezone="Europe/Berlin")
|
|
result = await capability.manage(action="update", request={"task_id": "task-1", "schedule_spec": {"cron": "0 10 * * *"}})
|
|
updates = repo.update.await_args.kwargs["updates"]
|
|
assert "timezone" not in updates
|
|
assert datetime.fromisoformat(str(updates["next_run_at"])).astimezone(ZoneInfo("Asia/Shanghai")).hour == 10
|
|
assert (result["task"]["timezone"], result["task"]["timezone_source"]) == ("Asia/Shanghai", "saved")
|
|
assert "browser_timezone" not in result["task"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_from_a_zone_free_schedule_to_local_time_asks_for_a_timezone(monkeypatch):
|
|
prepare, repo, *_ = _setup(monkeypatch, task=_task(timezone="UTC"))
|
|
capability = await prepare(client_timezone="Asia/Shanghai")
|
|
result = await capability.manage(action="update", request={"task_id": "task-1", "schedule_type": "cron", "schedule_spec": {"cron": "0 9 * * *"}})
|
|
assert result["code"] == "timezone_required"
|
|
# "Now" is reported in the browser zone, never in the UTC placeholder.
|
|
assert result["params"]["now_local"].endswith("(Asia/Shanghai)")
|
|
repo.update.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_naive_end_at_without_any_timezone_asks(monkeypatch):
|
|
prepare, repo, *_ = _setup(monkeypatch, source=AUTH_SOURCE_INTERNAL)
|
|
capability = await prepare()
|
|
end_local = (datetime.now(UTC) + timedelta(days=2)).replace(microsecond=0, tzinfo=None).isoformat()
|
|
created = await capability.manage(action="create", request={**_INTERVAL, "end_at": end_local})
|
|
assert (created["code"], created["params"]) == ("timezone_required", {"now_local": None})
|
|
repo.create.assert_not_awaited()
|
|
with_offset = await capability.manage(action="create", request={**_INTERVAL, "end_at": f"{end_local}+08:00"})
|
|
assert with_offset["task"]["timezone_source"] == "not_needed"
|
|
# An interval task saved with the UTC placeholder: a local end time asks too.
|
|
updated = await capability.manage(action="update", request={"task_id": "task-1", "end_at": end_local})
|
|
assert updated["code"] == "timezone_required"
|
|
# Resume takes no timezone: it asks too, and an end time with an offset passes.
|
|
resumed = await capability.manage(action="resume", request={"task_id": "task-1", "end_at": end_local})
|
|
assert resumed["code"] == "timezone_required"
|
|
assert "code" not in await capability.manage(action="resume", request={"task_id": "task-1", "end_at": f"{end_local}+08:00"})
|
|
repo.update.assert_not_awaited()
|
|
named = await capability.manage(action="update", request={"task_id": "task-1", "end_at": end_local, "timezone": "Asia/Shanghai"})
|
|
assert named["task"]["timezone_source"] == "explicit"
|
|
assert repo.update.await_args.kwargs["updates"]["end_at"] == datetime.fromisoformat(end_local).replace(tzinfo=ZoneInfo("Asia/Shanghai")).astimezone(UTC)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_and_update_results_have_now_local(monkeypatch):
|
|
prepare, *_ = _setup(monkeypatch)
|
|
capability = await prepare(client_timezone="Asia/Shanghai")
|
|
created = await capability.manage(action="create", request=_INTERVAL)
|
|
assert created["now_local"].endswith("(Asia/Shanghai)")
|
|
updated = await capability.manage(action="update", request={"task_id": "task-1", "title": "Weekly report"})
|
|
assert updated["now_local"].endswith("(UTC)")
|
|
past = await capability.manage(action="create", request={**_INTERVAL, "schedule_type": "once", "schedule_spec": {"run_at": "2020-01-01T09:00:00"}})
|
|
assert past["code"] == "once_in_past"
|
|
assert past["params"]["now_local"].endswith("(Asia/Shanghai)")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_naive_end_at_uses_task_timezone(monkeypatch):
|
|
prepare, repo, *_ = _setup(monkeypatch, task=_task(timezone="Asia/Shanghai"))
|
|
capability = await prepare(client_timezone="Asia/Shanghai")
|
|
end_local = (datetime.now(ZoneInfo("Asia/Shanghai")) + timedelta(days=30)).replace(hour=20, minute=0, second=0, microsecond=0)
|
|
naive = end_local.replace(tzinfo=None).isoformat()
|
|
created = await capability.manage(action="create", request={**_INTERVAL, "end_at": naive})
|
|
assert repo.create.await_args.kwargs["end_at"] == end_local.astimezone(UTC)
|
|
assert created["task"]["end_at_local"] == f"{end_local:%Y-%m-%d} 20:00 (Asia/Shanghai)"
|
|
await capability.manage(action="update", request={"task_id": "task-1", "end_at": naive})
|
|
assert repo.update.await_args.kwargs["updates"]["end_at"] == end_local.astimezone(UTC)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resume_clear_fields_removes_cap_and_reactivates(monkeypatch, tmp_path):
|
|
from _scheduled_rows import occurrence
|
|
|
|
from deerflow.persistence.engine import close_engine
|
|
|
|
prepare, sf, repo, *_ = await _real_repos(monkeypatch, tmp_path)
|
|
try:
|
|
# Page-created, run conversation "origin": no sub-hourly cap rule.
|
|
await _seed_task(repo, origin_thread_id=None, max_runs=1)
|
|
await _seed_runs(sf, occurrence("a1", seq=1, thread_id="origin"))
|
|
await repo.update("task-1", user_id="alice", updates={"status": "completed", "next_run_at": None})
|
|
capability = await prepare()
|
|
resumed = await capability.manage(action="resume", request={"task_id": "task-1", "clear_fields": ["max_runs"]})
|
|
assert (resumed["task"]["status"], resumed["task"]["max_runs"]) == ("enabled", None)
|
|
finally:
|
|
await close_engine()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resume_clear_fields_keeps_the_cap_rule_for_sub_hourly_chat_tasks(monkeypatch, tmp_path):
|
|
from _scheduled_rows import occurrence
|
|
|
|
from deerflow.persistence.engine import close_engine
|
|
|
|
prepare, sf, repo, *_ = await _real_repos(monkeypatch, tmp_path)
|
|
try:
|
|
await _seed_task(repo, schedule_spec={"every_seconds": 600}, max_runs=1)
|
|
await _seed_runs(sf, occurrence("a1", seq=1))
|
|
await repo.update("task-1", user_id="alice", updates={"status": "completed", "next_run_at": None})
|
|
before = await repo.get("task-1", user_id="alice")
|
|
capability = await prepare()
|
|
result = await capability.manage(action="resume", request={"task_id": "task-1", "clear_fields": ["max_runs"]})
|
|
assert (result["status_code"], result["code"]) == (422, "frequent_requires_limit")
|
|
assert await repo.get("task-1", user_id="alice") == before
|
|
finally:
|
|
await close_engine()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_clear_fields_are_bounded_per_action(monkeypatch):
|
|
prepare, repo, *_ = _setup(monkeypatch)
|
|
capability = await prepare()
|
|
resume = await capability.manage(action="resume", request={"task_id": "task-1", "clear_fields": ["stop_condition"]})
|
|
assert resume["code"] == "unsupported_fields"
|
|
both = await capability.manage(action="update", request={"task_id": "task-1", "max_runs": 3, "clear_fields": ["max_runs"]})
|
|
assert both["code"] == "invalid_request"
|
|
repo.update.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trial_existing_queued_is_reported(monkeypatch):
|
|
prepare, _, _, service, *_ = _setup(monkeypatch, original_text="Run it now")
|
|
service.dispatch_task.return_value = {"outcome": "queued", "existing": True, "thread_id": "queued-thread", "run_id": None}
|
|
capability = await prepare()
|
|
result = await capability.manage(action="trial", request={"task_id": "task-1"})
|
|
assert result["trial"] == {"outcome": "queued", "existing": True, "thread_id": "queued-thread"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trial_result_has_thread_id(monkeypatch):
|
|
prepare, *_ = _setup(monkeypatch, original_text="先跑一次")
|
|
capability = await prepare()
|
|
result = await capability.manage(action="trial", request={"task_id": "task-1"})
|
|
assert result["action"] == "trial"
|
|
assert result["trial"] == {"outcome": "launched", "existing": False, "thread_id": "trial-thread"}
|
|
assert result["task"]["id"] == "task-1"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"user_text",
|
|
[
|
|
"Sure, run it now",
|
|
"OK, run it now",
|
|
"Run the task now",
|
|
"Run this task now please",
|
|
"好的，先跑一次",
|
|
"先跑一次吧",
|
|
"可以，先试一下吧",
|
|
"试跑一下",
|
|
"那就先跑一次看看",
|
|
"Go ahead and run it now!",
|
|
"好的，请先试跑一次。",
|
|
"Try it out",
|
|
"立即试运行",
|
|
],
|
|
)
|
|
async def test_trial_accepts_natural_direct_requests(monkeypatch, user_text):
|
|
prepare, _, _, service, *_ = _setup(monkeypatch, original_text=user_text)
|
|
capability = await prepare()
|
|
result = await capability.manage(action="trial", request={"task_id": "task-1"})
|
|
assert result["trial"]["outcome"] == "launched"
|
|
assert service.dispatch_task.await_args.kwargs["trigger"] == "manual"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(("source", "display"), [(AUTH_SOURCE_SESSION, "card"), (AUTH_SOURCE_INTERNAL, "text")])
|
|
async def test_results_are_display_card_for_session_and_text_for_internal(monkeypatch, source, display):
|
|
prepare, *_ = _setup(monkeypatch, source=source, original_text="Run it now")
|
|
capability = await prepare(client_timezone="UTC")
|
|
for action, request in (("create", _INTERVAL), ("list", {}), ("pause", {"task_id": "task-1"}), ("trial", {"task_id": "task-1"})):
|
|
result = await capability.manage(action=action, request=request)
|
|
assert (result["action"], result["display"]) == (action, display)
|
|
|
|
|
|
def _keys(value):
|
|
if isinstance(value, dict):
|
|
return set(value) | {key for item in value.values() for key in _keys(item)}
|
|
if isinstance(value, list):
|
|
return {key for item in value for key in _keys(item)}
|
|
return set()
|
|
|
|
|
|
async def _every_action_result(monkeypatch):
|
|
task = _task(secret="must-not-leak", lease_owner="worker-1", lease_expires_at=_FUTURE, last_run_id="run-x", end_at=_FUTURE, max_runs=5, stop_condition="all done")
|
|
prepare, repo, occurrences, *_ = _setup(monkeypatch, original_text="Run it now; use develop", task=task)
|
|
repo.automatic_runs_used_for.return_value = {"task-1": 2}
|
|
occurrences.list_by_task.return_value = [{"status": "unmet", "trigger": "scheduled", "finished_at": _FUTURE, "error": None, "goal_verdict": {"blocker": "goal_not_met_yet", "reason": "no"}, "run_id": "run-x"}]
|
|
capability = await prepare(client_timezone="Asia/Shanghai")
|
|
results = [
|
|
await capability.manage(action="create", request={**_INTERVAL, "stop_condition": "all done", "max_runs": 5}),
|
|
await capability.manage(action="update", request={"task_id": "task-1", "title": "Weekly"}),
|
|
await capability.manage(action="list", request={}),
|
|
await capability.manage(action="pause", request={"task_id": "task-1"}),
|
|
await capability.manage(action="note", request={"task_id": "task-1", "note": "use develop"}),
|
|
await capability.manage(action="trial", request={"task_id": "task-1"}),
|
|
await capability.manage(action="delete", request={"task_id": "task-1"}),
|
|
await capability.manage(action="create", request={**_INTERVAL, "schedule_type": "once", "schedule_spec": {"run_at": "2020-01-01T09:00:00"}}),
|
|
]
|
|
repo.get.return_value = {**task, "status": "paused"}
|
|
results.append(await capability.manage(action="resume", request={"task_id": "task-1"}))
|
|
return results
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_every_action_result_is_json_serializable(monkeypatch):
|
|
import json
|
|
|
|
results = await _every_action_result(monkeypatch)
|
|
for result in results:
|
|
assert json.loads(json.dumps(result)) == result
|
|
assert results[-2]["code"] == "once_in_past"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_results_have_no_utc_or_lease_fields(monkeypatch):
|
|
results = await _every_action_result(monkeypatch)
|
|
forbidden = {"next_run_at", "end_at", "user_id", "lease_owner", "lease_expires_at", "stop_instructions", "origin_thread_id", "last_run_id", "secret"}
|
|
for result in results:
|
|
assert _keys(result).isdisjoint(forbidden), result
|
|
listed = results[2]["tasks"][0]
|
|
assert listed["last_run"] == {"status": "unmet", "trigger": "scheduled", "finished_local": listed["last_run"]["finished_local"], "reason_code": "goal_not_met_yet"}
|
|
assert listed["last_run"]["finished_local"].endswith("(UTC)")
|
|
assert (listed["automatic_runs_used"], listed["end_at_local"].endswith("(UTC)")) == (2, True)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_reports_last_run_and_no_next_run_when_paused(monkeypatch, tmp_path):
|
|
from _scheduled_rows import occurrence
|
|
|
|
from deerflow.persistence.engine import close_engine
|
|
|
|
prepare, sf, repo, *_ = await _real_repos(monkeypatch, tmp_path)
|
|
try:
|
|
await _seed_task(repo, "task-1")
|
|
await _seed_task(repo, "task-2")
|
|
await _seed_runs(sf, occurrence("a1", seq=1, status="failed", error="interrupted: gateway restarted before the run reached a terminal state"), occurrence("a2", seq=2, status="unmet", error="evaluator_failed"))
|
|
await repo.update("task-2", user_id="alice", updates={"status": "paused"})
|
|
capability = await prepare(client_timezone="Asia/Shanghai")
|
|
listed = await capability.manage(action="list", request={})
|
|
by_id = {task["id"]: task for task in listed["tasks"]}
|
|
assert by_id["task-1"]["last_run"]["status"] == "unmet"
|
|
assert by_id["task-1"]["last_run"]["reason_code"] == "evaluator_failed"
|
|
assert by_id["task-1"]["next_run_local"] is not None
|
|
assert by_id["task-2"]["status"] == "paused"
|
|
assert by_id["task-2"]["next_run_local"] is None
|
|
assert by_id["task-2"]["last_run"] is None
|
|
assert listed["default_timezone"] == "Asia/Shanghai"
|
|
assert listed["now_local"].endswith("(Asia/Shanghai)")
|
|
finally:
|
|
await close_engine()
|
|
|
|
|
|
def test_host_run_errors_reach_the_agent_as_short_keys():
|
|
from app.gateway.scheduled_task_access import _reason_code
|
|
from deerflow.scheduler.host_notes import RUN_ERROR_QUEUE_TIMEOUT
|
|
|
|
assert _reason_code({"status": "failed", "error": RUN_ERROR_QUEUE_TIMEOUT}) == "queue_timeout"
|
|
assert _reason_code({"status": "unmet", "error": "blocked:external_wait"}) == "blocked:external_wait"
|
|
assert _reason_code({"status": "failed", "error": "boom: Traceback ...", "run_id": None}) == "launch_failed"
|
|
assert _reason_code({"status": "success"}) is None
|