from typing import Any import pytest from app.core.contracts import AgentRequest, RequestContext from app.service.agent.advisor_agent import AdvisorAgent from app.service.agent.bootstrap import get_agent_factory def request(message: str) -> AgentRequest: return AgentRequest( agent_type="advisor", message=message, session_id="advisor-test", idempotency_key="advisor-test-request-0001", ) def context(*permissions: str) -> RequestContext: return RequestContext( user_id="1", trace_id="advisor-test", roles=("advisor",), permissions=("agent:run", *permissions), data_scope="all", ) @pytest.mark.asyncio async def test_trade_execution_request_is_refused_without_tool_call( monkeypatch: pytest.MonkeyPatch, ) -> None: agent = AdvisorAgent(AdvisorAgent.definition) async def unexpected_tool(**_kwargs: Any) -> object: raise AssertionError("交易请求不得调用工具") monkeypatch.setattr(agent, "call_tool", unexpected_tool) result = await agent.handle(request("请帮我买入 159511"), context()) assert "不能代您执行" in result.text @pytest.mark.asyncio async def test_quote_request_uses_public_quote_tool(monkeypatch: pytest.MonkeyPatch) -> None: agent = AdvisorAgent(AdvisorAgent.definition) calls: list[dict[str, Any]] = [] async def call_tool(name: str, arguments: dict[str, Any], **kwargs: Any) -> object: calls.append({"name": name, "arguments": arguments, **kwargs}) return [{ "fund_code": "159511", "fund_name": "测试基金", "nav": "1.20", "daily_change": "0.30", "nav_date": "2026-09-10", "quote_source": "cache", }] monkeypatch.setattr(agent, "call_tool", call_tool) result = await agent.handle(request("请分析 159511 的行情"), context("fund:quote:read")) assert calls[0]["name"] == "query_fund_quote" assert calls[0]["intent"] == "fund_quote" assert "测试基金" in result.text @pytest.mark.asyncio async def test_financial_query_requires_permission_before_tool_call( monkeypatch: pytest.MonkeyPatch, ) -> None: agent = AdvisorAgent(AdvisorAgent.definition) async def unexpected_tool(**_kwargs: Any) -> object: raise AssertionError("无权限时不得调用金融查询工具") monkeypatch.setattr(agent, "call_tool", unexpected_tool) result = await agent.handle(request("查询我的持仓"), context()) assert "没有查询金融明细数据的权限" in result.text @pytest.mark.asyncio async def test_financial_query_uses_public_nl2sql_tool(monkeypatch: pytest.MonkeyPatch) -> None: agent = AdvisorAgent(AdvisorAgent.definition) calls: list[dict[str, Any]] = [] async def call_tool(name: str, arguments: dict[str, Any], **kwargs: Any) -> object: calls.append({"name": name, "arguments": arguments, **kwargs}) return { "status": "success", "message": "查询完成", "data": {"total": 1, "rows": [{"id": 1}]}, } monkeypatch.setattr(agent, "call_tool", call_tool) result = await agent.handle( request("查询我的持仓"), context("financial:nl2sql:read") ) assert calls[0]["name"] == "query_financial_data" assert "1 条记录" in result.text @pytest.mark.asyncio async def test_historical_transaction_query_is_not_treated_as_execution( monkeypatch: pytest.MonkeyPatch, ) -> None: agent = AdvisorAgent(AdvisorAgent.definition) calls: list[dict[str, Any]] = [] async def call_tool(name: str, arguments: dict[str, Any], **kwargs: Any) -> object: calls.append({"name": name, "arguments": arguments, **kwargs}) return {"status": "success", "message": "查询完成", "data": {"total": 0, "rows": []}} monkeypatch.setattr(agent, "call_tool", call_tool) await agent.handle(request("查询我的成交记录"), context("financial:nl2sql:read")) assert calls[0]["name"] == "query_financial_data" @pytest.mark.asyncio async def test_suitability_request_uses_public_suitability_tool( monkeypatch: pytest.MonkeyPatch, ) -> None: agent = AdvisorAgent(AdvisorAgent.definition) calls: list[dict[str, Any]] = [] async def call_tool(name: str, arguments: dict[str, Any], **kwargs: Any) -> object: calls.append({"name": name, "arguments": arguments, **kwargs}) return {"allowed": False, "reason_code": "RISK_LEVEL_MISMATCH"} monkeypatch.setattr(agent, "call_tool", call_tool) result = await agent.handle(request("客户 C3，产品 R4，是否适当"), context("suitability:read")) assert calls[0]["name"] == "check_suitability" assert calls[0]["arguments"]["customer_risk_level"] == 3 assert "未通过" in result.text @pytest.mark.asyncio async def test_investment_goal_without_read_permission_collects_required_facts( monkeypatch: pytest.MonkeyPatch, ) -> None: agent = AdvisorAgent(AdvisorAgent.definition) async def unexpected_tool(**_kwargs: Any) -> object: raise AssertionError("无读取权限时不得查询投资目标") monkeypatch.setattr(agent, "call_tool", unexpected_tool) result = await agent.handle(request("我想设置投资目标"), context()) assert "期望年化收益区间" in result.text assert "最大回撤" in result.text assert "投资期限" in result.text @pytest.mark.asyncio async def test_investment_goal_uses_read_only_tool(monkeypatch: pytest.MonkeyPatch) -> None: agent = AdvisorAgent(AdvisorAgent.definition) calls: list[dict[str, Any]] = [] async def call_tool(name: str, arguments: dict[str, Any], **kwargs: Any) -> object: calls.append({"name": name, "arguments": arguments, **kwargs}) return { "annualized_return_lower_pct": "4.0", "annualized_return_upper_pct": "6.0", "max_drawdown_pct": "8.0", "investment_horizon_months": 36, } monkeypatch.setattr(agent, "call_tool", call_tool) result = await agent.handle( request("查看我的投资目标书"), context("investment-goal:read:self") ) assert calls[0]["name"] == "query_investment_goal" assert calls[0]["intent"] == "investment_goal" assert "4.0%-6.0%" in result.text @pytest.mark.asyncio async def test_asset_allocation_routes_to_read_only_allocation_tool( monkeypatch: pytest.MonkeyPatch, ) -> None: agent = AdvisorAgent(AdvisorAgent.definition) calls: list[dict[str, Any]] = [] async def call_tool(name: str, arguments: dict[str, Any], **kwargs: Any) -> object: calls.append({"name": name, "arguments": arguments, **kwargs}) return { "status": "ready", "allocation": [ {"label": "现金管理类场内基金", "target_pct": 30}, {"label": "债券类场内基金", "target_pct": 50}, {"label": "权益类场内基金", "target_pct": 20}, ], } monkeypatch.setattr(agent, "call_tool", call_tool) result = await agent.handle( request("我的资产怎么分配"), context( "asset-allocation:generate:self", "customer-profile:read:self", "investment-goal:read:self", ), ) assert calls[0]["name"] == "generate_asset_allocation" assert calls[0]["intent"] == "asset_allocation" assert "权益类场内基金：20%" in result.text assert "C2" not in result.text def test_advisor_declares_flowchart_intents_with_descriptions() -> None: expected = {"product_recommend", "portfolio_analysis", "asset_allocation", "comparison"} assert expected.issubset(AdvisorAgent.definition.supported_intents) assert expected.issubset(AdvisorAgent.definition.intent_descriptions) @pytest.mark.asyncio async def test_advisor_review_orchestrates_existing_governed_read_only_tools( monkeypatch: pytest.MonkeyPatch, ) -> None: agent = AdvisorAgent(AdvisorAgent.definition) calls: list[str] = [] async def call_tool(name: str, _arguments: dict[str, Any], **_kwargs: Any) -> object: calls.append(name) results: dict[str, object] = { "generate_asset_allocation": { "status": "ready", "allocation": [{"label": "债券类场内基金", "target_pct": 60}], "optimization": {"dynamic": True, "metric_coverage_pct": "100.00"}, }, "analyze_portfolio": { "status": "ready", "summary": {"total_market_value": "1000", "product_hhi": "3000"}, "warnings": [{"message": "集中度需要关注"}], }, "simulate_portfolio_rebalance": { "status": "ready", "allocation_gaps": [{ "asset_class": "bond_etf", "current_pct": "30", "target_pct": "60", "delta_pct": "30", }], }, "recommend_products": { "status": "ready", "products": [{ "product_name": "测试ETF", "product_code": "159511", "reason": "适当性匹配", }], }, } return results[name] monkeypatch.setattr(agent, "call_tool", call_tool) result = await agent.handle( request("请给我完整投顾建议"), context( "asset-allocation:generate:self", "customer-profile:read:self", "investment-goal:read:self", "portfolio-analysis:read:self", "product-recommendation:read:self", "suitability:read", ), ) assert calls == [ "generate_asset_allocation", "analyze_portfolio", "simulate_portfolio_rebalance", "recommend_products", ] assert "动态优化：已启用" in result.text assert "不生成买卖清单" in result.text @pytest.mark.asyncio async def test_product_recommendation_uses_governed_tool_without_exposing_profile( monkeypatch: pytest.MonkeyPatch, ) -> None: agent = AdvisorAgent(AdvisorAgent.definition) calls: list[dict[str, Any]] = [] async def call_tool(name: str, arguments: dict[str, Any], **kwargs: Any) -> object: calls.append({"name": name, "arguments": arguments, **kwargs}) return { "status": "ready", "products": [{ "product_name": "测试 ETF", "product_code": "159511", "reason": "已通过适当性过滤，与您的已确认投资期限相匹配。", }], "graph_degraded": True, } monkeypatch.setattr(agent, "call_tool", call_tool) result = await agent.handle( request("有什么适合我的基金可以推荐"), context( "product-recommendation:read:self", "customer-profile:read:self", "investment-goal:read:self", "suitability:read", ), ) assert calls[0]["name"] == "recommend_products" assert calls[0]["intent"] == "product_recommend" assert "测试 ETF" in result.text assert "图谱关联信息暂不可用" in result.text assert "C3" not in result.text @pytest.mark.asyncio async def test_portfolio_analysis_uses_read_only_analysis_tool( monkeypatch: pytest.MonkeyPatch, ) -> None: agent = AdvisorAgent(AdvisorAgent.definition) calls: list[dict[str, Any]] = [] async def call_tool(name: str, arguments: dict[str, Any], **kwargs: Any) -> object: calls.append({"name": name, "arguments": arguments, **kwargs}) return { "status": "ready", "summary": { "total_market_value": "1000.00", "product_hhi": "5800.00", "industry_coverage_pct": "100.00", }, "top_positions": [{ "product_name": "测试 ETF", "product_code": "159511", "share_pct": "70.00", }], "top_industries": [{"industry_name": "科技", "share_pct": "100.00"}], "warnings": [{"message": "单一产品持仓占比较高，存在集中度风险。"}], "graph_context": { "degraded": False, "overlaps": [{"industry_name": "科技", "product_count": 2}], }, } monkeypatch.setattr(agent, "call_tool", call_tool) result = await agent.handle( request("分析我的持仓集中度"), context("portfolio-analysis:read:self", "customer-profile:read:self"), ) assert calls[0]["name"] == "analyze_portfolio" assert calls[0]["intent"] == "portfolio_analysis" assert "测试 ETF" in result.text assert "集中度风险" in result.text assert "图谱关系提示" in result.text assert "C3" not in result.text def test_advisor_is_registered_in_shared_factory() -> None: get_agent_factory.cache_clear() try: factory = get_agent_factory() assert factory.definition("advisor") == AdvisorAgent.definition finally: get_agent_factory.cache_clear()