zhaoxi
8f1398c591
新增persona_template表和CRUD API，BaseIndividualModel增加node_affinity和template_origin_id字段， WorkerCluster支持多集群Ray资源调度，环境变量收敛到pydantic-settings统一校验， 数据库异常转换为结构化BusinessError/RetryableError，系统节点支持custom_system_prompt。 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
74 lines
1.8 KiB
Python
74 lines
1.8 KiB
Python
"""``database_exception`` 装饰器：透传异常并通过 logger 上报。"""
|
|
|
|
import pytest
|
|
from pydantic import BaseModel, ValidationError
|
|
from sqlalchemy.exc import IntegrityError, OperationalError
|
|
|
|
from kilostar.core.postgres_database.database_exception import database_exception
|
|
from kilostar.utils.error import UserNotExistError, BusinessError, RetryableError
|
|
|
|
|
|
async def test_normal_path_returns_value():
|
|
@database_exception
|
|
async def ok() -> int:
|
|
return 42
|
|
|
|
assert await ok() == 42
|
|
|
|
|
|
async def test_passes_args_and_kwargs():
|
|
@database_exception
|
|
async def add(a, b, *, c) -> int:
|
|
return a + b + c
|
|
|
|
assert await add(1, 2, c=3) == 6
|
|
|
|
|
|
async def test_validation_error_propagates():
|
|
class Model(BaseModel):
|
|
x: int
|
|
|
|
@database_exception
|
|
async def boom() -> None:
|
|
Model(x="not-int") # type: ignore[arg-type]
|
|
|
|
with pytest.raises(ValidationError):
|
|
await boom()
|
|
|
|
|
|
async def test_integrity_error_becomes_business_error():
|
|
@database_exception
|
|
async def boom() -> None:
|
|
raise IntegrityError("stmt", {}, Exception("dup"))
|
|
|
|
with pytest.raises(BusinessError) as exc_info:
|
|
await boom()
|
|
assert exc_info.value.http_status == 409
|
|
|
|
|
|
async def test_operational_error_becomes_retryable():
|
|
@database_exception
|
|
async def boom() -> None:
|
|
raise OperationalError("stmt", {}, Exception("conn"))
|
|
|
|
with pytest.raises(RetryableError):
|
|
await boom()
|
|
|
|
|
|
async def test_user_not_exist_error_propagates():
|
|
@database_exception
|
|
async def boom() -> None:
|
|
raise UserNotExistError("missing")
|
|
|
|
with pytest.raises(UserNotExistError):
|
|
await boom()
|
|
|
|
|
|
async def test_unexpected_error_propagates():
|
|
@database_exception
|
|
async def boom() -> None:
|
|
raise RuntimeError("unexpected")
|
|
|
|
with pytest.raises(RuntimeError):
|
|
await boom()
|