From 94c55e2d173ad6640640cb43d8009c27049ebfe9 Mon Sep 17 00:00:00 2001 From: Anik Bhattacharjee Date: Wed, 19 Aug 2026 09:51:33 -0400 Subject: [PATCH] LCORE-1794: Add OpenTelemetry spans to conversation v1 and v2 endpoints MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add coarse-grained OpenTelemetry spans to all 8 conversation endpoint handlers (4 in v1, 4 in v2) - Each handler emits a single span with operation-level metadata: conversation count (list), turn count and found flag (get), deleted flag (delete), updated flag (update) - All attributes are low-cardinality — no raw conversation IDs or user-derived free text in span attributes - Errors are auto-captured via start_as_current_span context manager setting StatusCode.ERROR Signed-off-by: Anik Bhattacharjee --- src/app/endpoints/conversations_v1.py | 583 +++++++++--------- src/app/endpoints/conversations_v2.py | 165 ++--- .../unit/app/endpoints/test_conversations.py | 366 +++++++++++ .../app/endpoints/test_conversations_v2.py | 259 +++++++- 4 files changed, 1013 insertions(+), 360 deletions(-) diff --git a/src/app/endpoints/conversations_v1.py b/src/app/endpoints/conversations_v1.py index 6ab693658..67a3a33d6 100644 --- a/src/app/endpoints/conversations_v1.py +++ b/src/app/endpoints/conversations_v1.py @@ -8,6 +8,7 @@ APIConnectionError, APIStatusError, ) +from opentelemetry import trace from sqlalchemy.exc import SQLAlchemyError from app.database import get_session @@ -56,6 +57,7 @@ ) logger = get_logger(__name__) +tracer = trace.get_tracer(__name__) router = APIRouter(tags=["conversations_v1"]) conversation_get_responses: dict[int | str, dict[str, Any]] = { @@ -125,54 +127,59 @@ async def get_conversations_list_endpoint_handler( auth: Any = Depends(get_auth_dependency()), ) -> ConversationsListResponse: """Handle request to retrieve all conversations for the authenticated user.""" - check_configuration_loaded(configuration) + with tracer.start_as_current_span("conversations_v1.list") as span: + check_configuration_loaded(configuration) - user_id = auth[0] + user_id = auth[0] - logger.info("Retrieving conversations for user %s", user_id) + logger.info("Retrieving conversations for user %s", user_id) - with get_session() as session: - try: - query = session.query(UserConversation) + with get_session() as session: + try: + query = session.query(UserConversation) + + filtered_query = ( + query + if Action.LIST_OTHERS_CONVERSATIONS + in request.state.authorized_actions + else query.filter_by(user_id=user_id) + ) - filtered_query = ( - query - if Action.LIST_OTHERS_CONVERSATIONS in request.state.authorized_actions - else query.filter_by(user_id=user_id) - ) + user_conversations = filtered_query.all() + + # Return conversation summaries with metadata + conversations = [ + ConversationDetails( + conversation_id=conv.id, + created_at=( + conv.created_at.isoformat() if conv.created_at else None + ), + last_message_at=( + conv.last_message_at.isoformat() + if conv.last_message_at + else None + ), + message_count=conv.message_count, + last_used_model=conv.last_used_model, + last_used_provider=conv.last_used_provider, + topic_summary=conv.topic_summary, + ) + for conv in user_conversations + ] - user_conversations = filtered_query.all() - - # Return conversation summaries with metadata - conversations = [ - ConversationDetails( - conversation_id=conv.id, - created_at=conv.created_at.isoformat() if conv.created_at else None, - last_message_at=( - conv.last_message_at.isoformat() - if conv.last_message_at - else None - ), - message_count=conv.message_count, - last_used_model=conv.last_used_model, - last_used_provider=conv.last_used_provider, - topic_summary=conv.topic_summary, + logger.info( + "Found %d conversations for user %s", len(conversations), user_id ) - for conv in user_conversations - ] - logger.info( - "Found %d conversations for user %s", len(conversations), user_id - ) + span.set_attribute("conversations.count", len(conversations)) + return ConversationsListResponse(conversations=conversations) - return ConversationsListResponse(conversations=conversations) - - except SQLAlchemyError as e: - logger.exception( - "Error retrieving conversations for user %s: %s", user_id, e - ) - response = InternalServerErrorResponse.database_error() - raise HTTPException(**response.model_dump()) from e + except SQLAlchemyError as e: + logger.exception( + "Error retrieving conversations for user %s: %s", user_id, e + ) + response = InternalServerErrorResponse.database_error() + raise HTTPException(**response.model_dump()) from e @router.get( @@ -204,89 +211,90 @@ async def get_conversation_endpoint_handler( # pylint: disable=too-many-locals, ConversationResponse: Structured response containing the conversation ID and simplified chat history """ - check_configuration_loaded(configuration) - - # Validate conversation ID format - if not check_suid(conversation_id): - logger.error("Invalid conversation ID format: %s", conversation_id) - response = BadRequestResponse( - resource="conversation", resource_id=conversation_id - ).model_dump() - raise HTTPException(**response) - - # Normalize the conversation ID for database operations (strip conv_ prefix if present) - normalized_conv_id = normalize_conversation_id(conversation_id) - logger.debug( - "GET conversation - original ID: %s, normalized ID: %s", - conversation_id, - normalized_conv_id, - ) - - user_id = auth[0] - conversation = validate_and_retrieve_conversation( - normalized_conv_id=normalized_conv_id, - user_id=user_id, - others_allowed=( - Action.READ_OTHERS_CONVERSATIONS in request.state.authorized_actions - ), - ) - logger.info( - "Retrieving conversation %s using Conversations API", normalized_conv_id - ) - - try: - client = AsyncOgxClientHolder().get_client() - - # Convert to llama-stack format (add 'conv_' prefix if needed) - llama_stack_conv_id = to_llama_stack_conversation_id(normalized_conv_id) - logger.debug( - "Calling llama-stack list_items with conversation_id: %s", - llama_stack_conv_id, - ) - - # Retrieve turns metadata from database (can be empty for legacy conversations) - db_turns = retrieve_conversation_turns(normalized_conv_id) - - # Use Conversations API to retrieve conversation items - items = await get_all_conversation_items(client, llama_stack_conv_id) - if not items: - logger.error("No items found for conversation %s", conversation_id) - response = NotFoundResponse( - resource="conversation", resource_id=normalized_conv_id + with tracer.start_as_current_span("conversations_v1.get") as span: + check_configuration_loaded(configuration) + + # Validate conversation ID format + if not check_suid(conversation_id): + logger.error("Invalid conversation ID format: %s", conversation_id) + response = BadRequestResponse( + resource="conversation", resource_id=conversation_id ).model_dump() raise HTTPException(**response) - logger.info( - "Successfully retrieved %d items for conversation %s", - len(items), + # Normalize the conversation ID for database operations + normalized_conv_id = normalize_conversation_id(conversation_id) + logger.debug( + "GET conversation - original ID: %s, normalized ID: %s", conversation_id, + normalized_conv_id, ) - # Build conversation turns from items and populate turns metadata - # Use conversation.created_at for legacy conversations without turn metadata - chat_history = build_conversation_turns_from_items( - items, db_turns, conversation.created_at + user_id = auth[0] + conversation = validate_and_retrieve_conversation( + normalized_conv_id=normalized_conv_id, + user_id=user_id, + others_allowed=( + Action.READ_OTHERS_CONVERSATIONS in request.state.authorized_actions + ), ) - - return ConversationResponse( - conversation_id=normalized_conv_id, - chat_history=chat_history, + logger.info( + "Retrieving conversation %s using Conversations API", normalized_conv_id ) - except APIConnectionError as e: - logger.error("Unable to connect to Llama Stack: %s", e) - response = ServiceUnavailableResponse( - backend_name="OGX", cause=str(e) - ).model_dump() - raise HTTPException(**response) from e + try: + client = AsyncOgxClientHolder().get_client() + + # Convert to llama-stack format (add 'conv_' prefix if needed) + llama_stack_conv_id = to_llama_stack_conversation_id(normalized_conv_id) + logger.debug( + "Calling llama-stack list_items with conversation_id: %s", + llama_stack_conv_id, + ) + + # Retrieve turns metadata from database + db_turns = retrieve_conversation_turns(normalized_conv_id) - except (APIStatusError, ConversationNotFoundError) as e: - # In library mode, ConversationNotFoundError is raised instead of APIStatusError - logger.error("Conversation not found: %s", e) - response = NotFoundResponse( - resource="conversation", resource_id=normalized_conv_id - ).model_dump() - raise HTTPException(**response) from e + # Use Conversations API to retrieve conversation items + items = await get_all_conversation_items(client, llama_stack_conv_id) + if not items: + logger.error("No items found for conversation %s", conversation_id) + response = NotFoundResponse( + resource="conversation", resource_id=normalized_conv_id + ).model_dump() + raise HTTPException(**response) + + logger.info( + "Successfully retrieved %d items for conversation %s", + len(items), + conversation_id, + ) + + # Build conversation turns from items and populate turns metadata + chat_history = build_conversation_turns_from_items( + items, db_turns, conversation.created_at + ) + + span.set_attribute("conversations.found", True) + span.set_attribute("conversations.turns.count", len(chat_history)) + return ConversationResponse( + conversation_id=normalized_conv_id, + chat_history=chat_history, + ) + + except APIConnectionError as e: + logger.error("Unable to connect to Llama Stack: %s", e) + response = ServiceUnavailableResponse( + backend_name="OGX", cause=str(e) + ).model_dump() + raise HTTPException(**response) from e + + except (APIStatusError, ConversationNotFoundError) as e: + logger.error("Conversation not found: %s", e) + response = NotFoundResponse( + resource="conversation", resource_id=normalized_conv_id + ).model_dump() + raise HTTPException(**response) from e @router.delete( @@ -315,91 +323,94 @@ async def delete_conversation_endpoint_handler( Returns: ConversationDeleteResponse: Response indicating the result of the deletion operation """ - check_configuration_loaded(configuration) - - # Validate conversation ID format - if not check_suid(conversation_id): - logger.error("Invalid conversation ID format: %s", conversation_id) - response = BadRequestResponse( - resource="conversation", resource_id=conversation_id - ).model_dump() - raise HTTPException(**response) - - # Normalize the conversation ID for database operations (strip conv_ prefix if present) - normalized_conv_id = normalize_conversation_id(conversation_id) - - # Check if user has access to delete this conversation - user_id = auth[0] - if not can_access_conversation( - normalized_conv_id, - user_id, - others_allowed=( - Action.DELETE_OTHERS_CONVERSATIONS in request.state.authorized_actions - ), - ): - logger.warning( - "User %s attempted to delete conversation %s they don't have access to", - user_id, + with tracer.start_as_current_span("conversations_v1.delete") as span: + check_configuration_loaded(configuration) + + # Validate conversation ID format + if not check_suid(conversation_id): + logger.error("Invalid conversation ID format: %s", conversation_id) + response = BadRequestResponse( + resource="conversation", resource_id=conversation_id + ).model_dump() + raise HTTPException(**response) + + # Normalize the conversation ID for database operations + normalized_conv_id = normalize_conversation_id(conversation_id) + + # Check if user has access to delete this conversation + user_id = auth[0] + if not can_access_conversation( normalized_conv_id, - ) - response = ForbiddenResponse.conversation( - action="delete", - resource_id=normalized_conv_id, - user_id=user_id, - ).model_dump() - raise HTTPException(**response) + user_id, + others_allowed=( + Action.DELETE_OTHERS_CONVERSATIONS in request.state.authorized_actions + ), + ): + logger.warning( + "User %s attempted to delete conversation %s they don't have access to", + user_id, + normalized_conv_id, + ) + response = ForbiddenResponse.conversation( + action="delete", + resource_id=normalized_conv_id, + user_id=user_id, + ).model_dump() + raise HTTPException(**response) - # If reached this, user is authorized to delete this conversation - try: - local_deleted = delete_conversation(normalized_conv_id) - if not local_deleted: - logger.info( - "Conversation %s not found locally when deleting.", + # If reached this, user is authorized to delete this conversation + try: + local_deleted = delete_conversation(normalized_conv_id) + if not local_deleted: + logger.info( + "Conversation %s not found locally when deleting.", + normalized_conv_id, + ) + except SQLAlchemyError as e: + logger.error( + "Database error while deleting conversation %s", normalized_conv_id, ) - except SQLAlchemyError as e: - logger.error( - "Database error while deleting conversation %s", - normalized_conv_id, + response = InternalServerErrorResponse.database_error() + raise HTTPException(**response.model_dump()) from e + + logger.info( + "Deleting conversation %s using Conversations API", normalized_conv_id ) - response = InternalServerErrorResponse.database_error() - raise HTTPException(**response.model_dump()) from e - logger.info("Deleting conversation %s using Conversations API", normalized_conv_id) + try: + # Get Llama Stack client + client = AsyncOgxClientHolder().get_client() - try: - # Get Llama Stack client - client = AsyncOgxClientHolder().get_client() + # Convert to llama-stack format (add 'conv_' prefix if needed) + llama_stack_conv_id = to_llama_stack_conversation_id(normalized_conv_id) - # Convert to llama-stack format (add 'conv_' prefix if needed) - llama_stack_conv_id = to_llama_stack_conversation_id(normalized_conv_id) + # Use Conversations API to delete the conversation + delete_response = await client.conversations.delete( + conversation_id=llama_stack_conv_id + ) + logger.info( + "Remote deletion of %s: success=%s", + normalized_conv_id, + delete_response.deleted, + ) - # Use Conversations API to delete the conversation - delete_response = await client.conversations.delete( - conversation_id=llama_stack_conv_id - ) - logger.info( - "Remote deletion of %s: success=%s", - normalized_conv_id, - delete_response.deleted, - ) + except APIConnectionError as e: + response = ServiceUnavailableResponse(backend_name="OGX", cause=str(e)) + raise HTTPException(**response.model_dump()) from e - except APIConnectionError as e: - response = ServiceUnavailableResponse(backend_name="OGX", cause=str(e)) - raise HTTPException(**response.model_dump()) from e + except (APIStatusError, ConversationNotFoundError, InvalidParameterError): + logger.warning( + "Conversation %s in LlamaStack not found. Treating as already deleted.", + normalized_conv_id, + ) - except (APIStatusError, ConversationNotFoundError, InvalidParameterError): - # In library mode, ConversationNotFoundError is raised instead of APIStatusError - logger.warning( - "Conversation %s in LlamaStack not found. Treating as already deleted.", - normalized_conv_id, + span.set_attribute("conversations.deleted", local_deleted) + return ConversationDeleteResponse( + conversation_id=normalized_conv_id, + deleted=local_deleted, ) - return ConversationDeleteResponse( - conversation_id=normalized_conv_id, - deleted=local_deleted, - ) - @router.put( "/conversations/{conversation_id}", @@ -427,117 +438,121 @@ async def update_conversation_endpoint_handler( Returns: ConversationUpdateResponse: Response indicating the result of the update operation """ - check_configuration_loaded(configuration) - - # Validate conversation ID format - if not check_suid(conversation_id): - logger.error("Invalid conversation ID format: %s", conversation_id) - response = BadRequestResponse( - resource="conversation", resource_id=conversation_id - ).model_dump() - raise HTTPException(**response) - - # Normalize the conversation ID for database operations (strip conv_ prefix if present) - normalized_conv_id = normalize_conversation_id(conversation_id) - - user_id = auth[0] - if not can_access_conversation( - normalized_conv_id, - user_id, - others_allowed=( - Action.QUERY_OTHERS_CONVERSATIONS in request.state.authorized_actions - ), - ): - logger.warning( - "User %s attempted to update conversation %s they don't have access to", - user_id, - normalized_conv_id, - ) - response = ForbiddenResponse.conversation( - action="update", resource_id=normalized_conv_id, user_id=user_id - ).model_dump() - raise HTTPException(**response) - - # If reached this, user is authorized to update this conversation - try: - conversation = retrieve_conversation(normalized_conv_id) - if conversation is None: - response = NotFoundResponse( - resource="conversation", resource_id=normalized_conv_id + with tracer.start_as_current_span("conversations_v1.update") as span: + check_configuration_loaded(configuration) + + # Validate conversation ID format + if not check_suid(conversation_id): + logger.error("Invalid conversation ID format: %s", conversation_id) + response = BadRequestResponse( + resource="conversation", resource_id=conversation_id ).model_dump() raise HTTPException(**response) - except SQLAlchemyError as e: - logger.error( - "Database error occurred while retrieving conversation %s.", - normalized_conv_id, - ) - response = InternalServerErrorResponse.database_error() - raise HTTPException(**response.model_dump()) from e - - logger.info( - "Updating metadata for conversation %s using Conversations API", - normalized_conv_id, - ) - - try: - # Get Llama Stack client - client = AsyncOgxClientHolder().get_client() + # Normalize the conversation ID for database operations + normalized_conv_id = normalize_conversation_id(conversation_id) - # Convert to llama-stack format (add 'conv_' prefix if needed) - llama_stack_conv_id = to_llama_stack_conversation_id(normalized_conv_id) + user_id = auth[0] + if not can_access_conversation( + normalized_conv_id, + user_id, + others_allowed=( + Action.QUERY_OTHERS_CONVERSATIONS in request.state.authorized_actions + ), + ): + logger.warning( + "User %s attempted to update conversation %s they don't have access to", + user_id, + normalized_conv_id, + ) + response = ForbiddenResponse.conversation( + action="update", resource_id=normalized_conv_id, user_id=user_id + ).model_dump() + raise HTTPException(**response) - # Prepare metadata with topic summary - metadata = {"topic_summary": update_request.topic_summary} + # If reached this, user is authorized to update this conversation + try: + conversation = retrieve_conversation(normalized_conv_id) + if conversation is None: + response = NotFoundResponse( + resource="conversation", resource_id=normalized_conv_id + ).model_dump() + raise HTTPException(**response) - # Use Conversations API to update the conversation metadata - await client.conversations.update( - conversation_id=llama_stack_conv_id, - metadata=metadata, - ) + except SQLAlchemyError as e: + logger.error( + "Database error occurred while retrieving conversation %s.", + normalized_conv_id, + ) + response = InternalServerErrorResponse.database_error() + raise HTTPException(**response.model_dump()) from e logger.info( - "Successfully updated metadata for conversation %s in LlamaStack", + "Updating metadata for conversation %s using Conversations API", normalized_conv_id, ) - # Also update in local database - with get_session() as session: - db_conversation = ( - session.query(UserConversation).filter_by(id=normalized_conv_id).first() + try: + # Get Llama Stack client + client = AsyncOgxClientHolder().get_client() + + # Convert to llama-stack format (add 'conv_' prefix if needed) + llama_stack_conv_id = to_llama_stack_conversation_id(normalized_conv_id) + + # Prepare metadata with topic summary + metadata = {"topic_summary": update_request.topic_summary} + + # Use Conversations API to update the conversation metadata + await client.conversations.update( + conversation_id=llama_stack_conv_id, + metadata=metadata, ) - if db_conversation: - db_conversation.topic_summary = update_request.topic_summary - session.commit() - logger.info( - "Successfully updated topic summary in local database for conversation %s", - normalized_conv_id, + + logger.info( + "Successfully updated metadata for conversation %s in LlamaStack", + normalized_conv_id, + ) + + # Also update in local database + with get_session() as session: + db_conversation = ( + session.query(UserConversation) + .filter_by(id=normalized_conv_id) + .first() ) + if db_conversation: + db_conversation.topic_summary = update_request.topic_summary + session.commit() + logger.info( + "Successfully updated topic summary in local database " + "for conversation %s", + normalized_conv_id, + ) + + span.set_attribute("conversations.updated", True) + return ConversationUpdateResponse( + conversation_id=normalized_conv_id, + success=True, + message="Topic summary updated successfully", + ) - return ConversationUpdateResponse( - conversation_id=normalized_conv_id, - success=True, - message="Topic summary updated successfully", - ) + except APIConnectionError as e: + response = ServiceUnavailableResponse( + backend_name="OGX", cause=str(e) + ).model_dump() + raise HTTPException(**response) from e - except APIConnectionError as e: - response = ServiceUnavailableResponse( - backend_name="OGX", cause=str(e) - ).model_dump() - raise HTTPException(**response) from e - - except (APIStatusError, ConversationNotFoundError) as e: - # In library mode, ConversationNotFoundError is raised instead of APIStatusError - logger.error("Conversation not found: %s", e) - response = NotFoundResponse( - resource="conversation", resource_id=normalized_conv_id - ).model_dump() - raise HTTPException(**response) from e - - except SQLAlchemyError as e: - logger.error( - "Database error occurred while updating conversation %s.", - normalized_conv_id, - ) - response = InternalServerErrorResponse.database_error() - raise HTTPException(**response.model_dump()) from e + except (APIStatusError, ConversationNotFoundError) as e: + logger.error("Conversation not found: %s", e) + response = NotFoundResponse( + resource="conversation", resource_id=normalized_conv_id + ).model_dump() + raise HTTPException(**response) from e + + except SQLAlchemyError as e: + logger.error( + "Database error occurred while updating conversation %s.", + normalized_conv_id, + ) + response = InternalServerErrorResponse.database_error() + raise HTTPException(**response.model_dump()) from e diff --git a/src/app/endpoints/conversations_v2.py b/src/app/endpoints/conversations_v2.py index 1f61220da..8905303dc 100644 --- a/src/app/endpoints/conversations_v2.py +++ b/src/app/endpoints/conversations_v2.py @@ -3,6 +3,7 @@ from typing import Any from fastapi import APIRouter, Depends, HTTPException, Request +from opentelemetry import trace from authentication import get_auth_dependency from authorization.middleware import authorize @@ -34,6 +35,7 @@ from utils.suid import check_suid logger = get_logger(__name__) +tracer = trace.get_tracer(__name__) router = APIRouter(tags=["conversations_v2"]) @@ -90,23 +92,27 @@ async def get_conversations_list_endpoint_handler( auth: Any = Depends(get_auth_dependency()), ) -> ConversationsListResponseV2: """Handle request to retrieve all conversations for the authenticated user.""" - check_configuration_loaded(configuration) + with tracer.start_as_current_span("conversations_v2.list") as span: + check_configuration_loaded(configuration) - user_id = auth[0] + user_id = auth[0] - logger.info("Retrieving conversations for user %s", user_id) + logger.info("Retrieving conversations for user %s", user_id) - skip_userid_check = auth[2] + skip_userid_check = auth[2] - if configuration.conversation_cache_configuration.type is None: - logger.warning("Conversation cache is not configured") - response = InternalServerErrorResponse.cache_unavailable() - raise HTTPException(**response.model_dump()) + if configuration.conversation_cache_configuration.type is None: + logger.warning("Conversation cache is not configured") + response = InternalServerErrorResponse.cache_unavailable() + raise HTTPException(**response.model_dump()) - conversations = configuration.conversation_cache.list(user_id, skip_userid_check) - logger.info("Conversations for user %s: %s", user_id, len(conversations)) + conversations = configuration.conversation_cache.list( + user_id, skip_userid_check + ) + logger.info("Conversations for user %s: %s", user_id, len(conversations)) - return ConversationsListResponseV2(conversations=conversations) + span.set_attribute("conversations.count", len(conversations)) + return ConversationsListResponseV2(conversations=conversations) @router.get( @@ -120,32 +126,35 @@ async def get_conversation_endpoint_handler( auth: Any = Depends(get_auth_dependency()), ) -> ConversationResponse: """Handle request to retrieve a conversation identified by its ID.""" - check_configuration_loaded(configuration) - check_valid_conversation_id(conversation_id) + with tracer.start_as_current_span("conversations_v2.get") as span: + check_configuration_loaded(configuration) + check_valid_conversation_id(conversation_id) - user_id = auth[0] - logger.info("Retrieving conversation %s for user %s", conversation_id, user_id) + user_id = auth[0] + logger.info("Retrieving conversation %s for user %s", conversation_id, user_id) - skip_userid_check = auth[2] + skip_userid_check = auth[2] - if configuration.conversation_cache_configuration.type is None: - logger.warning("Conversation cache is not configured") - response = InternalServerErrorResponse.cache_unavailable() - raise HTTPException(**response.model_dump()) + if configuration.conversation_cache_configuration.type is None: + logger.warning("Conversation cache is not configured") + response = InternalServerErrorResponse.cache_unavailable() + raise HTTPException(**response.model_dump()) - check_conversation_existence(user_id, conversation_id) + check_conversation_existence(user_id, conversation_id) - conversation = configuration.conversation_cache.get( - user_id, conversation_id, skip_userid_check - ) - # Each entry in conversation is a single turn - chat_history: list[ConversationTurn] = [ - build_conversation_turn_from_cache_entry(entry) for entry in conversation - ] - - return ConversationResponse( - conversation_id=conversation_id, chat_history=chat_history - ) + conversation = configuration.conversation_cache.get( + user_id, conversation_id, skip_userid_check + ) + # Each entry in conversation is a single turn + chat_history: list[ConversationTurn] = [ + build_conversation_turn_from_cache_entry(entry) for entry in conversation + ] + + span.set_attribute("conversations.found", True) + span.set_attribute("conversations.turns.count", len(chat_history)) + return ConversationResponse( + conversation_id=conversation_id, chat_history=chat_history + ) @router.delete( @@ -158,24 +167,28 @@ async def delete_conversation_endpoint_handler( auth: Any = Depends(get_auth_dependency()), ) -> ConversationDeleteResponse: """Handle request to delete a conversation by ID.""" - check_configuration_loaded(configuration) - check_valid_conversation_id(conversation_id) + with tracer.start_as_current_span("conversations_v2.delete") as span: + check_configuration_loaded(configuration) + check_valid_conversation_id(conversation_id) - user_id = auth[0] - logger.info("Deleting conversation %s for user %s", conversation_id, user_id) + user_id = auth[0] + logger.info("Deleting conversation %s for user %s", conversation_id, user_id) - skip_userid_check = auth[2] + skip_userid_check = auth[2] - if configuration.conversation_cache_configuration.type is None: - logger.warning("Conversation cache is not configured") - response = InternalServerErrorResponse.cache_unavailable() - raise HTTPException(**response.model_dump()) + if configuration.conversation_cache_configuration.type is None: + logger.warning("Conversation cache is not configured") + response = InternalServerErrorResponse.cache_unavailable() + raise HTTPException(**response.model_dump()) - logger.info("Deleting conversation %s for user %s", conversation_id, user_id) - deleted = configuration.conversation_cache.delete( - user_id, conversation_id, skip_userid_check - ) - return ConversationDeleteResponse(deleted=deleted, conversation_id=conversation_id) + logger.info("Deleting conversation %s for user %s", conversation_id, user_id) + deleted = configuration.conversation_cache.delete( + user_id, conversation_id, skip_userid_check + ) + span.set_attribute("conversations.deleted", deleted) + return ConversationDeleteResponse( + deleted=deleted, conversation_id=conversation_id + ) @router.put("/conversations/{conversation_id}", responses=conversation_update_responses) @@ -186,41 +199,43 @@ async def update_conversation_endpoint_handler( auth: Any = Depends(get_auth_dependency()), ) -> ConversationUpdateResponse: """Handle request to update a conversation topic summary by ID.""" - check_configuration_loaded(configuration) - check_valid_conversation_id(conversation_id) - - user_id = auth[0] - logger.info( - "Updating topic summary for conversation %s for user %s", - conversation_id, - user_id, - ) + with tracer.start_as_current_span("conversations_v2.update") as span: + check_configuration_loaded(configuration) + check_valid_conversation_id(conversation_id) + + user_id = auth[0] + logger.info( + "Updating topic summary for conversation %s for user %s", + conversation_id, + user_id, + ) - skip_userid_check = auth[2] + skip_userid_check = auth[2] - if configuration.conversation_cache_configuration.type is None: - logger.warning("Conversation cache is not configured") - response = InternalServerErrorResponse.cache_unavailable() - raise HTTPException(**response.model_dump()) + if configuration.conversation_cache_configuration.type is None: + logger.warning("Conversation cache is not configured") + response = InternalServerErrorResponse.cache_unavailable() + raise HTTPException(**response.model_dump()) - check_conversation_existence(user_id, conversation_id) + check_conversation_existence(user_id, conversation_id) - # Update the topic summary in the cache - configuration.conversation_cache.set_topic_summary( - user_id, conversation_id, update_request.topic_summary, skip_userid_check - ) + # Update the topic summary in the cache + configuration.conversation_cache.set_topic_summary( + user_id, conversation_id, update_request.topic_summary, skip_userid_check + ) - logger.info( - "Successfully updated topic summary for conversation %s for user %s", - conversation_id, - user_id, - ) + logger.info( + "Successfully updated topic summary for conversation %s for user %s", + conversation_id, + user_id, + ) - return ConversationUpdateResponse( - conversation_id=conversation_id, - success=True, - message="Topic summary updated successfully", - ) + span.set_attribute("conversations.updated", True) + return ConversationUpdateResponse( + conversation_id=conversation_id, + success=True, + message="Topic summary updated successfully", + ) def check_valid_conversation_id(conversation_id: str) -> None: diff --git a/tests/unit/app/endpoints/test_conversations.py b/tests/unit/app/endpoints/test_conversations.py index 3a634bc8c..54063443f 100644 --- a/tests/unit/app/endpoints/test_conversations.py +++ b/tests/unit/app/endpoints/test_conversations.py @@ -9,6 +9,10 @@ import pytest from fastapi import HTTPException, Request, status from ogx_client import APIConnectionError, APIStatusError, NotFoundError +from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( + InMemorySpanExporter, +) +from opentelemetry.trace import StatusCode from pytest_mock import MockerFixture, MockType from sqlalchemy.exc import SQLAlchemyError @@ -30,6 +34,7 @@ ConversationsListResponse, ConversationUpdateResponse, ) +from models.common import ConversationTurn, Message from models.config import Action from models.database.conversations import UserConversation, UserTurn from tests.unit.utils.auth_helpers import mock_authorization_resolvers @@ -2162,3 +2167,364 @@ async def test_sqlalchemy_error_in_database_update( detail = exc_info.value.detail assert isinstance(detail, dict) assert "Database" in detail["response"] # pyright: ignore[reportArgumentType] + + +class TestConversationsV1Otel: + """OTEL instrumentation tests for conversations v1 endpoints.""" + + @pytest.mark.asyncio + async def test_list_span_on_success( + self, + mocker: MockerFixture, + setup_configuration: AppConfig, + dummy_request: Request, + otel: tuple[Any, InMemorySpanExporter], + ) -> None: + """Test that listing conversations emits a span with count.""" + tracer, exporter = otel + mocker.patch("app.endpoints.conversations_v1.tracer", tracer) + mock_authorization_resolvers(mocker) + mocker.patch( + "app.endpoints.conversations_v1.configuration", setup_configuration + ) + + mock_conversations = [ + create_mock_conversation( + mocker, + VALID_CONVERSATION_ID, + "2024-01-01T00:00:00Z", + "2024-01-01T00:05:00Z", + 5, + "model", + "provider", + ), + ] + mock_database_session(mocker, mock_conversations) + + response = await get_conversations_list_endpoint_handler( + auth=MOCK_AUTH, request=dummy_request + ) + + assert isinstance(response, ConversationsListResponse) + spans = exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert span.name == "conversations_v1.list" + assert span.attributes["conversations.count"] == 1 + + @pytest.mark.asyncio + async def test_list_span_records_error( + self, + mocker: MockerFixture, + dummy_request: Request, + otel: tuple[Any, InMemorySpanExporter], + ) -> None: + """Test that the list span records an error when config is not loaded.""" + tracer, exporter = otel + mocker.patch("app.endpoints.conversations_v1.tracer", tracer) + mock_authorization_resolvers(mocker) + + mock_config = AppConfig() + mocker.patch("app.endpoints.conversations_v1.configuration", mock_config) + + with pytest.raises(HTTPException): + await get_conversations_list_endpoint_handler( + auth=MOCK_AUTH, request=dummy_request + ) + + spans = exporter.get_finished_spans() + assert len(spans) == 1 + assert spans[0].name == "conversations_v1.list" + assert spans[0].status.status_code == StatusCode.ERROR + + @pytest.mark.asyncio + async def test_get_span_on_success( # pylint: disable=too-many-locals + self, + mocker: MockerFixture, + setup_configuration: AppConfig, + dummy_request: Request, + otel: tuple[Any, InMemorySpanExporter], + ) -> None: + """Test that getting a conversation emits a span with turn count.""" + tracer, exporter = otel + mocker.patch("app.endpoints.conversations_v1.tracer", tracer) + mock_authorization_resolvers(mocker) + mocker.patch( + "app.endpoints.conversations_v1.configuration", setup_configuration + ) + mocker.patch("app.endpoints.conversations_v1.check_suid", return_value=True) + mocker.patch( + "app.endpoints.conversations_v1.normalize_conversation_id", + return_value=VALID_CONVERSATION_ID, + ) + + mock_conversation = mocker.Mock() + mock_conversation.created_at = datetime(2024, 1, 1, tzinfo=UTC) + mocker.patch( + "app.endpoints.conversations_v1.validate_and_retrieve_conversation", + return_value=mock_conversation, + ) + + mocker.patch( + "app.endpoints.conversations_v1.AsyncOgxClientHolder" + ).return_value.get_client.return_value = mocker.AsyncMock() + + mocker.patch( + "app.endpoints.conversations_v1.to_llama_stack_conversation_id", + return_value=f"conv_{VALID_CONVERSATION_ID}", + ) + + mock_database_session(mocker, db_turns=[create_mock_db_turn(mocker, 1)]) + + mocker.patch( + "app.endpoints.conversations_v1.get_all_conversation_items", + return_value=[mocker.Mock(), mocker.Mock()], + ) + + mock_turns = [ + ConversationTurn( + messages=[ + Message(content="q1", type="user", referenced_documents=None), + Message(content="r1", type="assistant", referenced_documents=None), + ], + provider="p", + model="m", + started_at="2024-01-01T00:00:00Z", + completed_at="2024-01-01T00:00:05Z", + ), + ConversationTurn( + messages=[ + Message(content="q2", type="user", referenced_documents=None), + Message(content="r2", type="assistant", referenced_documents=None), + ], + provider="p", + model="m", + started_at="2024-01-01T00:00:06Z", + completed_at="2024-01-01T00:00:10Z", + ), + ] + mocker.patch( + "app.endpoints.conversations_v1.build_conversation_turns_from_items", + return_value=mock_turns, + ) + + response = await get_conversation_endpoint_handler( + request=dummy_request, + conversation_id=VALID_CONVERSATION_ID, + auth=MOCK_AUTH, + ) + + assert isinstance(response, ConversationResponse) + spans = exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert span.name == "conversations_v1.get" + assert span.attributes["conversations.found"] is True + assert span.attributes["conversations.turns.count"] == 2 + + @pytest.mark.asyncio + async def test_get_span_records_error_on_connection_failure( + self, + mocker: MockerFixture, + setup_configuration: AppConfig, + dummy_request: Request, + otel: tuple[Any, InMemorySpanExporter], + ) -> None: + """Test that the get span records an error on API connection failure.""" + tracer, exporter = otel + mocker.patch("app.endpoints.conversations_v1.tracer", tracer) + mock_authorization_resolvers(mocker) + mocker.patch( + "app.endpoints.conversations_v1.configuration", setup_configuration + ) + mocker.patch("app.endpoints.conversations_v1.check_suid", return_value=True) + mocker.patch( + "app.endpoints.conversations_v1.normalize_conversation_id", + return_value=VALID_CONVERSATION_ID, + ) + mocker.patch( + "app.endpoints.conversations_v1.validate_and_retrieve_conversation", + return_value=mocker.Mock(), + ) + mocker.patch( + "app.endpoints.conversations_v1.AsyncOgxClientHolder" + ).return_value.get_client.side_effect = APIConnectionError( + request=mocker.Mock() + ) + + mock_database_session(mocker) + + with pytest.raises(HTTPException): + await get_conversation_endpoint_handler( + request=dummy_request, + conversation_id=VALID_CONVERSATION_ID, + auth=MOCK_AUTH, + ) + + spans = exporter.get_finished_spans() + assert len(spans) == 1 + assert spans[0].name == "conversations_v1.get" + assert spans[0].status.status_code == StatusCode.ERROR + + @pytest.mark.asyncio + async def test_delete_span_on_success( + self, + mocker: MockerFixture, + setup_configuration: AppConfig, + dummy_request: Request, + otel: tuple[Any, InMemorySpanExporter], + ) -> None: + """Test that deleting a conversation emits a span with deleted flag.""" + tracer, exporter = otel + mocker.patch("app.endpoints.conversations_v1.tracer", tracer) + mock_authorization_resolvers(mocker) + mocker.patch( + "app.endpoints.conversations_v1.configuration", setup_configuration + ) + mocker.patch("app.endpoints.conversations_v1.check_suid", return_value=True) + mocker.patch( + "app.endpoints.conversations_v1.normalize_conversation_id", + return_value=VALID_CONVERSATION_ID, + ) + + mock_database_session(mocker) + mocker.patch("utils.endpoints.delete_conversation", return_value=True) + mocker.patch( + "app.endpoints.conversations_v1.delete_conversation", return_value=True + ) + + mock_client = mocker.AsyncMock() + mock_client.conversations.delete.return_value = mocker.Mock(deleted=True) + mocker.patch( + "app.endpoints.conversations_v1.AsyncOgxClientHolder" + ).return_value.get_client.return_value = mock_client + + mocker.patch( + "app.endpoints.conversations_v1.to_llama_stack_conversation_id", + return_value=f"conv_{VALID_CONVERSATION_ID}", + ) + + response = await delete_conversation_endpoint_handler( + request=dummy_request, + conversation_id=VALID_CONVERSATION_ID, + auth=MOCK_AUTH, + ) + + assert isinstance(response, ConversationDeleteResponse) + spans = exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert span.name == "conversations_v1.delete" + assert span.attributes["conversations.deleted"] is True + + @pytest.mark.asyncio + async def test_delete_span_records_error( + self, + mocker: MockerFixture, + dummy_request: Request, + otel: tuple[Any, InMemorySpanExporter], + ) -> None: + """Test that the delete span records an error when config is not loaded.""" + tracer, exporter = otel + mocker.patch("app.endpoints.conversations_v1.tracer", tracer) + mock_authorization_resolvers(mocker) + + mock_config = AppConfig() + mocker.patch("app.endpoints.conversations_v1.configuration", mock_config) + + with pytest.raises(HTTPException): + await delete_conversation_endpoint_handler( + request=dummy_request, + conversation_id=VALID_CONVERSATION_ID, + auth=MOCK_AUTH, + ) + + spans = exporter.get_finished_spans() + assert len(spans) == 1 + assert spans[0].name == "conversations_v1.delete" + assert spans[0].status.status_code == StatusCode.ERROR + + @pytest.mark.asyncio + async def test_update_span_on_success( + self, + mocker: MockerFixture, + setup_configuration: AppConfig, + dummy_request: Request, + otel: tuple[Any, InMemorySpanExporter], + ) -> None: + """Test that updating a conversation emits a span with updated flag.""" + tracer, exporter = otel + mocker.patch("app.endpoints.conversations_v1.tracer", tracer) + mock_authorization_resolvers(mocker) + mocker.patch( + "app.endpoints.conversations_v1.configuration", setup_configuration + ) + mocker.patch("app.endpoints.conversations_v1.check_suid", return_value=True) + mocker.patch( + "app.endpoints.conversations_v1.normalize_conversation_id", + return_value=VALID_CONVERSATION_ID, + ) + + mock_conversation = mocker.Mock() + mocker.patch( + "app.endpoints.conversations_v1.retrieve_conversation", + return_value=mock_conversation, + ) + + mock_database_session(mocker) + + mock_client = mocker.AsyncMock() + mocker.patch( + "app.endpoints.conversations_v1.AsyncOgxClientHolder" + ).return_value.get_client.return_value = mock_client + + mocker.patch( + "app.endpoints.conversations_v1.to_llama_stack_conversation_id", + return_value=f"conv_{VALID_CONVERSATION_ID}", + ) + + update_request = ConversationUpdateRequest(topic_summary="New topic") + + response = await update_conversation_endpoint_handler( + request=dummy_request, + conversation_id=VALID_CONVERSATION_ID, + update_request=update_request, + auth=MOCK_AUTH, + ) + + assert isinstance(response, ConversationUpdateResponse) + spans = exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert span.name == "conversations_v1.update" + assert span.attributes["conversations.updated"] is True + + @pytest.mark.asyncio + async def test_update_span_records_error( + self, + mocker: MockerFixture, + dummy_request: Request, + otel: tuple[Any, InMemorySpanExporter], + ) -> None: + """Test that the update span records an error when config is not loaded.""" + tracer, exporter = otel + mocker.patch("app.endpoints.conversations_v1.tracer", tracer) + mock_authorization_resolvers(mocker) + + mock_config = AppConfig() + mocker.patch("app.endpoints.conversations_v1.configuration", mock_config) + + update_request = ConversationUpdateRequest(topic_summary="New topic") + + with pytest.raises(HTTPException): + await update_conversation_endpoint_handler( + request=dummy_request, + conversation_id=VALID_CONVERSATION_ID, + update_request=update_request, + auth=MOCK_AUTH, + ) + + spans = exporter.get_finished_spans() + assert len(spans) == 1 + assert spans[0].name == "conversations_v1.update" + assert spans[0].status.status_code == StatusCode.ERROR diff --git a/tests/unit/app/endpoints/test_conversations_v2.py b/tests/unit/app/endpoints/test_conversations_v2.py index 621aeae53..566ff7c12 100644 --- a/tests/unit/app/endpoints/test_conversations_v2.py +++ b/tests/unit/app/endpoints/test_conversations_v2.py @@ -1,4 +1,4 @@ -# pylint: disable=redefined-outer-name +# pylint: disable=redefined-outer-name,too-many-lines """Unit tests for the /conversations REST API endpoints.""" @@ -7,6 +7,10 @@ import pytest from fastapi import HTTPException, status +from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( + InMemorySpanExporter, +) +from opentelemetry.trace import StatusCode from pydantic import HttpUrl from pytest_mock import MockerFixture, MockType @@ -975,3 +979,256 @@ async def test_with_skip_userid_check( mock_configuration.conversation_cache.set_topic_summary.assert_called_once_with( "mock_user_id", VALID_CONVERSATION_ID, "New topic summary", True ) + + +class TestConversationsV2Otel: + """OTEL instrumentation tests for conversations v2 endpoints.""" + + @pytest.mark.asyncio + async def test_list_span_on_success( + self, + mocker: MockerFixture, + mock_configuration: MockType, + otel: tuple[Any, InMemorySpanExporter], + ) -> None: + """Test that listing conversations emits a span with count.""" + tracer, exporter = otel + mocker.patch("app.endpoints.conversations_v2.tracer", tracer) + mock_authorization_resolvers(mocker) + mocker.patch("app.endpoints.conversations_v2.configuration", mock_configuration) + mock_configuration.conversation_cache.list.return_value = [ + ConversationData( + conversation_id=VALID_CONVERSATION_ID, + topic_summary="summary1", + last_message_timestamp=1704067200.0, + ), + ConversationData( + conversation_id="456e7890-e12b-34d5-a678-901234567890", + topic_summary="summary2", + last_message_timestamp=1704067201.0, + ), + ] + + await get_conversations_list_endpoint_handler( + request=mocker.Mock(), auth=MOCK_AUTH + ) + + spans = exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert span.name == "conversations_v2.list" + assert span.attributes["conversations.count"] == 2 + + @pytest.mark.asyncio + async def test_list_span_records_error( + self, + mocker: MockerFixture, + otel: tuple[Any, InMemorySpanExporter], + ) -> None: + """Test that the list span records an error when config is not loaded.""" + tracer, exporter = otel + mocker.patch("app.endpoints.conversations_v2.tracer", tracer) + mock_authorization_resolvers(mocker) + + mock_config = AppConfig() + mock_config._configuration = None # pylint: disable=protected-access + mocker.patch("app.endpoints.conversations_v2.configuration", mock_config) + + with pytest.raises(HTTPException): + await get_conversations_list_endpoint_handler( + request=mocker.Mock(), auth=MOCK_AUTH + ) + + spans = exporter.get_finished_spans() + assert len(spans) == 1 + assert spans[0].name == "conversations_v2.list" + assert spans[0].status.status_code == StatusCode.ERROR + + @pytest.mark.asyncio + async def test_get_span_on_success( + self, + mocker: MockerFixture, + mock_configuration: MockType, + otel: tuple[Any, InMemorySpanExporter], + ) -> None: + """Test that getting a conversation emits a span with turn count.""" + tracer, exporter = otel + mocker.patch("app.endpoints.conversations_v2.tracer", tracer) + mock_authorization_resolvers(mocker) + mocker.patch("app.endpoints.conversations_v2.configuration", mock_configuration) + mocker.patch("app.endpoints.conversations_v2.check_suid", return_value=True) + mock_configuration.conversation_cache.list.return_value = [ + mocker.Mock(conversation_id=VALID_CONVERSATION_ID) + ] + mock_configuration.conversation_cache.get.return_value = [ + CacheEntry( + query="q1", + response="r1", + provider="p", + model="m", + started_at="2024-01-01T00:00:00Z", + completed_at="2024-01-01T00:00:05Z", + ), + CacheEntry( + query="q2", + response="r2", + provider="p", + model="m", + started_at="2024-01-01T00:00:06Z", + completed_at="2024-01-01T00:00:10Z", + ), + ] + + await get_conversation_endpoint_handler( + request=mocker.Mock(), + conversation_id=VALID_CONVERSATION_ID, + auth=MOCK_AUTH, + ) + + spans = exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert span.name == "conversations_v2.get" + assert span.attributes["conversations.found"] is True + assert span.attributes["conversations.turns.count"] == 2 + + @pytest.mark.asyncio + async def test_get_span_records_error_on_not_found( + self, + mocker: MockerFixture, + mock_configuration: MockType, + otel: tuple[Any, InMemorySpanExporter], + ) -> None: + """Test that the get span records an error when conversation not found.""" + tracer, exporter = otel + mocker.patch("app.endpoints.conversations_v2.tracer", tracer) + mock_authorization_resolvers(mocker) + mocker.patch("app.endpoints.conversations_v2.configuration", mock_configuration) + mocker.patch("app.endpoints.conversations_v2.check_suid", return_value=True) + mock_configuration.conversation_cache.list.return_value = [] + + with pytest.raises(HTTPException): + await get_conversation_endpoint_handler( + request=mocker.Mock(), + conversation_id=VALID_CONVERSATION_ID, + auth=MOCK_AUTH, + ) + + spans = exporter.get_finished_spans() + assert len(spans) == 1 + assert spans[0].name == "conversations_v2.get" + assert spans[0].status.status_code == StatusCode.ERROR + + @pytest.mark.asyncio + async def test_delete_span_on_success( + self, + mocker: MockerFixture, + mock_configuration: MockType, + otel: tuple[Any, InMemorySpanExporter], + ) -> None: + """Test that deleting a conversation emits a span with deleted flag.""" + tracer, exporter = otel + mocker.patch("app.endpoints.conversations_v2.tracer", tracer) + mock_authorization_resolvers(mocker) + mocker.patch("app.endpoints.conversations_v2.configuration", mock_configuration) + mocker.patch("app.endpoints.conversations_v2.check_suid", return_value=True) + mock_configuration.conversation_cache.delete.return_value = True + + await delete_conversation_endpoint_handler( + request=mocker.Mock(), + conversation_id=VALID_CONVERSATION_ID, + auth=MOCK_AUTH, + ) + + spans = exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert span.name == "conversations_v2.delete" + assert span.attributes["conversations.deleted"] is True + + @pytest.mark.asyncio + async def test_delete_span_records_error( + self, + mocker: MockerFixture, + otel: tuple[Any, InMemorySpanExporter], + ) -> None: + """Test that the delete span records an error when config is not loaded.""" + tracer, exporter = otel + mocker.patch("app.endpoints.conversations_v2.tracer", tracer) + mock_authorization_resolvers(mocker) + + mock_config = AppConfig() + mock_config._configuration = None # pylint: disable=protected-access + mocker.patch("app.endpoints.conversations_v2.configuration", mock_config) + + with pytest.raises(HTTPException): + await delete_conversation_endpoint_handler( + request=mocker.Mock(), + conversation_id=VALID_CONVERSATION_ID, + auth=MOCK_AUTH, + ) + + spans = exporter.get_finished_spans() + assert len(spans) == 1 + assert spans[0].name == "conversations_v2.delete" + assert spans[0].status.status_code == StatusCode.ERROR + + @pytest.mark.asyncio + async def test_update_span_on_success( + self, + mocker: MockerFixture, + mock_configuration: MockType, + otel: tuple[Any, InMemorySpanExporter], + ) -> None: + """Test that updating a conversation emits a span with updated flag.""" + tracer, exporter = otel + mocker.patch("app.endpoints.conversations_v2.tracer", tracer) + mock_authorization_resolvers(mocker) + mocker.patch("app.endpoints.conversations_v2.configuration", mock_configuration) + mocker.patch("app.endpoints.conversations_v2.check_suid", return_value=True) + mock_configuration.conversation_cache.list.return_value = [ + mocker.Mock(conversation_id=VALID_CONVERSATION_ID) + ] + + update_request = ConversationUpdateRequest(topic_summary="New summary") + + await update_conversation_endpoint_handler( + conversation_id=VALID_CONVERSATION_ID, + update_request=update_request, + auth=MOCK_AUTH, + ) + + spans = exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert span.name == "conversations_v2.update" + assert span.attributes["conversations.updated"] is True + + @pytest.mark.asyncio + async def test_update_span_records_error_on_not_found( + self, + mocker: MockerFixture, + mock_configuration: MockType, + otel: tuple[Any, InMemorySpanExporter], + ) -> None: + """Test that the update span records an error when conversation not found.""" + tracer, exporter = otel + mocker.patch("app.endpoints.conversations_v2.tracer", tracer) + mock_authorization_resolvers(mocker) + mocker.patch("app.endpoints.conversations_v2.configuration", mock_configuration) + mocker.patch("app.endpoints.conversations_v2.check_suid", return_value=True) + mock_configuration.conversation_cache.list.return_value = [] + + update_request = ConversationUpdateRequest(topic_summary="New summary") + + with pytest.raises(HTTPException): + await update_conversation_endpoint_handler( + conversation_id=VALID_CONVERSATION_ID, + update_request=update_request, + auth=MOCK_AUTH, + ) + + spans = exporter.get_finished_spans() + assert len(spans) == 1 + assert spans[0].name == "conversations_v2.update" + assert spans[0].status.status_code == StatusCode.ERROR