From c17a0f4e1873ae598c79d6ad5d8b3b0a2bbff8fd Mon Sep 17 00:00:00 2001 From: Asuka Minato Date: Wed, 15 Jul 2026 14:42:12 +0900 Subject: [PATCH] test: move feedback service coverage to unit tests (#38927) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> --- .../services/test_feedback_service.py | 348 ------------------ .../services/test_feedback_service.py | 244 ++++++++++++ 2 files changed, 244 insertions(+), 348 deletions(-) delete mode 100644 api/tests/test_containers_integration_tests/services/test_feedback_service.py create mode 100644 api/tests/unit_tests/services/test_feedback_service.py diff --git a/api/tests/test_containers_integration_tests/services/test_feedback_service.py b/api/tests/test_containers_integration_tests/services/test_feedback_service.py deleted file mode 100644 index c2b0385fc74..00000000000 --- a/api/tests/test_containers_integration_tests/services/test_feedback_service.py +++ /dev/null @@ -1,348 +0,0 @@ -"""Unit tests for FeedbackService.""" - -import json -from datetime import datetime -from types import SimpleNamespace -from unittest import mock - -import pytest - -from models.enums import FeedbackFromSource, FeedbackRating -from models.model import App, Conversation, Message -from services.feedback_service import FeedbackService - - -def _execute_result(rows): - result = mock.Mock() - result.all.return_value = rows - return result - - -class TestFeedbackService: - """Test FeedbackService methods.""" - - @pytest.fixture - def mock_db_session(self): - """Mock database session passed explicitly to the service.""" - return mock.Mock() - - @pytest.fixture - def sample_data(self): - """Create sample data for testing.""" - app_id = "test-app-id" - - # Create mock models - app = App(id=app_id, name="Test App") - - conversation = Conversation(id="test-conversation-id", app_id=app_id, name="Test Conversation") - - message = Message( - id="test-message-id", - conversation_id="test-conversation-id", - query="What is AI?", - answer="AI is artificial intelligence.", - inputs={"query": "What is AI?"}, - created_at=datetime(2024, 1, 1, 10, 0, 0), - ) - - # Use SimpleNamespace to avoid ORM model constructor issues - user_feedback = SimpleNamespace( - id="user-feedback-id", - app_id=app_id, - conversation_id="test-conversation-id", - message_id="test-message-id", - rating=FeedbackRating.LIKE, - from_source=FeedbackFromSource.USER, - content="Great answer!", - from_end_user_id="user-123", - from_account_id=None, - from_account=None, # Mock account object - created_at=datetime(2024, 1, 1, 10, 5, 0), - ) - - admin_feedback = SimpleNamespace( - id="admin-feedback-id", - app_id=app_id, - conversation_id="test-conversation-id", - message_id="test-message-id", - rating=FeedbackRating.DISLIKE, - from_source=FeedbackFromSource.ADMIN, - content="Could be more detailed", - from_end_user_id=None, - from_account_id="admin-456", - from_account=SimpleNamespace(name="Admin User"), # Mock account object - created_at=datetime(2024, 1, 1, 10, 10, 0), - ) - - return { - "app": app, - "conversation": conversation, - "message": message, - "user_feedback": user_feedback, - "admin_feedback": admin_feedback, - } - - def test_export_feedbacks_csv_format(self, mock_db_session, sample_data): - """Test exporting feedback data in CSV format.""" - mock_db_session.execute.return_value = _execute_result( - [ - ( - sample_data["user_feedback"], - sample_data["message"], - sample_data["conversation"], - sample_data["app"], - sample_data["user_feedback"].from_account, - ) - ] - ) - - # Test CSV export - result = FeedbackService.export_feedbacks( - app_id=sample_data["app"].id, session=mock_db_session, format_type="csv" - ) - # Verify response structure - assert hasattr(result, "headers") - assert "text/csv" in result.headers["Content-Type"] - assert "attachment" in result.headers["Content-Disposition"] - - # Check CSV content - csv_content = result.get_data(as_text=True) - # Verify essential headers exist (order may include additional columns) - assert "feedback_id" in csv_content - assert "app_name" in csv_content - assert "conversation_id" in csv_content - assert sample_data["app"].name in csv_content - assert sample_data["message"].query in csv_content - - def test_export_feedbacks_json_format(self, mock_db_session, sample_data): - """Test exporting feedback data in JSON format.""" - mock_db_session.execute.return_value = _execute_result( - [ - ( - sample_data["admin_feedback"], - sample_data["message"], - sample_data["conversation"], - sample_data["app"], - sample_data["admin_feedback"].from_account, - ) - ] - ) - - # Test JSON export - result = FeedbackService.export_feedbacks( - app_id=sample_data["app"].id, session=mock_db_session, format_type="json" - ) - - # Verify response structure - assert hasattr(result, "headers") - assert "application/json" in result.headers["Content-Type"] - assert "attachment" in result.headers["Content-Disposition"] - - # Check JSON content - json_content = json.loads(result.get_data(as_text=True)) - assert "export_info" in json_content - assert "feedback_data" in json_content - assert json_content["export_info"]["app_id"] == sample_data["app"].id - assert json_content["export_info"]["total_records"] == 1 - - def test_export_feedbacks_with_filters(self, mock_db_session, sample_data): - """Test exporting feedback with various filters.""" - mock_db_session.execute.return_value = _execute_result( - [ - ( - sample_data["admin_feedback"], - sample_data["message"], - sample_data["conversation"], - sample_data["app"], - sample_data["admin_feedback"].from_account, - ) - ] - ) - - # Test with filters - result = FeedbackService.export_feedbacks( - app_id=sample_data["app"].id, - session=mock_db_session, - from_source=FeedbackFromSource.ADMIN, - rating=FeedbackRating.DISLIKE, - has_comment=True, - start_date="2024-01-01", - end_date="2024-12-31", - format_type="csv", - ) - - # Verify query was executed (filters are baked into the select statement) - assert mock_db_session.execute.called - - def test_export_feedbacks_no_data(self, mock_db_session, sample_data): - """Test exporting feedback when no data exists.""" - mock_db_session.execute.return_value = _execute_result([]) - - result = FeedbackService.export_feedbacks( - app_id=sample_data["app"].id, session=mock_db_session, format_type="csv" - ) - - # Should return an empty CSV with headers only - assert hasattr(result, "headers") - assert "text/csv" in result.headers["Content-Type"] - csv_content = result.get_data(as_text=True) - # Headers should exist (order can include additional columns) - assert "feedback_id" in csv_content - assert "app_name" in csv_content - assert "conversation_id" in csv_content - # No data rows expected - assert len([line for line in csv_content.strip().splitlines() if line.strip()]) == 1 - - def test_export_feedbacks_invalid_date_format(self, mock_db_session, sample_data): - """Test exporting feedback with invalid date format.""" - - # Test with invalid start_date - with pytest.raises(ValueError, match="Invalid start_date format"): - FeedbackService.export_feedbacks( - app_id=sample_data["app"].id, session=mock_db_session, start_date="invalid-date-format" - ) - - # Test with invalid end_date - with pytest.raises(ValueError, match="Invalid end_date format"): - FeedbackService.export_feedbacks( - app_id=sample_data["app"].id, session=mock_db_session, end_date="invalid-date-format" - ) - - def test_export_feedbacks_invalid_format(self, mock_db_session, sample_data): - """Test exporting feedback with unsupported format.""" - - with pytest.raises(ValueError, match="Unsupported format"): - FeedbackService.export_feedbacks( - app_id=sample_data["app"].id, - session=mock_db_session, - format_type="xml", # Unsupported format - ) - - def test_export_feedbacks_long_response_truncation(self, mock_db_session, sample_data): - """Test that long AI responses are truncated in export.""" - - # Create message with long response - long_message = Message( - id="long-message-id", - conversation_id="test-conversation-id", - query="What is AI?", - answer="A" * 600, # 600 character response - inputs={"query": "What is AI?"}, - created_at=datetime(2024, 1, 1, 10, 0, 0), - ) - - mock_db_session.execute.return_value = _execute_result( - [ - ( - sample_data["user_feedback"], - long_message, - sample_data["conversation"], - sample_data["app"], - sample_data["user_feedback"].from_account, - ) - ] - ) - - # Test export - result = FeedbackService.export_feedbacks( - app_id=sample_data["app"].id, session=mock_db_session, format_type="json" - ) - - # Check JSON content - json_content = json.loads(result.get_data(as_text=True)) - exported_answer = json_content["feedback_data"][0]["ai_response"] - - # Should be truncated with ellipsis - assert len(exported_answer) <= 503 # 500 + "..." - assert exported_answer.endswith("...") - assert len(exported_answer) > 500 # Should be close to limit - - def test_export_feedbacks_unicode_content(self, mock_db_session, sample_data): - """Test exporting feedback with unicode content (Chinese characters).""" - - # Create feedback with Chinese content (use SimpleNamespace to avoid ORM constructor constraints) - chinese_feedback = SimpleNamespace( - id="chinese-feedback-id", - app_id=sample_data["app"].id, - conversation_id="test-conversation-id", - message_id="test-message-id", - rating=FeedbackRating.DISLIKE, - from_source=FeedbackFromSource.USER, - content="回答不够详细，需要更多信息", - from_end_user_id="user-123", - from_account_id=None, - created_at=datetime(2024, 1, 1, 10, 5, 0), - ) - - # Create Chinese message - chinese_message = Message( - id="chinese-message-id", - conversation_id="test-conversation-id", - query="什么是人工智能？", - answer="人工智能是模拟人类智能的技术。", - inputs={"query": "什么是人工智能？"}, - created_at=datetime(2024, 1, 1, 10, 0, 0), - ) - - mock_db_session.execute.return_value = _execute_result( - [ - ( - chinese_feedback, - chinese_message, - sample_data["conversation"], - sample_data["app"], - None, - ) - ] - ) - - # Test export - result = FeedbackService.export_feedbacks( - app_id=sample_data["app"].id, session=mock_db_session, format_type="csv" - ) - - # Check that unicode content is preserved - csv_content = result.get_data(as_text=True) - assert "什么是人工智能？" in csv_content - assert "回答不够详细，需要更多信息" in csv_content - assert "人工智能是模拟人类智能的技术" in csv_content - - def test_export_feedbacks_emoji_ratings(self, mock_db_session, sample_data): - """Test that rating emojis are properly formatted in export.""" - mock_db_session.execute.return_value = _execute_result( - [ - ( - sample_data["user_feedback"], - sample_data["message"], - sample_data["conversation"], - sample_data["app"], - sample_data["user_feedback"].from_account, - ), - ( - sample_data["admin_feedback"], - sample_data["message"], - sample_data["conversation"], - sample_data["app"], - sample_data["admin_feedback"].from_account, - ), - ] - ) - - # Test export - result = FeedbackService.export_feedbacks( - app_id=sample_data["app"].id, session=mock_db_session, format_type="json" - ) - - # Check JSON content for emoji ratings - json_content = json.loads(result.get_data(as_text=True)) - feedback_data = json_content["feedback_data"] - - # Should have both feedback records - assert len(feedback_data) == 2 - - # Check that emojis are properly set - like_feedback = next(f for f in feedback_data if f["feedback_rating_raw"] == "like") - dislike_feedback = next(f for f in feedback_data if f["feedback_rating_raw"] == "dislike") - - assert like_feedback["feedback_rating"] == "👍" - assert dislike_feedback["feedback_rating"] == "👎" diff --git a/api/tests/unit_tests/services/test_feedback_service.py b/api/tests/unit_tests/services/test_feedback_service.py new file mode 100644 index 00000000000..36022936ba7 --- /dev/null +++ b/api/tests/unit_tests/services/test_feedback_service.py @@ -0,0 +1,244 @@ +"""Unit tests for feedback exports using SQLite-backed ORM rows. + +The export query joins five tables and applies its filters in SQL, so these tests +use a real SQLite session instead of stubbing ``Session.execute`` results. +""" + +import json +from datetime import datetime +from decimal import Decimal +from typing import TypedDict + +import pytest +from sqlalchemy.orm import Session + +from models.account import Account +from models.enums import FeedbackFromSource, FeedbackRating +from models.model import App, AppMode, Conversation, ConversationFromSource, IconType, Message, MessageFeedback +from services.feedback_service import FeedbackService + +APP_ID = "11111111-1111-1111-1111-111111111111" +TENANT_ID = "22222222-2222-2222-2222-222222222222" +CONVERSATION_ID = "33333333-3333-3333-3333-333333333333" +MESSAGE_ID = "44444444-4444-4444-4444-444444444444" +ACCOUNT_ID = "55555555-5555-5555-5555-555555555555" +END_USER_ID = "66666666-6666-6666-6666-666666666666" + + +class FeedbackSample(TypedDict): + app: App + conversation: Conversation + message: Message + user_feedback: MessageFeedback + admin_feedback: MessageFeedback + + +@pytest.fixture +def sample_data(sqlite_session: Session) -> FeedbackSample: + """Persist a complete user/admin feedback graph for export queries.""" + + app = App( + tenant_id=TENANT_ID, + name="Test App", + description="", + mode=AppMode.CHAT, + icon_type=IconType.EMOJI, + icon="🤖", + icon_background="#FFFFFF", + enable_site=False, + enable_api=False, + max_active_requests=0, + ) + app.id = APP_ID + conversation = Conversation( + app_id=APP_ID, + mode=AppMode.CHAT, + name="Test Conversation", + from_source=ConversationFromSource.API, + from_end_user_id=END_USER_ID, + ) + conversation.id = CONVERSATION_ID + conversation._inputs = {} + message = Message( + app_id=APP_ID, + conversation_id=CONVERSATION_ID, + query="What is AI?", + message={"role": "user", "content": "What is AI?"}, + answer="AI is artificial intelligence.", + message_unit_price=Decimal(0), + answer_unit_price=Decimal(0), + currency="USD", + from_source=ConversationFromSource.API, + from_end_user_id=END_USER_ID, + created_at=datetime(2024, 1, 1, 10, 0, 0), + ) + message.id = MESSAGE_ID + message._inputs = {"query": "What is AI?"} + account = Account(name="Admin User", email="admin@example.com") + account.id = ACCOUNT_ID + user_feedback = MessageFeedback( + app_id=APP_ID, + conversation_id=CONVERSATION_ID, + message_id=MESSAGE_ID, + rating=FeedbackRating.LIKE, + from_source=FeedbackFromSource.USER, + content="Great answer!", + from_end_user_id=END_USER_ID, + ) + user_feedback.created_at = datetime(2024, 1, 1, 10, 5, 0) + admin_feedback = MessageFeedback( + app_id=APP_ID, + conversation_id=CONVERSATION_ID, + message_id=MESSAGE_ID, + rating=FeedbackRating.DISLIKE, + from_source=FeedbackFromSource.ADMIN, + content="Could be more detailed", + from_account_id=ACCOUNT_ID, + ) + admin_feedback.created_at = datetime(2024, 1, 1, 10, 10, 0) + sqlite_session.add_all([app, conversation, message, account, user_feedback, admin_feedback]) + sqlite_session.commit() + return { + "app": app, + "conversation": conversation, + "message": message, + "user_feedback": user_feedback, + "admin_feedback": admin_feedback, + } + + +@pytest.mark.parametrize( + "sqlite_session", + [(App, Conversation, Message, MessageFeedback, Account)], + indirect=True, +) +class TestFeedbackService: + """Exercise feedback export formatting and its database-side filters.""" + + def test_export_feedbacks_csv_format(self, sqlite_session: Session, sample_data: FeedbackSample) -> None: + result = FeedbackService.export_feedbacks( + app_id=APP_ID, + session=sqlite_session, + from_source=FeedbackFromSource.USER, + format_type="csv", + ) + + assert "text/csv" in result.headers["Content-Type"] + assert "attachment" in result.headers["Content-Disposition"] + csv_content = result.get_data(as_text=True) + assert "feedback_id" in csv_content + assert "app_name" in csv_content + assert "conversation_id" in csv_content + assert sample_data["app"].name in csv_content + assert sample_data["message"].query in csv_content + assert sample_data["user_feedback"].content in csv_content + assert sample_data["admin_feedback"].content not in csv_content + + def test_export_feedbacks_json_format(self, sqlite_session: Session, sample_data: FeedbackSample) -> None: + result = FeedbackService.export_feedbacks( + app_id=APP_ID, + session=sqlite_session, + from_source=FeedbackFromSource.ADMIN, + format_type="json", + ) + + assert "application/json" in result.headers["Content-Type"] + assert "attachment" in result.headers["Content-Disposition"] + json_content = json.loads(result.get_data(as_text=True)) + assert json_content["export_info"]["app_id"] == APP_ID + assert json_content["export_info"]["total_records"] == 1 + assert json_content["feedback_data"][0]["from_account_name"] == "Admin User" + + def test_export_feedbacks_with_filters(self, sqlite_session: Session, sample_data: FeedbackSample) -> None: + result = FeedbackService.export_feedbacks( + app_id=APP_ID, + session=sqlite_session, + from_source=FeedbackFromSource.ADMIN, + rating=FeedbackRating.DISLIKE, + has_comment=True, + start_date="2024-01-01", + end_date="2024-12-31", + format_type="csv", + ) + + csv_content = result.get_data(as_text=True) + assert sample_data["admin_feedback"].content in csv_content + assert sample_data["user_feedback"].content not in csv_content + assert len([line for line in csv_content.strip().splitlines() if line.strip()]) == 2 + + def test_export_feedbacks_no_data(self, sqlite_session: Session, sample_data: FeedbackSample) -> None: + result = FeedbackService.export_feedbacks( + app_id="77777777-7777-7777-7777-777777777777", + session=sqlite_session, + format_type="csv", + ) + + csv_content = result.get_data(as_text=True) + assert "text/csv" in result.headers["Content-Type"] + assert "feedback_id" in csv_content + assert "app_name" in csv_content + assert "conversation_id" in csv_content + assert len([line for line in csv_content.strip().splitlines() if line.strip()]) == 1 + + def test_export_feedbacks_invalid_date_format(self, sqlite_session: Session, sample_data: FeedbackSample) -> None: + with pytest.raises(ValueError, match="Invalid start_date format"): + FeedbackService.export_feedbacks( + app_id=APP_ID, + session=sqlite_session, + start_date="invalid-date-format", + ) + + with pytest.raises(ValueError, match="Invalid end_date format"): + FeedbackService.export_feedbacks( + app_id=APP_ID, + session=sqlite_session, + end_date="invalid-date-format", + ) + + def test_export_feedbacks_invalid_format(self, sqlite_session: Session, sample_data: FeedbackSample) -> None: + with pytest.raises(ValueError, match="Unsupported format"): + FeedbackService.export_feedbacks(app_id=APP_ID, session=sqlite_session, format_type="xml") + + def test_export_feedbacks_long_response_truncation( + self, sqlite_session: Session, sample_data: FeedbackSample + ) -> None: + sample_data["message"].answer = "A" * 600 + sqlite_session.commit() + + result = FeedbackService.export_feedbacks( + app_id=APP_ID, + session=sqlite_session, + from_source=FeedbackFromSource.USER, + format_type="json", + ) + + exported_answer = json.loads(result.get_data(as_text=True))["feedback_data"][0]["ai_response"] + assert exported_answer == "A" * 500 + "..." + + def test_export_feedbacks_unicode_content(self, sqlite_session: Session, sample_data: FeedbackSample) -> None: + sample_data["message"].query = "什么是人工智能？" + sample_data["message"].answer = "人工智能是模拟人类智能的技术。" + sample_data["user_feedback"].content = "回答不够详细，需要更多信息" + sqlite_session.commit() + + result = FeedbackService.export_feedbacks( + app_id=APP_ID, + session=sqlite_session, + from_source=FeedbackFromSource.USER, + format_type="csv", + ) + + csv_content = result.get_data(as_text=True) + assert "什么是人工智能？" in csv_content + assert "回答不够详细，需要更多信息" in csv_content + assert "人工智能是模拟人类智能的技术" in csv_content + + def test_export_feedbacks_emoji_ratings(self, sqlite_session: Session, sample_data: FeedbackSample) -> None: + result = FeedbackService.export_feedbacks(app_id=APP_ID, session=sqlite_session, format_type="json") + + feedback_data = json.loads(result.get_data(as_text=True))["feedback_data"] + assert len(feedback_data) == 2 + like_feedback = next(item for item in feedback_data if item["feedback_rating_raw"] == "like") + dislike_feedback = next(item for item in feedback_data if item["feedback_rating_raw"] == "dislike") + assert like_feedback["feedback_rating"] == "👍" + assert dislike_feedback["feedback_rating"] == "👎"