import json from datetime import UTC, datetime from decimal import Decimal from unittest.mock import patch from uuid import UUID, uuid4 import pytest from sqlalchemy import inspect, select from sqlalchemy.orm import Session from core.workflow.file_reference import build_file_reference from graphon.file import FILE_MODEL_IDENTITY, FileTransferMethod from models.account import Account, Tenant from models.base import Base, TypeBase from models.dataset import DatasetCollectionBinding from models.enums import ( ApiTokenType, CollectionBindingType, ConversationFromSource, ConversationStatus, CustomizeTokenStrategy, EndUserType, FeedbackFromSource, FeedbackRating, MessageStatus, ) from models.model import ( AccountTrialAppRecord, AnnotationReplyConfig, ApiToken, App, AppAnnotationHitHistory, AppAnnotationSetting, AppMCPServer, AppMode, AppModelConfig, AppStar, Conversation, EndUser, IconType, InstalledApp, Message, MessageAnnotation, MessageFeedback, RecommendedApp, Site, load_annotation_reply_config, ) from models.workflow import Workflow, WorkflowType """Regression coverage for the ``@property``→session-parameter refactor on ``AppAnnotationHitHistory`` accessors. Covers two of `api/models/model.py`'s legacy `@property` accessors that reached for the global ``db.session`` internally and have been converted to plain methods taking an explicit ``session: Session`` (per the pattern established in #40370/#40797/#41394/#41830, tracked in #40372): - ``AppAnnotationHitHistory.account`` (joins through ``MessageAnnotation.account_id``) - ``AppAnnotationHitHistory.annotation_create_account`` (looks up ``self.account_id`` directly) Each accessor is exercised against the real ``sqlite_session`` fixture (a genuine SQLAlchemy ``Session`` bound to a pristine full-schema SQLite database) so the assertions cover actual query behaviour rather than a mock's recorded call. """ def _annotation_hit_history_session_accessors_persist_account(session: Session) -> Account: account = Account(name="Test Account", email="test@example.com") session.add(account) session.flush() return account def _annotation_hit_history_session_accessors_persist_annotation( session: Session, *, app_id: str, account_id: str ) -> MessageAnnotation: annotation = MessageAnnotation( app_id=app_id, conversation_id=None, message_id=None, question="What is AI?", content="AI stands for Artificial Intelligence.", account_id=account_id, ) session.add(annotation) session.flush() return annotation def _hit_history(*, app_id: str, annotation_id: str, account_id: str) -> AppAnnotationHitHistory: return AppAnnotationHitHistory( app_id=app_id, annotation_id=annotation_id, source="api", question="What is AI?", account_id=account_id, score=0.95, message_id=str(uuid4()), annotation_question="What is AI?", annotation_content="AI stands for Artificial Intelligence.", ) class TestAnnotationHitHistorySessionAccessorsCases: class TestAppAnnotationHitHistoryAccount: def test_returns_the_annotation_creator(self, sqlite_session: Session) -> None: app_id = str(uuid4()) creator = _annotation_hit_history_session_accessors_persist_account(sqlite_session) annotation = _annotation_hit_history_session_accessors_persist_annotation( sqlite_session, app_id=app_id, account_id=creator.id ) history = _hit_history(app_id=app_id, annotation_id=annotation.id, account_id=str(uuid4())) result = history.account(session=sqlite_session) assert result is not None assert result.id == creator.id def test_returns_none_when_annotation_missing(self, sqlite_session: Session) -> None: history = _hit_history(app_id=str(uuid4()), annotation_id=str(uuid4()), account_id=str(uuid4())) assert history.account(session=sqlite_session) is None class TestAppAnnotationHitHistoryAnnotationCreateAccount: def test_returns_the_hit_requester(self, sqlite_session: Session) -> None: app_id = str(uuid4()) requester = _annotation_hit_history_session_accessors_persist_account(sqlite_session) annotation = _annotation_hit_history_session_accessors_persist_annotation( sqlite_session, app_id=app_id, account_id=str(uuid4()) ) history = _hit_history(app_id=app_id, annotation_id=annotation.id, account_id=requester.id) result = history.annotation_create_account(session=sqlite_session) assert result is not None assert result.id == requester.id def test_returns_none_when_account_missing(self, sqlite_session: Session) -> None: history = _hit_history(app_id=str(uuid4()), annotation_id=str(uuid4()), account_id=str(uuid4())) assert history.annotation_create_account(session=sqlite_session) is None """Regression coverage for the ``@property``→session-parameter refactor on ``AppAnnotationSetting.collection_binding_detail``. The legacy ``@property`` reached for the global ``db.session`` internally and has been converted to a plain method taking an explicit ``session: Session`` (per the pattern established in #40370/#40797/#41394/#41830, tracked in #40372). The accessor is exercised against the real ``sqlite_session`` fixture (a genuine SQLAlchemy ``Session`` bound to a pristine full-schema SQLite database) so the assertions cover actual query behaviour rather than a mock's recorded call. """ def _persist_collection_binding(session: Session) -> DatasetCollectionBinding: binding = DatasetCollectionBinding( collection_name="test_collection", provider_name="test_provider", model_name="test_model", type=CollectionBindingType.DATASET, ) session.add(binding) session.flush() return binding def _persist_setting(session: Session, *, collection_binding_id: str) -> AppAnnotationSetting: setting = AppAnnotationSetting( app_id=str(uuid4()), score_threshold=0.8, collection_binding_id=collection_binding_id, created_user_id=str(uuid4()), updated_user_id=str(uuid4()), ) session.add(setting) session.flush() return setting class TestAppAnnotationSettingSessionAccessorsCases: class TestAppAnnotationSettingCollectionBindingDetail: def test_returns_the_bound_collection_binding(self, sqlite_session: Session) -> None: binding = _persist_collection_binding(sqlite_session) setting = _persist_setting(sqlite_session, collection_binding_id=binding.id) result = setting.collection_binding_detail(session=sqlite_session) assert result is not None assert result.id == binding.id def test_returns_none_when_binding_missing(self, sqlite_session: Session) -> None: setting = _persist_setting(sqlite_session, collection_binding_id=str(uuid4())) assert setting.collection_binding_detail(session=sqlite_session) is None """Regression coverage for the ``db.session``→session-parameter refactor on ``AppMCPServer.generate_server_code``. The static method reached for the global ``db.session`` internally and has been converted to take an explicit ``session: Session`` (per the pattern established in #40370/#40797/#41394, tracked in #40372). """ def _persist_server(session: Session, *, server_code: str) -> AppMCPServer: server = AppMCPServer( tenant_id=str(uuid4()), app_id=str(uuid4()), name="Test MCP Server", description="Test server", server_code=server_code, status="active", parameters="{}", ) session.add(server) session.flush() return server class TestAppMcpServerSessionAccessorsCases: @staticmethod def test_generate_server_code_retries_a_taken_code_through_the_given_session(sqlite_session: Session) -> None: _persist_server(sqlite_session, server_code="taken-code") with patch("models.model.generate_string", side_effect=["taken-code", "fresh-code"]): result = AppMCPServer.generate_server_code(16, session=sqlite_session) assert result == "fresh-code" """ Comprehensive unit tests for App models. This test suite covers: - App configuration validation - App-Message relationships - Conversation model integrity - Annotation model relationships """ class TestAppModelsCases: class TestAppModelValidation: """Test suite for App model validation and basic operations.""" def test_app_creation_with_required_fields(self) -> None: """Test creating an app with all required fields.""" # Arrange tenant_id = str(uuid4()) created_by = str(uuid4()) # Act app = App( tenant_id=tenant_id, name="Test App", mode=AppMode.CHAT, enable_site=True, enable_api=False, created_by=created_by, ) # Assert assert app.name == "Test App" assert app.tenant_id == tenant_id assert app.mode == AppMode.CHAT assert app.enable_site is True assert app.enable_api is False assert app.created_by == created_by def test_app_creation_with_optional_fields(self) -> None: """Test creating an app with optional fields.""" # Arrange & Act app = App( tenant_id=str(uuid4()), name="Test App", mode=AppMode.COMPLETION, enable_site=True, enable_api=True, created_by=str(uuid4()), description="Test description", icon_type=IconType.EMOJI, icon="🤖", icon_background="#FF5733", is_demo=True, is_public=False, api_rpm=100, api_rph=1000, ) # Assert assert app.description == "Test description" assert app.icon_type == IconType.EMOJI assert app.icon == "🤖" assert app.icon_background == "#FF5733" assert app.is_demo is True assert app.is_public is False assert app.api_rpm == 100 assert app.api_rph == 1000 def test_app_mode_validation(self) -> None: """Test app mode enum values.""" # Assert expected_modes = { "chat", "completion", "workflow", "advanced-chat", "agent-chat", "agent", "channel", "rag-pipeline", } assert {mode.value for mode in AppMode} == expected_modes def test_app_mode_value_of(self) -> None: """Test AppMode.value_of method.""" # Act & Assert assert AppMode.value_of("chat") == AppMode.CHAT assert AppMode.value_of("completion") == AppMode.COMPLETION assert AppMode.value_of("workflow") == AppMode.WORKFLOW with pytest.raises(ValueError, match="invalid mode value"): AppMode.value_of("invalid_mode") def test_icon_type_validation(self) -> None: """Test icon type enum values.""" # Assert assert {t.value for t in IconType} == {"image", "emoji", "link"} @pytest.mark.parametrize("sqlite_session", [(App, AppModelConfig)], indirect=True) def test_app_desc_or_prompt_with_description(self, sqlite_session: Session) -> None: """Test desc_or_prompt_with_session when description exists.""" # Arrange app = App( tenant_id=str(uuid4()), name="Test App", mode=AppMode.CHAT, enable_site=True, enable_api=False, created_by=str(uuid4()), description="App description", ) # Act result = app.desc_or_prompt_with_session(session=sqlite_session) # Assert assert result == "App description" @pytest.mark.parametrize("sqlite_session", [(App, AppModelConfig)], indirect=True) def test_app_desc_or_prompt_without_description(self, sqlite_session: Session) -> None: """Test desc_or_prompt property when description is empty.""" # Arrange app = App( tenant_id=str(uuid4()), name="Test App", mode=AppMode.CHAT, enable_site=True, enable_api=False, created_by=str(uuid4()), description="", ) decoy_config = AppModelConfig(app_id=str(uuid4()), pre_prompt="A decoy prompt") sqlite_session.add_all([app, decoy_config]) sqlite_session.flush() # Act result = app.desc_or_prompt_with_session(session=sqlite_session) # Assert assert result == "" @pytest.mark.parametrize("sqlite_session", [(App, AppModelConfig)], indirect=True) def test_app_is_agent_false_when_not_configured_as_agent(self, sqlite_session: Session) -> None: """`is_agent_with_session` returns False when the config has no agent mode.""" # Arrange app = App( tenant_id=str(uuid4()), name="Test App", mode=AppMode.CHAT, enable_site=True, enable_api=False, created_by=str(uuid4()), app_model_config_id=str(uuid4()), ) sqlite_session.add(app) sqlite_session.flush() # Act result = app.is_agent_with_session(session=sqlite_session) # Assert assert result is False @pytest.mark.parametrize("sqlite_session", [(App, AppModelConfig)], indirect=True) def test_app_is_agent_with_session_persists_mode_across_sessions(self, sqlite_session: Session) -> None: app = App( tenant_id=str(uuid4()), name="Test App", mode=AppMode.CHAT, enable_site=True, enable_api=False, created_by=str(uuid4()), ) sqlite_session.add(app) sqlite_session.flush() model_config = AppModelConfig( app_id=app.id, agent_mode=json.dumps({"enabled": True, "strategy": "react"}), ) sqlite_session.add(model_config) sqlite_session.flush() app.app_model_config_id = model_config.id sqlite_session.commit() with Session(sqlite_session.get_bind(), expire_on_commit=False) as migration_session: assert app.is_agent_with_session(session=migration_session) is True sqlite_session.expire_all() persisted_app = sqlite_session.get(App, app.id) assert persisted_app is not None assert persisted_app.mode == AppMode.AGENT_CHAT @pytest.mark.parametrize("sqlite_session", [(App, AppModelConfig)], indirect=True) def test_app_mode_compatible_with_agent(self, sqlite_session: Session) -> None: """Test mode_compatible_with_agent property.""" # Arrange app = App( tenant_id=str(uuid4()), name="Test App", mode=AppMode.CHAT, enable_site=True, enable_api=False, created_by=str(uuid4()), ) decoy_config = AppModelConfig( app_id=str(uuid4()), agent_mode=json.dumps({"enabled": True, "strategy": "react"}), ) sqlite_session.add_all([app, decoy_config]) sqlite_session.flush() # Act result = app.mode_compatible_with_agent_with_session(session=sqlite_session) # Assert assert result == AppMode.CHAT @pytest.mark.parametrize("sqlite_session", [(App, AppModelConfig)], indirect=True) def test_app_mode_compatible_with_agent_reports_agent_chat(self, sqlite_session: Session) -> None: """A CHAT app whose own config enables agent mode reports AGENT_CHAT.""" # Arrange app = App( tenant_id=str(uuid4()), name="Test App", mode=AppMode.CHAT, enable_site=True, enable_api=False, created_by=str(uuid4()), ) sqlite_session.add(app) sqlite_session.flush() app.app_model_config_id = str(uuid4()) config = AppModelConfig( app_id=app.id, agent_mode=json.dumps({"enabled": True, "strategy": "react"}), ) app_model_config_id = app.app_model_config_id assert app_model_config_id is not None config.id = app_model_config_id sqlite_session.add(config) sqlite_session.flush() # Act result = app.mode_compatible_with_agent_with_session(session=sqlite_session) # Assert assert result == AppMode.AGENT_CHAT @pytest.mark.parametrize("sqlite_session", [(App, AppModelConfig)], indirect=True) def test_app_model_config_with_session_reads_through_the_caller_session(self, sqlite_session: Session) -> None: """`app_model_config_with_session` resolves the linked config through the caller's session.""" # Arrange app = App( tenant_id=str(uuid4()), name="Test App", mode=AppMode.CHAT, enable_site=True, enable_api=False, created_by=str(uuid4()), ) sqlite_session.add(app) sqlite_session.flush() model_config = AppModelConfig(app_id=app.id, agent_mode=json.dumps({"enabled": False})) sqlite_session.add(model_config) sqlite_session.flush() app.app_model_config_id = model_config.id sqlite_session.commit() # Act: a session the model never owns, proving the lookup does not reach for a global one with Session(sqlite_session.get_bind(), expire_on_commit=False) as caller_session: result = app.app_model_config_with_session(session=caller_session) # Assert assert result is not None assert result.id == model_config.id assert result in caller_session @pytest.mark.parametrize("sqlite_session", [(App, AppModelConfig)], indirect=True) def test_app_model_config_with_session_returns_none_when_unlinked(self, sqlite_session: Session) -> None: """An app without `app_model_config_id` resolves to None without querying.""" # Arrange app = App( tenant_id=str(uuid4()), name="Test App", mode=AppMode.CHAT, enable_site=True, enable_api=False, created_by=str(uuid4()), ) sqlite_session.add(app) sqlite_session.flush() # Act / Assert: an unbound session would raise if the guard were dropped with Session(expire_on_commit=False) as unbound_session: assert app.app_model_config_with_session(session=unbound_session) is None @pytest.mark.parametrize("sqlite_session", [(App, AppModelConfig)], indirect=True) def test_workflow_with_session_reads_through_the_caller_session(self, sqlite_session: Session) -> None: """`workflow_with_session` resolves the published workflow through the caller's session.""" # Arrange app = App( tenant_id=str(uuid4()), name="Test App", mode=AppMode.WORKFLOW, enable_site=True, enable_api=False, created_by=str(uuid4()), ) sqlite_session.add(app) sqlite_session.flush() created_by = app.created_by assert created_by is not None workflow = Workflow( tenant_id=app.tenant_id, app_id=app.id, type=WorkflowType.WORKFLOW, version="1", graph=json.dumps({"nodes": [], "edges": []}), created_by=created_by, ) workflow._features = "{}" sqlite_session.add(workflow) sqlite_session.flush() app.workflow_id = workflow.id sqlite_session.commit() # Act: a session the model never owns, proving the lookup does not reach for a global one with Session(sqlite_session.get_bind(), expire_on_commit=False) as caller_session: result = app.workflow_with_session(session=caller_session) # Assert assert result is not None assert result.id == workflow.id assert result in caller_session @pytest.mark.parametrize("sqlite_session", [(App, AppModelConfig)], indirect=True) def test_workflow_with_session_returns_none_when_unpublished(self, sqlite_session: Session) -> None: """An app without `workflow_id` resolves to None without querying.""" # Arrange app = App( tenant_id=str(uuid4()), name="Test App", mode=AppMode.WORKFLOW, enable_site=True, enable_api=False, created_by=str(uuid4()), ) sqlite_session.add(app) sqlite_session.flush() # Act / Assert: an unbound session would raise if the guard were dropped with Session(expire_on_commit=False) as unbound_session: assert app.workflow_with_session(session=unbound_session) is None @pytest.mark.parametrize("sqlite_session", [(App, AppModelConfig)], indirect=True) def test_deleted_tools_checks_plugin_builtin_providers_through_core_plugin_service( self, sqlite_session: Session ) -> None: """Plugin-backed built-in tools are checked through core PluginService.""" # Arrange app = App( tenant_id="tenant-1", name="Test App", mode=AppMode.CHAT, enable_site=True, enable_api=False, created_by=str(uuid4()), ) sqlite_session.add(app) sqlite_session.flush() app_model_config = AppModelConfig( app_id=app.id, agent_mode=json.dumps( { "enabled": True, "strategy": "function_call", "tools": [ { "provider_type": "builtin", "provider_id": "langgenius/openai/openai", "tool_name": "chat", "tool_parameters": {}, } ], "prompt": None, } ), ) sqlite_session.add(app_model_config) sqlite_session.flush() app.app_model_config_id = app_model_config.id sqlite_session.flush() # Act with ( patch("core.tools.tool_manager.ToolManager.get_hardcoded_provider", side_effect=Exception), patch("core.plugin.plugin_service.PluginService.check_tools_existence", return_value=[False]) as exists, ): result = app.deleted_tools_with_session(session=sqlite_session) # Assert assert result == [{"type": "builtin", "tool_name": "chat", "provider_id": "langgenius/openai/openai"}] exists.assert_called_once() assert exists.call_args.args[0] == "tenant-1" assert [str(provider_id) for provider_id in exists.call_args.args[1]] == ["langgenius/openai/openai"] class TestAppModelConfig: """Test suite for AppModelConfig model.""" def test_legacy_dataset_configs_default_retrieval_model_without_dropping_settings(self) -> None: config = AppModelConfig( app_id=str(uuid4()), dataset_configs=json.dumps({"reranking_enable": True, "top_k": 8}), ) assert config.dataset_configs_dict == { "retrieval_model": "single", "reranking_enable": True, "top_k": 8, } def test_app_model_config_creation(self) -> None: """Test creating an AppModelConfig.""" # Arrange app_id = str(uuid4()) created_by = str(uuid4()) # Act config = AppModelConfig( app_id=app_id, provider="openai", model_id="gpt-4", created_by=created_by, ) # Assert assert config.app_id == app_id assert config.provider == "openai" assert config.model_id == "gpt-4" assert config.created_by == created_by def test_app_model_config_with_configs_json(self) -> None: """Test AppModelConfig with JSON configs.""" # Arrange configs = {"temperature": 0.7, "max_tokens": 1000} # Act config = AppModelConfig( app_id=str(uuid4()), provider="openai", model_id="gpt-4", created_by=str(uuid4()), configs=configs, ) # Assert assert config.configs == configs def test_app_model_config_model_dict_property(self) -> None: """Test model_dict property.""" # Arrange model_data = {"provider": "openai", "name": "gpt-4"} config = AppModelConfig( app_id=str(uuid4()), provider="openai", model_id="gpt-4", created_by=str(uuid4()), model=json.dumps(model_data), ) # Act result = config.model_dict # Assert assert result == model_data def test_app_model_config_model_dict_empty(self) -> None: """Test model_dict property when model is None.""" # Arrange config = AppModelConfig( app_id=str(uuid4()), provider="openai", model_id="gpt-4", created_by=str(uuid4()), model=None, ) # Act result = config.model_dict # Assert assert result == {} def test_app_model_config_suggested_questions_list(self) -> None: """Test suggested_questions_list property.""" # Arrange questions = ["What can you do?", "How does this work?"] config = AppModelConfig( app_id=str(uuid4()), provider="openai", model_id="gpt-4", created_by=str(uuid4()), suggested_questions=json.dumps(questions), ) # Act result = config.suggested_questions_list # Assert assert result == questions def test_to_dict_uses_injected_annotation_reply(self) -> None: config = AppModelConfig(app_id=str(uuid4())) annotation_reply: AnnotationReplyConfig = {"enabled": False} result = config.to_dict(annotation_reply=annotation_reply) assert result["annotation_reply"] == annotation_reply class TestAnnotationReplyConfigLoader: @pytest.mark.parametrize("sqlite_session", [(AppAnnotationSetting, DatasetCollectionBinding)], indirect=True) def test_load_annotation_reply_config_returns_disabled_when_setting_missing( self, sqlite_session: Session ) -> None: binding = DatasetCollectionBinding( provider_name="decoy-provider", model_name="decoy-model", type=CollectionBindingType.ANNOTATION, collection_name="decoy-collection", ) sqlite_session.add(binding) sqlite_session.flush() sqlite_session.add( AppAnnotationSetting( app_id="other-app", score_threshold=0.9, collection_binding_id=binding.id, created_user_id="user-1", updated_user_id="user-1", ) ) sqlite_session.flush() result = load_annotation_reply_config(sqlite_session, "app-1") assert result == {"enabled": False} @pytest.mark.parametrize("sqlite_session", [(AppAnnotationSetting, DatasetCollectionBinding)], indirect=True) def test_load_annotation_reply_config_returns_embedding_model(self, sqlite_session: Session) -> None: collection_binding = DatasetCollectionBinding( provider_name="provider", model_name="embedding", type=CollectionBindingType.ANNOTATION, collection_name="annotation-collection", ) sqlite_session.add(collection_binding) sqlite_session.flush() annotation_setting = AppAnnotationSetting( app_id="app-1", score_threshold=0.7, collection_binding_id=collection_binding.id, created_user_id="user-1", updated_user_id="user-1", ) sqlite_session.add(annotation_setting) sqlite_session.flush() result = load_annotation_reply_config(sqlite_session, "app-1") assert result == { "id": annotation_setting.id, "enabled": True, "score_threshold": 0.7, "embedding_model": { "embedding_provider_name": "provider", "embedding_model_name": "embedding", }, } @pytest.mark.parametrize("sqlite_session", [(AppAnnotationSetting, DatasetCollectionBinding)], indirect=True) def test_load_annotation_reply_config_raises_when_binding_missing(self, sqlite_session: Session) -> None: sqlite_session.add( AppAnnotationSetting( app_id="app-1", score_threshold=0.7, collection_binding_id=str(uuid4()), created_user_id="user-1", updated_user_id="user-1", ) ) sqlite_session.flush() with pytest.raises(ValueError, match="Collection binding detail not found"): load_annotation_reply_config(sqlite_session, "app-1") class TestConversationModel: """Test suite for Conversation model integrity.""" def test_app_returns_matching_app_from_caller_session(self, sqlite_session: Session) -> None: other_app = App( tenant_id=str(uuid4()), name="Other app", mode=AppMode.CHAT, enable_site=False, enable_api=False ) app = App( tenant_id=str(uuid4()), name="Conversation app", mode=AppMode.CHAT, enable_site=False, enable_api=False ) sqlite_session.add_all([other_app, app]) sqlite_session.flush() conversation = Conversation(app_id=app.id) assert conversation.app(session=sqlite_session) is app def test_app_returns_none_when_app_is_missing(self, sqlite_session: Session) -> None: other_app = App( tenant_id=str(uuid4()), name="Other app", mode=AppMode.CHAT, enable_site=False, enable_api=False ) sqlite_session.add(other_app) sqlite_session.flush() conversation = Conversation(app_id=str(uuid4())) assert conversation.app(session=sqlite_session) is None def test_conversation_creation_with_required_fields(self) -> None: """Test creating a conversation with required fields.""" # Arrange app_id = str(uuid4()) from_end_user_id = str(uuid4()) # Act conversation = Conversation( app_id=app_id, mode=AppMode.CHAT, name="Test Conversation", status=ConversationStatus.NORMAL, from_source=ConversationFromSource.API, from_end_user_id=from_end_user_id, ) # Assert assert conversation.app_id == app_id assert conversation.mode == AppMode.CHAT assert conversation.name == "Test Conversation" assert conversation.status == "normal" assert conversation.from_source == "api" assert conversation.from_end_user_id == from_end_user_id def test_conversation_with_inputs(self, sqlite_session: Session) -> None: """Test conversation inputs round-trip through the session-aware accessor.""" # Arrange inputs = {"query": "Hello", "context": "test"} conversation = Conversation( app_id=str(uuid4()), mode=AppMode.CHAT, name="Test Conversation", status=ConversationStatus.NORMAL, from_source=ConversationFromSource.API, from_end_user_id=str(uuid4()), ) conversation._inputs = inputs # Act result = conversation.inputs_with_session(session=sqlite_session) # Assert assert result == inputs def test_conversation_inputs_is_write_only(self) -> None: """Reading conversation.inputs must fail loudly; reads go through inputs_with_session.""" conversation = Conversation(app_id=str(uuid4()), _inputs={}) with pytest.raises(AttributeError, match="no getter"): _ = conversation.inputs def test_conversation_inputs_setter(self) -> None: """Test conversation inputs setter.""" # Arrange conversation = Conversation( app_id=str(uuid4()), mode=AppMode.CHAT, name="Test Conversation", status=ConversationStatus.NORMAL, from_source=ConversationFromSource.API, from_end_user_id=str(uuid4()), ) inputs = {"query": "Hello", "context": "test"} # Act conversation.inputs = inputs # Assert assert conversation._inputs == inputs @pytest.mark.parametrize("sqlite_session", [(Conversation, Message)], indirect=True) def test_conversation_summary_or_query_with_summary(self, sqlite_session: Session) -> None: """Test summary_or_query_with_session when summary exists.""" # Arrange conversation = Conversation( app_id=str(uuid4()), mode=AppMode.CHAT, name="Test Conversation", status=ConversationStatus.NORMAL, from_source=ConversationFromSource.API, from_end_user_id=str(uuid4()), summary="Test summary", ) # Act result = conversation.summary_or_query_with_session(session=sqlite_session) # Assert assert result == "Test summary" @pytest.mark.parametrize("sqlite_session", [(Conversation, Message)], indirect=True) def test_conversation_summary_or_query_without_summary(self, sqlite_session: Session) -> None: """Test summary_or_query_with_session when summary is empty.""" # Arrange conversation = Conversation( app_id=str(uuid4()), mode=AppMode.CHAT, name="Test Conversation", status=ConversationStatus.NORMAL, from_source=ConversationFromSource.API, from_end_user_id=str(uuid4()), summary=None, ) conversation._inputs = {} sqlite_session.add(conversation) sqlite_session.flush() first_message = Message( app_id=conversation.app_id, conversation_id=conversation.id, query="First message query", message={"role": "user", "content": "First message query"}, answer="First answer", message_unit_price=Decimal(0), answer_unit_price=Decimal(0), currency="USD", from_source=ConversationFromSource.API, ) first_message.created_at = datetime(2024, 1, 1) first_message._inputs = {} later_message = Message( app_id=conversation.app_id, conversation_id=conversation.id, query="Later message query", message={"role": "user", "content": "Later message query"}, answer="Later answer", message_unit_price=Decimal(0), answer_unit_price=Decimal(0), currency="USD", from_source=ConversationFromSource.API, ) later_message.created_at = datetime(2024, 1, 2) later_message._inputs = {} sqlite_session.add_all([later_message, first_message]) sqlite_session.flush() # Act result = conversation.summary_or_query_with_session(session=sqlite_session) # Assert assert result == "First message query" @pytest.mark.parametrize( "sqlite_session", [(Conversation, AppModelConfig, AppAnnotationSetting)], indirect=True ) def test_model_config_uses_caller_session_for_annotation_reply(self, sqlite_session: Session) -> None: app_id = str(uuid4()) app_model_config = AppModelConfig(app_id=app_id, pre_prompt="Persisted prompt") sqlite_session.add(app_model_config) sqlite_session.flush() conversation = Conversation( app_id=app_id, app_model_config_id=app_model_config.id, mode=AppMode.CHAT, name="Test Conversation", status=ConversationStatus.NORMAL, from_source=ConversationFromSource.API, from_end_user_id=str(uuid4()), model_id="model-1", model_provider="provider-1", ) conversation._inputs = {} sqlite_session.add(conversation) sqlite_session.flush() result = conversation.model_config_with_session(session=sqlite_session) assert result["annotation_reply"] == {"enabled": False} assert result["pre_prompt"] == "Persisted prompt" assert result["model_id"] == "model-1" assert result["provider"] == "provider-1" @pytest.mark.parametrize("sqlite_session", [(Conversation, AppAnnotationSetting)], indirect=True) def test_override_model_config_uses_caller_session_for_annotation_reply(self, sqlite_session: Session) -> None: app_id = str(uuid4()) conversation = Conversation( app_id=app_id, mode=AppMode.CHAT, name="Test Conversation", status=ConversationStatus.NORMAL, from_source=ConversationFromSource.API, from_end_user_id=str(uuid4()), override_model_configs=json.dumps({"model": {"provider": "openai", "name": "gpt-4"}}), model_id="model-1", model_provider="provider-1", ) conversation._inputs = {} sqlite_session.add(conversation) sqlite_session.flush() result = conversation.model_config_with_session(session=sqlite_session) assert result["annotation_reply"] == {"enabled": False} assert result["model"] == {"provider": "openai", "name": "gpt-4"} assert result["model_id"] == "model-1" assert result["provider"] == "provider-1" def test_conversation_in_debug_mode(self) -> None: """Test in_debug_mode property.""" # Arrange conversation = Conversation( app_id=str(uuid4()), mode=AppMode.CHAT, name="Test Conversation", status=ConversationStatus.NORMAL, from_source=ConversationFromSource.API, from_end_user_id=str(uuid4()), override_model_configs='{"model": "gpt-4"}', ) # Act result = conversation.in_debug_mode # Assert assert result is True @pytest.mark.parametrize("sqlite_session", [(Conversation,)], indirect=True) def test_conversation_to_dict_serialization(self, sqlite_session: Session) -> None: """Test conversation to_dict method.""" # Arrange app_id = str(uuid4()) from_end_user_id = str(uuid4()) conversation = Conversation( app_id=app_id, mode=AppMode.CHAT, name="Test Conversation", status=ConversationStatus.NORMAL, from_source=ConversationFromSource.API, from_end_user_id=from_end_user_id, dialogue_count=5, ) conversation.id = str(uuid4()) conversation._inputs = {"query": "test"} # Act result = conversation.to_dict(session=sqlite_session) # Assert assert result["id"] == conversation.id assert result["app_id"] == app_id assert result["mode"] == AppMode.CHAT assert result["name"] == "Test Conversation" assert result["status"] == "normal" assert result["from_source"] == "api" assert result["from_end_user_id"] == from_end_user_id assert result["dialogue_count"] == 5 assert result["inputs"] == {"query": "test"} class TestMessageModel: """Test suite for Message model and App-Message relationships.""" def test_message_creation_with_required_fields(self) -> None: """Test creating a message with required fields.""" # Arrange app_id = str(uuid4()) conversation_id = str(uuid4()) # Act message = Message( app_id=app_id, conversation_id=conversation_id, query="What is AI?", message={"role": "user", "content": "What is AI?"}, answer="AI stands for Artificial Intelligence.", message_unit_price=Decimal("0.0001"), answer_unit_price=Decimal("0.0002"), currency="USD", from_source=ConversationFromSource.API, ) # Assert assert message.app_id == app_id assert message.conversation_id == conversation_id assert message.query == "What is AI?" assert message.answer == "AI stands for Artificial Intelligence." assert message.currency == "USD" assert message.from_source == "api" def test_message_with_inputs(self, sqlite_session: Session) -> None: """Test message inputs round-trip through the session-aware accessor.""" # Arrange inputs = {"query": "Hello", "context": "test"} message = Message( app_id=str(uuid4()), conversation_id=str(uuid4()), query="Test query", message={"role": "user", "content": "Test"}, answer="Test answer", message_unit_price=Decimal("0.0001"), answer_unit_price=Decimal("0.0002"), currency="USD", from_source=ConversationFromSource.API, _inputs=inputs, ) # Act result = message.inputs_with_session(session=sqlite_session) # Assert assert result == inputs def test_message_inputs_is_write_only(self) -> None: """Reading message.inputs must fail loudly; reads go through inputs_with_session.""" message = Message(app_id=str(uuid4()), _inputs={}) with pytest.raises(AttributeError, match="no getter"): _ = message.inputs def test_message_inputs_setter(self) -> None: """Test message inputs setter.""" # Arrange message = Message( app_id=str(uuid4()), conversation_id=str(uuid4()), query="Test query", message={"role": "user", "content": "Test"}, answer="Test answer", message_unit_price=Decimal("0.0001"), answer_unit_price=Decimal("0.0002"), currency="USD", from_source=ConversationFromSource.API, ) inputs = {"query": "Hello", "context": "test"} # Act message.inputs = inputs # Assert assert message._inputs == inputs def test_message_in_debug_mode(self) -> None: """Test message in_debug_mode property.""" # Arrange message = Message( app_id=str(uuid4()), conversation_id=str(uuid4()), query="Test query", message={"role": "user", "content": "Test"}, answer="Test answer", message_unit_price=Decimal("0.0001"), answer_unit_price=Decimal("0.0002"), currency="USD", from_source=ConversationFromSource.API, override_model_configs='{"model": "gpt-4"}', ) # Act result = message.in_debug_mode # Assert assert result is True def test_message_metadata_dict_property(self) -> None: """Test message_metadata_dict property.""" # Arrange metadata = {"retriever_resources": ["doc1", "doc2"], "usage": {"tokens": 100}} message = Message( app_id=str(uuid4()), conversation_id=str(uuid4()), query="Test query", message={"role": "user", "content": "Test"}, answer="Test answer", message_unit_price=Decimal("0.0001"), answer_unit_price=Decimal("0.0002"), currency="USD", from_source=ConversationFromSource.API, message_metadata=json.dumps(metadata), ) # Act result = message.message_metadata_dict # Assert assert result == metadata def test_message_metadata_dict_empty(self) -> None: """Test message_metadata_dict when metadata is None.""" # Arrange message = Message( app_id=str(uuid4()), conversation_id=str(uuid4()), query="Test query", message={"role": "user", "content": "Test"}, answer="Test answer", message_unit_price=Decimal("0.0001"), answer_unit_price=Decimal("0.0002"), currency="USD", from_source=ConversationFromSource.API, message_metadata=None, ) # Act result = message.message_metadata_dict # Assert assert result == {} def test_retriever_resources_missing_key_returns_empty_list(self) -> None: """message_metadata present without retriever_resources must yield []. Regression for #41688: web GET /messages failed Pydantic list validation when metadata existed but the key was absent (property returned None). """ message = Message( app_id=str(uuid4()), conversation_id=str(uuid4()), query="Test query", message={"role": "user", "content": "Test"}, answer="Test answer", message_unit_price=Decimal("0.0001"), answer_unit_price=Decimal("0.0002"), currency="USD", from_source=ConversationFromSource.API, message_metadata=json.dumps({"usage": {"tokens": 1}}), ) assert message.retriever_resources == [] def test_retriever_resources_none_metadata_returns_empty_list(self) -> None: """No message_metadata should still expose retriever_resources as [].""" message = Message( app_id=str(uuid4()), conversation_id=str(uuid4()), query="Test query", message={"role": "user", "content": "Test"}, answer="Test answer", message_unit_price=Decimal("0.0001"), answer_unit_price=Decimal("0.0002"), currency="USD", from_source=ConversationFromSource.API, message_metadata=None, ) assert message.retriever_resources == [] def test_retriever_resources_present_passthrough(self) -> None: """Existing retriever_resources list should be returned unchanged.""" resources = [{"dataset_id": "ds-1", "document_id": "doc-1"}] message = Message( app_id=str(uuid4()), conversation_id=str(uuid4()), query="Test query", message={"role": "user", "content": "Test"}, answer="Test answer", message_unit_price=Decimal("0.0001"), answer_unit_price=Decimal("0.0002"), currency="USD", from_source=ConversationFromSource.API, message_metadata=json.dumps({"retriever_resources": resources}), ) assert message.retriever_resources == resources def test_retriever_resources_explicit_null_returns_empty_list(self) -> None: """Explicit retriever_resources: null must yield [], not None. WebMessageListItem.retriever_resources is list[RetrieverResource]. dict.get(..., []) does not replace an explicit JSON null. """ message = Message( app_id=str(uuid4()), conversation_id=str(uuid4()), query="Test query", message={"role": "user", "content": "Test"}, answer="Test answer", message_unit_price=Decimal("0.0001"), answer_unit_price=Decimal("0.0002"), currency="USD", from_source=ConversationFromSource.API, message_metadata=json.dumps({"retriever_resources": None}), ) assert message.retriever_resources == [] @pytest.mark.parametrize("sqlite_session", [(Message,)], indirect=True) def test_message_to_dict_serialization(self, sqlite_session: Session) -> None: """Test message to_dict method.""" # Arrange app_id = str(uuid4()) conversation_id = str(uuid4()) now = datetime.now(UTC) message = Message( app_id=app_id, conversation_id=conversation_id, query="Test query", message={"role": "user", "content": "Test"}, answer="Test answer", message_unit_price=Decimal("0.0001"), answer_unit_price=Decimal("0.0002"), total_price=Decimal("0.0003"), currency="USD", from_source=ConversationFromSource.API, status=MessageStatus.NORMAL, id=str(uuid4()), _inputs={"query": "test"}, ) message.created_at = now message.updated_at = now # Act result = message.to_dict(session=sqlite_session) # Assert assert result["id"] == message.id assert result["app_id"] == app_id assert result["conversation_id"] == conversation_id assert result["query"] == "Test query" assert result["answer"] == "Test answer" assert result["status"] == "normal" assert result["from_source"] == "api" assert result["inputs"] == {"query": "test"} assert "created_at" in result assert "updated_at" in result def test_message_from_dict_deserialization(self) -> None: """Test message from_dict method.""" # Arrange message_id = str(uuid4()) app_id = str(uuid4()) conversation_id = str(uuid4()) data = { "id": message_id, "app_id": app_id, "conversation_id": conversation_id, "model_id": "gpt-4", "inputs": {"query": "test"}, "query": "Test query", "message": {"role": "user", "content": "Test"}, "answer": "Test answer", "total_price": Decimal("0.0003"), "status": "normal", "error": None, "message_metadata": {"usage": {"tokens": 100}}, "from_source": "api", "from_end_user_id": None, "from_account_id": None, "created_at": "2024-01-01T00:00:00", "updated_at": "2024-01-01T00:00:00", "agent_based": False, "workflow_run_id": None, } # Act message = Message.from_dict(data) # Assert assert message.id == message_id assert message.app_id == app_id assert message.conversation_id == conversation_id assert message.query == "Test query" assert message.answer == "Test answer" class TestMessageAnnotation: """Test suite for MessageAnnotation and annotation relationships.""" def test_message_annotation_creation(self) -> None: """Test creating a message annotation.""" # Arrange app_id = str(uuid4()) conversation_id = str(uuid4()) message_id = str(uuid4()) account_id = str(uuid4()) # Act annotation = MessageAnnotation( app_id=app_id, conversation_id=conversation_id, message_id=message_id, question="What is AI?", content="AI stands for Artificial Intelligence.", account_id=account_id, ) # Assert assert annotation.app_id == app_id assert annotation.conversation_id == conversation_id assert annotation.message_id == message_id assert annotation.question == "What is AI?" assert annotation.content == "AI stands for Artificial Intelligence." assert annotation.account_id == account_id def test_message_annotation_without_message_id(self) -> None: """Test creating annotation without message_id.""" # Arrange app_id = str(uuid4()) account_id = str(uuid4()) # Act annotation = MessageAnnotation( app_id=app_id, question="What is AI?", conversation_id=None, message_id=None, content="AI stands for Artificial Intelligence.", account_id=account_id, ) # Assert assert annotation.app_id == app_id assert annotation.message_id is None assert annotation.conversation_id is None assert annotation.question == "What is AI?" assert annotation.content == "AI stands for Artificial Intelligence." def test_message_annotation_hit_count_default(self) -> None: """Test annotation hit_count default value.""" # Arrange annotation = MessageAnnotation( app_id=str(uuid4()), question="Test question", conversation_id=None, message_id=None, content="Test content", account_id=str(uuid4()), ) # Act & Assert - default value is set by database # Model instantiation doesn't set server defaults assert hasattr(annotation, "hit_count") class TestAppAnnotationSetting: """Test suite for AppAnnotationSetting model.""" def test_app_annotation_setting_creation(self) -> None: """Test creating an app annotation setting.""" # Arrange app_id = str(uuid4()) collection_binding_id = str(uuid4()) created_user_id = str(uuid4()) updated_user_id = str(uuid4()) # Act setting = AppAnnotationSetting( app_id=app_id, score_threshold=0.8, collection_binding_id=collection_binding_id, created_user_id=created_user_id, updated_user_id=updated_user_id, ) # Assert assert setting.app_id == app_id assert setting.score_threshold == 0.8 assert setting.collection_binding_id == collection_binding_id assert setting.created_user_id == created_user_id assert setting.updated_user_id == updated_user_id def test_app_annotation_setting_score_threshold_validation(self) -> None: """Test score threshold values.""" # Arrange & Act setting_high = AppAnnotationSetting( app_id=str(uuid4()), score_threshold=0.95, collection_binding_id=str(uuid4()), created_user_id=str(uuid4()), updated_user_id=str(uuid4()), ) setting_low = AppAnnotationSetting( app_id=str(uuid4()), score_threshold=0.5, collection_binding_id=str(uuid4()), created_user_id=str(uuid4()), updated_user_id=str(uuid4()), ) # Assert assert setting_high.score_threshold == 0.95 assert setting_low.score_threshold == 0.5 class TestAppAnnotationHitHistory: """Test suite for AppAnnotationHitHistory model.""" def test_app_annotation_hit_history_creation(self) -> None: """Test creating an annotation hit history.""" # Arrange app_id = str(uuid4()) annotation_id = str(uuid4()) message_id = str(uuid4()) account_id = str(uuid4()) # Act history = AppAnnotationHitHistory( app_id=app_id, annotation_id=annotation_id, source="api", question="What is AI?", account_id=account_id, score=0.95, message_id=message_id, annotation_question="What is AI?", annotation_content="AI stands for Artificial Intelligence.", ) # Assert assert history.app_id == app_id assert history.annotation_id == annotation_id assert history.source == "api" assert history.question == "What is AI?" assert history.account_id == account_id assert history.score == 0.95 assert history.message_id == message_id assert history.annotation_question == "What is AI?" assert history.annotation_content == "AI stands for Artificial Intelligence." def test_app_annotation_hit_history_score_values(self) -> None: """Test annotation hit history with different score values.""" # Arrange & Act history_high = AppAnnotationHitHistory( app_id=str(uuid4()), annotation_id=str(uuid4()), source="api", question="Test", account_id=str(uuid4()), score=0.99, message_id=str(uuid4()), annotation_question="Test", annotation_content="Content", ) history_low = AppAnnotationHitHistory( app_id=str(uuid4()), annotation_id=str(uuid4()), source="api", question="Test", account_id=str(uuid4()), score=0.6, message_id=str(uuid4()), annotation_question="Test", annotation_content="Content", ) # Assert assert history_high.score == 0.99 assert history_low.score == 0.6 class TestSiteModel: """Test suite for Site model.""" def test_site_core_bulk_insert_generates_ids(self, sqlite_session: Session) -> None: """Bulk inserts generate an ID for every site row.""" app_ids = [str(uuid4()), str(uuid4())] sqlite_session.add_all( [ Site( app_id=app_id, title=f"Site {index}", default_language="en-US", customize_token_strategy=CustomizeTokenStrategy.UUID, ) for index, app_id in enumerate(app_ids) ] ) sqlite_session.commit() site_ids = sqlite_session.scalars(select(Site.id).where(Site.app_id.in_(app_ids))).all() assert len(site_ids) == len(app_ids) assert len(set(site_ids)) == len(app_ids) assert all(UUID(site_id).version == 4 for site_id in site_ids) def test_site_creation_with_required_fields(self) -> None: """Test creating a site with required fields.""" # Arrange app_id = str(uuid4()) # Act site = Site( app_id=app_id, title="Test Site", default_language="en-US", customize_token_strategy=CustomizeTokenStrategy.UUID, ) # Assert assert site.app_id == app_id assert site.title == "Test Site" assert site.default_language == "en-US" assert site.customize_token_strategy == CustomizeTokenStrategy.UUID def test_site_creation_with_optional_fields(self) -> None: """Test creating a site with optional fields.""" # Arrange & Act site = Site( app_id=str(uuid4()), title="Test Site", default_language="en-US", customize_token_strategy=CustomizeTokenStrategy.UUID, icon_type=IconType.EMOJI, icon="🌐", icon_background="#0066CC", description="Test site description", copyright="© 2024 Test", privacy_policy="https://example.com/privacy", ) # Assert assert site.icon_type == IconType.EMOJI assert site.icon == "🌐" assert site.icon_background == "#0066CC" assert site.description == "Test site description" assert site.copyright == "© 2024 Test" assert site.privacy_policy == "https://example.com/privacy" def test_site_custom_disclaimer_setter(self) -> None: """Test site custom_disclaimer setter.""" # Arrange site = Site( app_id=str(uuid4()), title="Test Site", default_language="en-US", customize_token_strategy=CustomizeTokenStrategy.UUID, ) # Act site.custom_disclaimer = "This is a test disclaimer" # Assert assert site.custom_disclaimer == "This is a test disclaimer" def test_site_custom_disclaimer_exceeds_limit(self) -> None: """Test site custom_disclaimer with excessive length.""" # Arrange site = Site( app_id=str(uuid4()), title="Test Site", default_language="en-US", customize_token_strategy=CustomizeTokenStrategy.UUID, ) long_disclaimer = "x" * 513 # Exceeds 512 character limit # Act & Assert with pytest.raises(ValueError, match="Custom disclaimer cannot exceed 512 characters"): site.custom_disclaimer = long_disclaimer class TestModelIntegration: """Test suite for model integration scenarios.""" def test_complete_app_conversation_message_hierarchy(self) -> None: """Test complete hierarchy from app to message.""" # Arrange tenant_id = str(uuid4()) app_id = str(uuid4()) conversation_id = str(uuid4()) message_id = str(uuid4()) created_by = str(uuid4()) # Create app app = App( tenant_id=tenant_id, name="Test App", mode=AppMode.CHAT, enable_site=True, enable_api=True, created_by=created_by, id=app_id, ) # Create conversation conversation = Conversation( app_id=app_id, mode=AppMode.CHAT, name="Test Conversation", status=ConversationStatus.NORMAL, from_source=ConversationFromSource.API, from_end_user_id=str(uuid4()), ) conversation.id = conversation_id # Create message message = Message( app_id=app_id, conversation_id=conversation_id, query="Test query", message={"role": "user", "content": "Test"}, answer="Test answer", message_unit_price=Decimal("0.0001"), answer_unit_price=Decimal("0.0002"), currency="USD", from_source=ConversationFromSource.API, id=message_id, ) # Assert assert app.id == app_id assert conversation.app_id == app_id assert message.app_id == app_id assert message.conversation_id == conversation_id assert app.mode == AppMode.CHAT assert conversation.mode == AppMode.CHAT def test_app_with_annotation_setting(self) -> None: """Test app with annotation setting.""" # Arrange app_id = str(uuid4()) collection_binding_id = str(uuid4()) created_user_id = str(uuid4()) # Create app app = App( tenant_id=str(uuid4()), name="Test App", mode=AppMode.CHAT, enable_site=True, enable_api=True, created_by=created_user_id, id=app_id, ) # Create annotation setting setting = AppAnnotationSetting( app_id=app_id, score_threshold=0.85, collection_binding_id=collection_binding_id, created_user_id=created_user_id, updated_user_id=created_user_id, ) # Assert assert setting.app_id == app.id assert setting.score_threshold == 0.85 def test_message_with_annotation(self) -> None: """Test message with annotation.""" # Arrange app_id = str(uuid4()) conversation_id = str(uuid4()) message_id = str(uuid4()) account_id = str(uuid4()) # Create message message = Message( app_id=app_id, conversation_id=conversation_id, query="What is AI?", message={"role": "user", "content": "What is AI?"}, answer="AI stands for Artificial Intelligence.", message_unit_price=Decimal("0.0001"), answer_unit_price=Decimal("0.0002"), currency="USD", from_source=ConversationFromSource.API, id=message_id, ) # Create annotation annotation = MessageAnnotation( app_id=app_id, conversation_id=conversation_id, message_id=message_id, question="What is AI?", content="AI stands for Artificial Intelligence.", account_id=account_id, ) # Assert assert annotation.app_id == message.app_id assert annotation.conversation_id == message.conversation_id assert annotation.message_id == message.id def test_annotation_hit_history_tracking(self) -> None: """Test annotation hit history tracking.""" # Arrange app_id = str(uuid4()) annotation_id = str(uuid4()) message_id = str(uuid4()) account_id = str(uuid4()) # Create annotation annotation = MessageAnnotation( app_id=app_id, question="What is AI?", content="AI stands for Artificial Intelligence.", conversation_id=None, message_id=message_id, account_id=account_id, ) annotation.id = annotation_id # Create hit history history = AppAnnotationHitHistory( app_id=app_id, annotation_id=annotation_id, source="api", question="What is AI?", account_id=account_id, score=0.92, message_id=message_id, annotation_question="What is AI?", annotation_content="AI stands for Artificial Intelligence.", ) # Assert assert history.app_id == annotation.app_id assert history.annotation_id == annotation.id assert history.score == 0.92 def test_app_with_site(self) -> None: """Test app with site.""" # Arrange app_id = str(uuid4()) # Create app app = App( tenant_id=str(uuid4()), name="Test App", mode=AppMode.CHAT, enable_site=True, enable_api=True, created_by=str(uuid4()), id=app_id, ) # Create site site = Site( app_id=app_id, title="Test Site", default_language="en-US", customize_token_strategy=CustomizeTokenStrategy.UUID, ) # Assert assert site.app_id == app.id assert app.enable_site is True """Regression coverage for the ``@property``→session-parameter refactor on ``App.tenant``. The legacy ``@property`` reached for the global ``db.session`` internally and has been converted to a plain method taking an explicit ``session`` (per the pattern established in #41830/#41831 for ``InstalledApp.tenant`` / ``ApiToolProvider.tenant``, tracked in #40372). """ def _app_session_accessors_persist_tenant(session: Session) -> Tenant: tenant = Tenant(name="Test Tenant") session.add(tenant) session.flush() return tenant def _app_session_accessors_persist_app(session: Session, *, tenant_id: str) -> App: app = App( tenant_id=tenant_id, name="Test App", mode=AppMode.CHAT, enable_site=True, enable_api=False, created_by=str(uuid4()), ) session.add(app) session.flush() return app class TestAppSessionAccessorsCases: class TestAppTenant: def test_returns_the_matching_tenant(self, sqlite_session: Session) -> None: tenant = _app_session_accessors_persist_tenant(sqlite_session) app = _app_session_accessors_persist_app(sqlite_session, tenant_id=tenant.id) result = app.tenant(session=sqlite_session) assert result is not None assert result.id == tenant.id def test_returns_none_when_tenant_missing(self, sqlite_session: Session) -> None: app = _app_session_accessors_persist_app(sqlite_session, tenant_id=str(uuid4())) assert app.tenant(session=sqlite_session) is None """Regression coverage for the ``@property``→session-parameter refactor on ``InstalledApp.tenant``. The legacy ``@property`` reached for the global ``db.session`` internally and has been converted to a plain method taking an explicit ``session: Session`` (per the pattern established in #40370/#40797/#41394/#41830/#41885, tracked in #40372). """ def _installed_app_session_accessors_persist_tenant(session: Session) -> Tenant: tenant = Tenant(name="Test Tenant") session.add(tenant) session.flush() return tenant def _persist_installed_app(session: Session, *, tenant_id: str) -> InstalledApp: installed_app = InstalledApp( tenant_id=tenant_id, app_id=str(uuid4()), app_owner_tenant_id=str(uuid4()), position=0, is_pinned=False, last_used_at=None, ) session.add(installed_app) session.flush() return installed_app class TestInstalledAppSessionAccessorsCases: class TestInstalledAppTenant: def test_returns_the_matching_tenant(self, sqlite_session: Session) -> None: tenant = _installed_app_session_accessors_persist_tenant(sqlite_session) installed_app = _persist_installed_app(sqlite_session, tenant_id=tenant.id) result = installed_app.tenant(session=sqlite_session) assert result is not None assert result.id == tenant.id def test_returns_none_when_tenant_missing(self, sqlite_session: Session) -> None: installed_app = _persist_installed_app(sqlite_session, tenant_id=str(uuid4())) assert installed_app.tenant(session=sqlite_session) is None """Regression coverage for the ``@property``→session-parameter refactor on Message feedback/annotation accessors. Covers three of `api/models/model.py`'s legacy `@property` accessors that reached for the global ``db.session`` internally and have been removed in favour of their existing explicit-session methods (per the pattern established in #40370/#40797/#41394/#41830, tracked in #40372): - ``MessageFeedback.from_account_with_session`` - ``MessageAnnotation.account_with_session`` - ``MessageAnnotation.annotation_create_account_with_session`` Each accessor is exercised against the real ``sqlite_session`` fixture (a genuine SQLAlchemy ``Session`` bound to a pristine full-schema SQLite database) so the assertions cover actual query behaviour rather than a mock's recorded call. """ def _message_annotation_session_accessors_persist_account(session: Session, *, email: str | None = None) -> Account: account = Account(name="Test Account", email=email or "test@example.com") session.add(account) session.flush() return account def _persist_feedback(session: Session, *, app_id: str, from_account_id: str | None) -> MessageFeedback: feedback = MessageFeedback( app_id=app_id, conversation_id=str(uuid4()), message_id=str(uuid4()), rating=FeedbackRating.LIKE, from_source=FeedbackFromSource.USER, from_account_id=from_account_id, ) session.add(feedback) session.flush() return feedback def _message_annotation_session_accessors_persist_annotation( session: Session, *, app_id: str, account_id: str ) -> MessageAnnotation: annotation = MessageAnnotation( app_id=app_id, conversation_id=None, message_id=None, question="What is AI?", content="AI stands for Artificial Intelligence.", account_id=account_id, ) session.add(annotation) session.flush() return annotation class TestMessageAnnotationSessionAccessorsCases: class TestMessageFeedbackFromAccount: def test_returns_the_feedback_author(self, sqlite_session: Session) -> None: account = _message_annotation_session_accessors_persist_account(sqlite_session) feedback = _persist_feedback(sqlite_session, app_id=str(uuid4()), from_account_id=account.id) result = feedback.from_account_with_session(session=sqlite_session) assert result is not None assert result.id == account.id def test_returns_none_when_account_missing(self, sqlite_session: Session) -> None: feedback = _persist_feedback(sqlite_session, app_id=str(uuid4()), from_account_id=str(uuid4())) assert feedback.from_account_with_session(session=sqlite_session) is None class TestMessageAnnotationAccount: def test_returns_the_annotation_creator(self, sqlite_session: Session) -> None: creator = _message_annotation_session_accessors_persist_account(sqlite_session, email="creator@example.com") annotation = _message_annotation_session_accessors_persist_annotation( sqlite_session, app_id=str(uuid4()), account_id=creator.id ) result = annotation.account_with_session(session=sqlite_session) assert result is not None assert result.id == creator.id def test_returns_none_when_account_missing(self, sqlite_session: Session) -> None: annotation = _message_annotation_session_accessors_persist_annotation( sqlite_session, app_id=str(uuid4()), account_id=str(uuid4()) ) assert annotation.account_with_session(session=sqlite_session) is None class TestMessageAnnotationAnnotationCreateAccount: def test_returns_the_annotation_creator(self, sqlite_session: Session) -> None: creator = _message_annotation_session_accessors_persist_account( sqlite_session, email="creator2@example.com" ) annotation = _message_annotation_session_accessors_persist_annotation( sqlite_session, app_id=str(uuid4()), account_id=creator.id ) result = annotation.annotation_create_account_with_session(session=sqlite_session) assert result is not None assert result.id == creator.id def test_returns_none_when_account_missing(self, sqlite_session: Session) -> None: annotation = _message_annotation_session_accessors_persist_annotation( sqlite_session, app_id=str(uuid4()), account_id=str(uuid4()) ) assert annotation.annotation_create_account_with_session(session=sqlite_session) is None def _wrap_md(url: str) -> str: """ Wrap a raw URL into the markdown that re_sign_file_url_answer expects: [link]() """ return f"please click [file]({url}) to download." def _build_local_file_mapping(record_id: str, *, tenant_id: str | None = None) -> dict[str, object]: mapping: dict[str, object] = { "dify_model_identity": FILE_MODEL_IDENTITY, "transfer_method": FileTransferMethod.LOCAL_FILE, "reference": build_file_reference(record_id=record_id), "type": "document", "filename": "example.txt", "extension": ".txt", "mime_type": "text/plain", "size": 1, } if tenant_id is not None: mapping["tenant_id"] = tenant_id return mapping class TestModelCases: @staticmethod @pytest.fixture(autouse=True) def patch_file_helpers(monkeypatch: pytest.MonkeyPatch) -> None: """ Patch file_helpers.get_signed_file_url to a deterministic stub. """ monkeypatch.setattr( "graphon.file.helpers.get_signed_file_url", lambda file_id: f"https://signed.example/{file_id}", ) @staticmethod def test_file_preview_valid_replaced() -> None: """ Valid file-preview URL must be re-signed: - Extract upload_file_id correctly - Replace the original URL with the signed URL """ upload_id = "abc-123" url = f"/files/{upload_id}/file-preview?timestamp=111&nonce=222&sign=333" msg = Message(answer=_wrap_md(url)) out = msg.re_sign_file_url_answer assert f"https://signed.example/{upload_id}" in out assert url not in out @staticmethod def test_file_preview_misspelled_not_replaced() -> None: """ Misspelled endpoint 'file-previe?timestamp=' should NOT be rewritten. """ upload_id = "zzz-001" # path deliberately misspelled: file-previe? (missing 'w') # and we append ¬e=file-preview to trick the old `"file-preview" in url` check. url = f"/files/{upload_id}/file-previe?timestamp=111&nonce=222&sign=333¬e=file-preview" original = _wrap_md(url) msg = Message(answer=original) out = msg.re_sign_file_url_answer # Expect NO replacement, should not rewrite misspelled file-previe URL assert out == original @staticmethod def test_image_preview_valid_replaced() -> None: """ Valid image-preview URL must be re-signed. """ upload_id = "img-789" url = f"/files/{upload_id}/image-preview?timestamp=123&nonce=456&sign=789" msg = Message(answer=_wrap_md(url)) out = msg.re_sign_file_url_answer assert f"https://signed.example/{upload_id}" in out assert url not in out @staticmethod def test_image_preview_misspelled_not_replaced() -> None: """ Misspelled endpoint 'image-previe?timestamp=' should NOT be rewritten. """ upload_id = "img-err-42" url = f"/files/{upload_id}/image-previe?timestamp=1&nonce=2&sign=3¬e=image-preview" original = _wrap_md(url) msg = Message(answer=original) out = msg.re_sign_file_url_answer # Expect NO replacement, should not rewrite misspelled image-previe URL assert out == original @staticmethod @pytest.mark.parametrize("owner_cls", [Conversation, Message]) def test_inputs_restore_external_remote_url_file_mappings( owner_cls: type[Conversation] | type[Message], sqlite_session: Session ) -> None: owner = owner_cls(app_id="app-1") owner.inputs = { "file": { "dify_model_identity": FILE_MODEL_IDENTITY, "transfer_method": FileTransferMethod.REMOTE_URL, "type": "document", "url": "https://example.com/report.pdf", "filename": "report.pdf", "extension": ".pdf", "mime_type": "application/pdf", "size": 1, } } restored_file = owner.inputs_with_session(session=sqlite_session)["file"] assert restored_file.transfer_method == FileTransferMethod.REMOTE_URL assert restored_file.remote_url == "https://example.com/report.pdf" @staticmethod @pytest.mark.parametrize("sqlite_session", [(App,)], indirect=True) def test_message_inputs_resolve_file_tenant_with_caller_session(sqlite_session: Session) -> None: app = App( id="app-1", tenant_id="tenant-1", name="File owner", description="", mode=AppMode.CHAT, icon_type=IconType.EMOJI, icon="file", icon_background="#FFFFFF", enable_site=False, enable_api=False, max_active_requests=0, ) decoy = App( id="other-app", tenant_id="other-tenant", name="Decoy", description="", mode=AppMode.CHAT, icon_type=IconType.EMOJI, icon="file", icon_background="#FFFFFF", enable_site=False, enable_api=False, max_active_requests=0, ) sqlite_session.add_all([decoy, app]) sqlite_session.flush() message = Message(app_id="app-1") message.inputs = {"file": _build_local_file_mapping("upload-1")} with patch( "models.model.build_file_from_input_mapping", side_effect=lambda **kwargs: kwargs["tenant_resolver"](), ): inputs = message.inputs_with_session(session=sqlite_session) assert inputs["file"] == "tenant-1" # session.scalar.assert_called_once() # removed: see #40799 conftest note @staticmethod def test_file_url_bare_url_re_signed() -> None: """#40788: bare tool file URL (no markdown wrapping) must be re-signed. Bare URLs are produced by the agent runtime when the model returns a non-linkified file reference. The previous regex only matched [text](url) markdown form, so the bare URL kept INTERNAL_FILES_URL. """ upload_id = "bare-1" url = f"/files/{upload_id}/file-preview?timestamp=1&nonce=2&sign=3" msg = Message(answer=url) # bare, no markdown wrapping out = msg.re_sign_file_url_answer assert f"https://signed.example/{upload_id}" in out assert url not in out @staticmethod def test_file_url_backticked_url_re_signed() -> None: """#40788: backticked tool file URL must be re-signed. Backticks are used by the runtime to quote file references inline. The previous regex only matched [text](url), so backticked URLs stayed on INTERNAL_FILES_URL. """ upload_id = "tick-2" url = f"/files/{upload_id}/file-preview?timestamp=10&nonce=20&sign=30" msg = Message(answer=f"see `{url}` for details") out = msg.re_sign_file_url_answer assert f"https://signed.example/{upload_id}" in out assert url not in out @staticmethod def test_file_url_mixed_formats_all_re_signed() -> None: """#40788: all three URL shapes in the same answer are re-signed. Sanity check that the three patterns cooperate — bare, backticked, and markdown — without double-signing or skipping any. """ upload_ids = ["mix-a", "mix-b", "mix-c"] bare = f"/files/{upload_ids[0]}/file-preview?timestamp=1&nonce=2&sign=3" tick = f"/files/{upload_ids[1]}/file-preview?timestamp=10&nonce=20&sign=30" md = f"/files/{upload_ids[2]}/file-preview?timestamp=100&nonce=200&sign=300" msg = Message(answer=f"raw {bare} ` {tick} ` and [file]({md})") out = msg.re_sign_file_url_answer for uid in upload_ids: assert f"https://signed.example/{uid}" in out for url in (bare, tick, md): assert url not in out """Regression coverage for the ``@property``→session-parameter refactor on ``RecommendedApp`` and ``AccountTrialAppRecord``. The legacy ``@property`` accessors reached for the global ``db.session`` internally and have been converted to plain methods taking an explicit ``session: Session`` (per the pattern established in #40370/#41394/#41830/#42000, tracked in #40372). """ def _recommended_app_session_accessors_persist_app(session: Session) -> App: app = App( tenant_id=str(uuid4()), name="Test App", mode=AppMode.CHAT, enable_site=True, enable_api=False, created_by=str(uuid4()), description="App description", ) session.add(app) session.flush() return app def _persist_recommended_app(session: Session, *, app_id: str) -> RecommendedApp: recommended_app = RecommendedApp( app_id=app_id, description={"en_US": "A recommended app"}, copyright="Dify", privacy_policy="https://dify.ai/privacy", category="Assistant", ) session.add(recommended_app) session.flush() return recommended_app def _recommended_app_session_accessors_persist_account(session: Session) -> Account: account = Account(name="Test User", email=f"user-{uuid4()}@example.com") session.add(account) session.flush() return account def _persist_trial_record(session: Session, *, account_id: str, app_id: str) -> AccountTrialAppRecord: record = AccountTrialAppRecord(account_id=account_id, app_id=app_id, count=1) session.add(record) session.flush() return record class TestRecommendedAppSessionAccessorsCases: class TestRecommendedAppApp: def test_returns_the_matching_app(self, sqlite_session: Session) -> None: app = _recommended_app_session_accessors_persist_app(sqlite_session) recommended_app = _persist_recommended_app(sqlite_session, app_id=app.id) result = recommended_app.app(session=sqlite_session) assert result is not None assert result.id == app.id def test_returns_none_when_app_missing(self, sqlite_session: Session) -> None: recommended_app = _persist_recommended_app(sqlite_session, app_id=str(uuid4())) assert recommended_app.app(session=sqlite_session) is None class TestAccountTrialAppRecordAccessors: def test_app_returns_the_matching_app(self, sqlite_session: Session) -> None: app = _recommended_app_session_accessors_persist_app(sqlite_session) record = _persist_trial_record(sqlite_session, account_id=str(uuid4()), app_id=app.id) result = record.app(session=sqlite_session) assert result is not None assert result.id == app.id def test_app_returns_none_when_app_missing(self, sqlite_session: Session) -> None: record = _persist_trial_record(sqlite_session, account_id=str(uuid4()), app_id=str(uuid4())) assert record.app(session=sqlite_session) is None def test_user_returns_the_matching_account(self, sqlite_session: Session) -> None: account = _recommended_app_session_accessors_persist_account(sqlite_session) record = _persist_trial_record(sqlite_session, account_id=account.id, app_id=str(uuid4())) result = record.user(session=sqlite_session) assert result is not None assert result.id == account.id def test_user_returns_none_when_account_missing(self, sqlite_session: Session) -> None: record = _persist_trial_record(sqlite_session, account_id=str(uuid4()), app_id=str(uuid4())) assert record.user(session=sqlite_session) is None TARGET_MODELS = (App, AppStar, Conversation, Message, EndUser, ApiToken) def test_app_conversation_models_use_typebase_registry() -> None: for model in TARGET_MODELS: mapper = inspect(model) assert mapper.registry is TypeBase.registry assert mapper.registry is not Base.registry def test_generated_identifiers_and_collection_defaults_are_independent() -> None: first = Conversation( app_id="app-1", mode=AppMode.CHAT, name="First", from_source=ConversationFromSource.API, ) second = Conversation( app_id="app-1", mode=AppMode.CHAT, name="Second", from_source=ConversationFromSource.API, ) assert first.id != second.id assert first._inputs == second._inputs == {} assert first._inputs is not second._inputs @pytest.mark.parametrize("sqlite_session", [TARGET_MODELS], indirect=True) def test_app_conversation_models_persist_with_relationships(sqlite_session: Session) -> None: app = App( id="app-1", tenant_id="tenant-1", name="Typed app", mode=AppMode.CHAT, enable_site=True, enable_api=True, ) star = AppStar(id="star-1", tenant_id=app.tenant_id, app_id=app.id, account_id="account-1") conversation = Conversation( id="conversation-1", app_id=app.id, mode=AppMode.CHAT, name="Typed conversation", from_source=ConversationFromSource.API, ) message = Message( id="message-1", app_id=app.id, conversation_id=conversation.id, query="hello", message={"role": "user"}, answer="world", from_source=ConversationFromSource.API, total_price=Decimal(0), ) end_user = EndUser( id="end-user-1", tenant_id=app.tenant_id, app_id=app.id, type=EndUserType.BROWSER, session_id="session-1", ) token = ApiToken( id="token-1", app_id=app.id, tenant_id=app.tenant_id, type=ApiTokenType.APP, token="secret", ) sqlite_session.add_all([app, star, conversation, message, end_user, token]) sqlite_session.commit() persisted_conversation = sqlite_session.get(Conversation, conversation.id) assert persisted_conversation is not None assert [item.id for item in persisted_conversation.messages] == [message.id] assert sqlite_session.get(AppStar, star.id) is not None assert sqlite_session.get(EndUser, end_user.id) is not None assert sqlite_session.get(ApiToken, token.id) is not None assert message.created_at is not None