import unittest from pydantic import ValidationError from services.message_policy import MESSAGE_CONFIRMATION from services.message_stage import ( MessageDisplaySpec, MessageStageRunner, MessageStageSpec, ) from services.test_runs.mock_tools import ( TextClientToolPort, ToolSimulationError, ToolSimulationResolver, ) from test_schemas import TestCaseDefinition def definition_with_simulation(simulation: dict) -> TestCaseDefinition: return TestCaseDefinition.model_validate( { "turns": [ { "id": "turn_1", "userInput": "查询订单", "behaviors": [ { "id": "behavior_1", "type": "tool_call", "toolId": "tool_order", "functionName": "query_order", "expectation": "called", "minCalls": 1, "maxCalls": 1, "paramAssertions": [], } ], } ], "toolSimulations": [simulation], } ) class ToolSimulationSchemaTests(unittest.TestCase): def test_called_assertion_requires_simulation(self): with self.assertRaisesRegex(ValidationError, "缺少工具模拟"): TestCaseDefinition.model_validate( { "turns": [ { "id": "turn_1", "userInput": "查询订单", "behaviors": [ { "id": "behavior_1", "type": "tool_call", "toolId": "tool_order", "functionName": "query_order", "expectation": "called", "minCalls": 1, "paramAssertions": [], } ], } ] } ) def test_sequence_turn_scope_must_reference_existing_turn(self): with self.assertRaisesRegex(ValidationError, "不存在的轮次"): definition_with_simulation( { "toolId": "tool_order", "functionName": "query_order", "mode": "sequence", "steps": [ { "id": "step_1", "scope": {"phase": "turn", "turnId": "missing"}, "outcome": "success", "body": "{}", "delayMs": 0, } ], } ) def test_legacy_equal_responses_become_fixed_simulation(self): definition = TestCaseDefinition.model_validate( { "turns": [ { "id": "turn_1", "userInput": "第一次查询", "behaviors": [ { "id": "behavior_1", "type": "tool_call", "toolId": "tool_order", "functionName": "query_order", "expectation": "called", "minCalls": 1, "paramAssertions": [], "mockResponse": { "outcome": "success", "body": '{"orderId":"A1"}', "delayMs": 0, }, } ], }, { "id": "turn_2", "userInput": "再次查询", "behaviors": [ { "id": "behavior_2", "type": "tool_call", "toolId": "tool_order", "functionName": "query_order", "expectation": "called", "minCalls": 1, "paramAssertions": [], "mockResponse": { "outcome": "success", "body": '{"orderId":"A1"}', "delayMs": 0, }, } ], }, ] } ) self.assertEqual(definition.tool_simulations[0].mode, "fixed") dumped = definition.model_dump(mode="json", by_alias=True) self.assertNotIn("mockResponse", dumped["turns"][0]["behaviors"][0]) def test_legacy_different_responses_become_scoped_sequence(self): definition = TestCaseDefinition.model_validate( { "turns": [ { "id": "turn_1", "userInput": "第一次查询", "behaviors": [ { "id": "behavior_1", "type": "tool_call", "toolId": "tool_order", "functionName": "query_order", "expectation": "called", "minCalls": 2, "paramAssertions": [], "mockResponse": { "outcome": "success", "body": '{"orderId":"A1"}', "delayMs": 0, }, } ], }, { "id": "turn_2", "userInput": "再次查询", "behaviors": [ { "id": "behavior_2", "type": "tool_call", "toolId": "tool_order", "functionName": "query_order", "expectation": "called", "minCalls": 1, "paramAssertions": [], "mockResponse": { "outcome": "error", "body": '{"message":"not found"}', "delayMs": 0, }, } ], }, ] } ) simulation = definition.tool_simulations[0] self.assertEqual(simulation.mode, "sequence") self.assertEqual(len(simulation.steps), 3) self.assertEqual(simulation.steps[0].scope.turn_id, "turn_1") self.assertEqual(simulation.steps[2].scope.turn_id, "turn_2") class ToolSimulationResolverTests(unittest.IsolatedAsyncioTestCase): async def test_client_message_stage_keeps_unmocked_failure_for_runner(self): resolver = ToolSimulationResolver([]) stage = MessageStageRunner(TextClientToolPort(resolver)) result = await stage.run( MessageStageSpec( display=MessageDisplaySpec(title="确认", message="请确认"), completion_policy=MESSAGE_CONFIRMATION, ) ) self.assertFalse(result.succeeded) self.assertIn("UNMOCKED_TOOL_CALL", result.error or "") with self.assertRaisesRegex(ToolSimulationError, "UNMOCKED_TOOL_CALL"): resolver.raise_if_failed() async def test_fixed_response_can_be_reused(self): definition = definition_with_simulation( { "toolId": "tool_order", "functionName": "query_order", "mode": "fixed", "response": { "outcome": "success", "body": '{"orderId":"A1"}', "delayMs": 0, }, } ) resolver = ToolSimulationResolver(definition.tool_simulations) resolver.enter_turn("turn_1") first = await resolver.resolve( tool_id="tool_order", function_name="query_order" ) second = await resolver.resolve( tool_id="tool_order", function_name="query_order" ) self.assertEqual(first, {"orderId": "A1", "status": "ok"}) self.assertEqual(second, first) async def test_sequence_consumes_client_choices_then_exhausts(self): definition = definition_with_simulation( { "toolId": "tool_order", "functionName": "query_order", "mode": "sequence", "steps": [ { "id": "step_1", "scope": {"phase": "turn", "turnId": "turn_1"}, "outcome": "success", "body": '{"action":"selected","value":"basic"}', "delayMs": 0, }, { "id": "step_2", "scope": {"phase": "any"}, "outcome": "success", "body": '{"action":"dismissed"}', "delayMs": 0, }, ], } ) resolver = ToolSimulationResolver(definition.tool_simulations) resolver.enter_turn("turn_1") port = TextClientToolPort(resolver) selected = await port.call("query_order", {}) dismissed = await port.call("query_order", {}) self.assertEqual(selected["data"]["value"], "basic") self.assertEqual(dismissed["data"]["action"], "dismissed") with self.assertRaisesRegex( ToolSimulationError, "TOOL_SIMULATION_EXHAUSTED" ): await port.call("query_order", {}) async def test_missing_and_scope_mismatch_are_distinct(self): definition = definition_with_simulation( { "toolId": "tool_order", "functionName": "query_order", "mode": "sequence", "steps": [ { "id": "step_1", "scope": {"phase": "turn", "turnId": "turn_1"}, "outcome": "success", "body": "{}", "delayMs": 0, } ], } ) resolver = ToolSimulationResolver(definition.tool_simulations) resolver.enter_startup() with self.assertRaisesRegex( ToolSimulationError, "TOOL_SIMULATION_SCOPE_MISMATCH" ): await resolver.resolve( tool_id="tool_order", function_name="query_order" ) with self.assertRaisesRegex( ToolSimulationError, "TOOL_SIMULATION_SCOPE_MISMATCH" ): resolver.raise_if_failed() with self.assertRaisesRegex(ToolSimulationError, "UNMOCKED_TOOL_CALL"): await resolver.resolve(tool_id="missing", function_name="missing") if __name__ == "__main__": unittest.main()