吉鑫升andClaude Opus 4.7
358d5c72a6
- 业务 ORM 数据支持 MySQL 后端(asyncmy),新增 PortableJSON/
BeijingDateTime 可移植列类型、迁移脚本与切库脚本
- 修复 GET /api/scheduled-tasks/runs/{id} 的 _load_run 无限递归
(误调用自身,应委托 service.get_run)导致的 RecursionError 500
- 定时任务名按用户唯一:create/update/admin_update 主动校验,
冲突抛 DuplicateTaskNameError,路由转 409 而非裸 IntegrityError 500
- 补充回归测试 test_scheduled_task_run_lookup / _name_uniqueness
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
81 lines
2.6 KiB
Python
81 lines
2.6 KiB
Python
"""Regression tests for the scheduled-task run lookup helper.
|
|
|
|
Guards against a recursion bug where ``_load_run`` called itself instead of
|
|
delegating to ``service.get_run`` — every ``GET /api/scheduled-tasks/runs/{id}``
|
|
blew the stack with ``RecursionError`` and returned 500.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from app.gateway.routers import scheduled_tasks as st
|
|
|
|
|
|
class _FakeService:
|
|
"""Records delegation so the test can assert no self-recursion happened."""
|
|
|
|
def __init__(self, *, owner_run=None, admin_run=None) -> None:
|
|
self._owner_run = owner_run
|
|
self._admin_run = admin_run
|
|
self.get_run_calls: list[tuple[str, str]] = []
|
|
self.admin_get_run_calls: list[str] = []
|
|
|
|
async def get_run(self, user_id: str, run_id: str):
|
|
self.get_run_calls.append((user_id, run_id))
|
|
return self._owner_run
|
|
|
|
async def admin_get_run(self, run_id: str):
|
|
self.admin_get_run_calls.append(run_id)
|
|
return self._admin_run
|
|
|
|
|
|
class _User:
|
|
def __init__(self, system_role: str | None) -> None:
|
|
self.system_role = system_role
|
|
|
|
|
|
def _patch_user(monkeypatch, user) -> None:
|
|
async def _fake(_request):
|
|
return user
|
|
|
|
monkeypatch.setattr(st, "get_optional_user_from_request", _fake)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_load_run_returns_owner_run_without_recursing(monkeypatch):
|
|
"""Owner lookup hits service.get_run exactly once — no self-recursion."""
|
|
_patch_user(monkeypatch, _User("user"))
|
|
service = _FakeService(owner_run={"id": "r1"})
|
|
|
|
run = await st._load_run(service, object(), "r1", "alice")
|
|
|
|
assert run == {"id": "r1"}
|
|
assert service.get_run_calls == [("alice", "r1")]
|
|
assert service.admin_get_run_calls == [] # admin fallback skipped when found
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_load_run_admin_fallback_when_owner_miss(monkeypatch):
|
|
"""An admin who is not the owner falls back to admin_get_run."""
|
|
_patch_user(monkeypatch, _User("admin"))
|
|
service = _FakeService(owner_run=None, admin_run={"id": "r2"})
|
|
|
|
run = await st._load_run(service, object(), "r2", "bob")
|
|
|
|
assert run == {"id": "r2"}
|
|
assert service.get_run_calls == [("bob", "r2")]
|
|
assert service.admin_get_run_calls == ["r2"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_load_run_non_admin_miss_returns_none(monkeypatch):
|
|
"""A non-admin who is not the owner gets None (router turns it into 404)."""
|
|
_patch_user(monkeypatch, _User("user"))
|
|
service = _FakeService(owner_run=None, admin_run={"id": "r3"})
|
|
|
|
run = await st._load_run(service, object(), "r3", "carol")
|
|
|
|
assert run is None
|
|
assert service.admin_get_run_calls == [] # no admin fallback for non-admins
|