# Copyright 2026 Emcie Co Ltd. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import asyncio from lagom import Container from fastapi import status import httpx from parlant.core.services.tools.plugins import tool from parlant.core.tools import ToolResult, ToolContext from parlant.core.services.tools.service_registry import ServiceRegistry from tests.test_utilities import run_service_server AMOUNT_OF_TIME_TO_WAIT_FOR_EVALUATION_TO_START_RUNNING = 2 async def test_that_an_evaluation_can_be_created_and_fetched_with_completed_status( async_client: httpx.AsyncClient, ) -> None: response = await async_client.post( "/evaluations", json={ "payloads": [ { "kind": "guideline", "guideline": { "content": { "condition": "the customer greets you", "action": "greet them back with 'Hello'", }, "tool_ids": [ {"service_name": "google_calendar", "tool_name": "get_events"} ], "operation": "add", "action_proposition": True, "properties_proposition": True, }, } ], }, ) assert response.status_code == status.HTTP_201_CREATED evaluation_id = response.raise_for_status().json()["id"] content = (await async_client.get(f"/evaluations/{evaluation_id}")).raise_for_status().json() assert content["status"] == "completed" assert len(content["invoices"]) == 1 invoice = content["invoices"][0] assert invoice["approved"] assert invoice["data"] assert invoice["data"]["guideline"]["properties_proposition"]["internal_action"] is None async def test_that_an_evaluation_can_be_fetched_with_running_status( async_client: httpx.AsyncClient, ) -> None: response = await async_client.post( "/evaluations", json={ "payloads": [ { "kind": "guideline", "guideline": { "content": { "condition": "the customer greets you", "action": "greet them back with 'Hello'", }, "operation": "add", "action_proposition": True, "properties_proposition": True, "tool_ids": [ {"service_name": "google_calendar", "tool_name": "get_events"} ], }, } ], }, ) evaluation_id = response.raise_for_status().json()["id"] await asyncio.sleep(AMOUNT_OF_TIME_TO_WAIT_FOR_EVALUATION_TO_START_RUNNING) content = ( (await async_client.get(f"/evaluations/{evaluation_id}", params={"wait_for_completion": 0})) .raise_for_status() .json() ) assert content["status"] in {"running", "completed"} async def test_that_an_error_is_returned_when_no_payloads_are_provided( async_client: httpx.AsyncClient, ) -> None: response = await async_client.post("/evaluations", json={"payloads": []}) assert response.status_code == status.HTTP_422_UNPROCESSABLE_CONTENT data = response.json() assert "detail" in data assert data["detail"] == "No payloads provided for the evaluation task." async def test_that_properties_proposition_is_evaluated( async_client: httpx.AsyncClient, ) -> None: response = await async_client.post( "/evaluations", json={ "payloads": [ { "kind": "guideline", "guideline": { "content": { "condition": "the customer asks for a discount", "action": "maintain a helpful tone and ask the customer what discount they would like", }, "operation": "add", "action_proposition": True, "properties_proposition": True, "tool_ids": [ {"service_name": "google_calendar", "tool_name": "get_events"} ], }, } ], }, ) assert response.status_code == status.HTTP_201_CREATED evaluation_id = response.raise_for_status().json()["id"] content = (await async_client.get(f"/evaluations/{evaluation_id}")).raise_for_status().json() assert content["status"] == "completed" assert len(content["invoices"]) == 1 invoice = content["invoices"][0] assert invoice["approved"] assert invoice["data"] assert invoice["data"]["guideline"]["properties_proposition"]["continuous"] assert invoice["data"]["guideline"]["properties_proposition"]["customer_dependent_action_data"][ "is_customer_dependent" ] assert invoice["data"]["guideline"]["properties_proposition"]["customer_dependent_action_data"][ "customer_action" ] assert invoice["data"]["guideline"]["properties_proposition"]["customer_dependent_action_data"][ "agent_action" ] async def test_that_action_proposition_is_evaluated( async_client: httpx.AsyncClient, container: Container, ) -> None: @tool def my_tool(context: ToolContext, arg_1: int, arg_2: int) -> ToolResult: return ToolResult(arg_1 + arg_2) service_registry = container[ServiceRegistry] async with run_service_server([my_tool]) as server: await service_registry.update_tool_service( name="my_service", kind="sdk", url=server.url, ) response = await async_client.post( "/evaluations", json={ "payloads": [ { "kind": "guideline", "guideline": { "content": { "condition": "the customer asks for a discount", }, "tool_ids": [{"service_name": "my_service", "tool_name": "my_tool"}], "operation": "add", "action_proposition": True, "properties_proposition": False, }, } ], }, ) assert response.status_code == status.HTTP_201_CREATED evaluation_id = response.raise_for_status().json()["id"] content = ( (await async_client.get(f"/evaluations/{evaluation_id}")).raise_for_status().json() ) assert content["status"] == "completed" assert len(content["invoices"]) == 1 invoice = content["invoices"][0] assert invoice["approved"] assert invoice["data"] guideline_data = invoice["data"]["guideline"] assert isinstance(guideline_data["properties_proposition"], dict) assert guideline_data["properties_proposition"].get("internal_action") is not None assert guideline_data["action_proposition"] is not None async def test_that_agent_intention_condition_is_included_in_properties_proposition( async_client: httpx.AsyncClient, ) -> None: response = await async_client.post( "/evaluations", json={ "payloads": [ { "kind": "guideline", "guideline": { "content": { "condition": "you want to upsell the customer", "action": "recommend the premium plan", }, "operation": "add", "action_proposition": False, "properties_proposition": True, "tool_ids": [], }, } ], }, ) assert response.status_code == status.HTTP_201_CREATED evaluation_id = response.raise_for_status().json()["id"] content = (await async_client.get(f"/evaluations/{evaluation_id}")).raise_for_status().json() assert content["status"] == "completed" assert len(content["invoices"]) == 1 invoice = content["invoices"][0] assert invoice["approved"] assert invoice["data"] properties = invoice["data"]["guideline"]["properties_proposition"] assert "agent_intention_condition" in properties async def test_that_payload_proposition_flags_are_correctly_returned_in_invoice( async_client: httpx.AsyncClient, ) -> None: response = await async_client.post( "/evaluations", json={ "payloads": [ { "kind": "guideline", "guideline": { "content": { "condition": "the customer greets you", "action": "greet them back with 'Hello'", }, "tool_ids": [], "operation": "add", "action_proposition": True, "properties_proposition": False, }, } ], }, ) assert response.status_code == status.HTTP_201_CREATED evaluation_id = response.raise_for_status().json()["id"] content = (await async_client.get(f"/evaluations/{evaluation_id}")).raise_for_status().json() assert content["status"] == "completed" assert len(content["invoices"]) == 1 invoice = content["invoices"][0] payload_guideline = invoice["payload"]["guideline"] assert payload_guideline["action_proposition"] is True assert payload_guideline["properties_proposition"] is False async def test_that_error_is_returned_when_no_propositions_are_provided_in_a_payload( async_client: httpx.AsyncClient, ) -> None: response = await async_client.post( "/evaluations", json={ "payloads": [ { "kind": "guideline", "guideline": { "content": { "condition": "the customer greets you", "action": "greet them back with 'Hello'", }, "tool_ids": [ {"service_name": "google_calendar", "tool_name": "get_events"} ], "operation": "add", }, } ], }, ) assert response.status_code == status.HTTP_422_UNPROCESSABLE_CONTENT data = response.json() assert "detail" in data assert ( data["detail"] == "At least one of action_proposition, properties_proposition or journey_node_proposition must be enabled" ) async def test_that_error_is_returned_when_all_propositions_are_disabled_in_a_payload( async_client: httpx.AsyncClient, ) -> None: response = await async_client.post( "/evaluations", json={ "payloads": [ { "kind": "guideline", "guideline": { "content": { "condition": "the customer greets you", "action": "greet them back with 'Hello'", }, "tool_ids": [ {"service_name": "google_calendar", "tool_name": "get_events"} ], "operation": "add", "action_proposition": False, "properties_proposition": False, }, } ], }, ) assert response.status_code == status.HTTP_422_UNPROCESSABLE_CONTENT data = response.json() assert "detail" in data assert ( data["detail"] == "At least one of action_proposition, properties_proposition or journey_node_proposition must be enabled" )