55 lines
2.2 KiB
Python
55 lines
2.2 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from sqlalchemy import create_engine
|
||
|
|
from sqlalchemy.orm import Session, sessionmaker
|
||
|
|
from sqlalchemy.pool import StaticPool
|
||
|
|
|
||
|
|
from app.core.agent_enums import AgentAssetDomain, AgentAssetStatus
|
||
|
|
from app.db.base import Base
|
||
|
|
from app.models.agent_asset import AgentAsset
|
||
|
|
from app.schemas.agent_asset import AgentAssetRiskRuleGenerateRequest
|
||
|
|
from app.services.agent_asset_rule_library import AgentAssetRuleLibraryManager
|
||
|
|
from app.services.risk_rule_generation_jobs import RiskRuleGenerationJobService
|
||
|
|
|
||
|
|
|
||
|
|
class FailingRuntimeChatService:
|
||
|
|
def complete(self, *args, **kwargs) -> str:
|
||
|
|
raise RuntimeError("Hermes semantic plan failed")
|
||
|
|
|
||
|
|
|
||
|
|
def build_session() -> Session:
|
||
|
|
engine = create_engine(
|
||
|
|
"sqlite+pysqlite:///:memory:",
|
||
|
|
connect_args={"check_same_thread": False},
|
||
|
|
poolclass=StaticPool,
|
||
|
|
)
|
||
|
|
Base.metadata.create_all(bind=engine)
|
||
|
|
return sessionmaker(bind=engine, autoflush=False, autocommit=False)()
|
||
|
|
|
||
|
|
|
||
|
|
def test_background_generation_failure_keeps_error_detail_and_last_operation(tmp_path) -> None:
|
||
|
|
with build_session() as db:
|
||
|
|
manager = AgentAssetRuleLibraryManager(rule_root=tmp_path / "rules")
|
||
|
|
service = RiskRuleGenerationJobService(
|
||
|
|
db,
|
||
|
|
rule_library_manager=manager,
|
||
|
|
runtime_chat_service=FailingRuntimeChatService(),
|
||
|
|
)
|
||
|
|
body = AgentAssetRiskRuleGenerateRequest(
|
||
|
|
business_domain=AgentAssetDomain.EXPENSE,
|
||
|
|
expense_category="travel",
|
||
|
|
rule_title="差旅异常规则",
|
||
|
|
natural_language="差旅报销票据城市与申报目的地不一致时提示风险。",
|
||
|
|
)
|
||
|
|
asset_id = service.enqueue_rule_asset_generation(body, actor="pytest")
|
||
|
|
|
||
|
|
service.complete_rule_asset_generation(asset_id, body, actor="pytest")
|
||
|
|
|
||
|
|
asset = db.get(AgentAsset, asset_id)
|
||
|
|
assert asset is not None
|
||
|
|
assert asset.status == AgentAssetStatus.FAILED.value
|
||
|
|
assert asset.config_json["generation_status"] == AgentAssetStatus.FAILED.value
|
||
|
|
assert asset.config_json["generation_error"] == "Hermes semantic plan failed"
|
||
|
|
assert asset.config_json["last_operation"]["action"] == "generation_failed"
|
||
|
|
assert asset.config_json["last_operation"]["actor"] == "pytest"
|