from unittest.mock import patch import pytest from faker import Faker from core.rag.index_processor.constant.built_in_field import BuiltInField from models.account import Account, Tenant, TenantAccountJoin, TenantAccountRole from models.dataset import Dataset, DatasetMetadata, DatasetMetadataBinding, Document from services.entities.knowledge_entities.knowledge_entities import MetadataArgs from services.metadata_service import MetadataService class TestMetadataService: """Integration tests for MetadataService using testcontainers.""" @pytest.fixture def mock_external_service_dependencies(self): """Mock setup for external service dependencies.""" with ( patch("services.metadata_service.current_user") as mock_current_user, patch("services.metadata_service.redis_client") as mock_redis_client, patch("services.dataset_service.DocumentService") as mock_document_service, ): # Setup default mock returns mock_redis_client.get.return_value = None mock_redis_client.set.return_value = True mock_redis_client.delete.return_value = 1 yield { "current_user": mock_current_user, "redis_client": mock_redis_client, "document_service": mock_document_service, } def _create_test_account_and_tenant(self, db_session_with_containers, mock_external_service_dependencies): """ Helper method to create a test account and tenant for testing. Args: db_session_with_containers: Database session from testcontainers infrastructure mock_external_service_dependencies: Mock dependencies Returns: tuple: (account, tenant) - Created account and tenant instances """ fake = Faker() # Create account account = Account( email=fake.email(), name=fake.name(), interface_language="en-US", status="active", ) from extensions.ext_database import db db.session.add(account) db.session.commit() # Create tenant for the account tenant = Tenant( name=fake.company(), status="normal", ) db.session.add(tenant) db.session.commit() # Create tenant-account join join = TenantAccountJoin( tenant_id=tenant.id, account_id=account.id, role=TenantAccountRole.OWNER.value, current=True, ) db.session.add(join) db.session.commit() # Set current tenant for account account.current_tenant = tenant return account, tenant def _create_test_dataset(self, db_session_with_containers, mock_external_service_dependencies, account, tenant): """ Helper method to create a test dataset for testing. Args: db_session_with_containers: Database session from testcontainers infrastructure mock_external_service_dependencies: Mock dependencies account: Account instance tenant: Tenant instance Returns: Dataset: Created dataset instance """ fake = Faker() dataset = Dataset( tenant_id=tenant.id, name=fake.company(), description=fake.text(max_nb_chars=100), data_source_type="upload_file", created_by=account.id, built_in_field_enabled=False, ) from extensions.ext_database import db db.session.add(dataset) db.session.commit() return dataset def _create_test_document(self, db_session_with_containers, mock_external_service_dependencies, dataset, account): """ Helper method to create a test document for testing. Args: db_session_with_containers: Database session from testcontainers infrastructure mock_external_service_dependencies: Mock dependencies dataset: Dataset instance account: Account instance Returns: Document: Created document instance """ fake = Faker() document = Document( tenant_id=dataset.tenant_id, dataset_id=dataset.id, position=1, data_source_type="upload_file", data_source_info="{}", batch="test-batch", name=fake.file_name(), created_from="web", created_by=account.id, doc_form="text", doc_language="en", ) from extensions.ext_database import db db.session.add(document) db.session.commit() return document def test_create_metadata_success(self, db_session_with_containers, mock_external_service_dependencies): """ Test successful metadata creation with valid parameters. """ # Arrange: Create test data account, tenant = self._create_test_account_and_tenant( db_session_with_containers, mock_external_service_dependencies ) dataset = self._create_test_dataset( db_session_with_containers, mock_external_service_dependencies, account, tenant ) # Setup mocks mock_external_service_dependencies["current_user"].current_tenant_id = tenant.id mock_external_service_dependencies["current_user"].id = account.id metadata_args = MetadataArgs(type="string", name="test_metadata") # Act: Execute the method under test result = MetadataService.create_metadata(dataset.id, metadata_args) # Assert: Verify the expected outcomes assert result is not None assert result.name == "test_metadata" assert result.type == "string" assert result.dataset_id == dataset.id assert result.tenant_id == tenant.id assert result.created_by == account.id # Verify database state from extensions.ext_database import db db.session.refresh(result) assert result.id is not None assert result.created_at is not None def test_create_metadata_name_too_long(self, db_session_with_containers, mock_external_service_dependencies): """ Test metadata creation fails when name exceeds 255 characters. """ # Arrange: Create test data account, tenant = self._create_test_account_and_tenant( db_session_with_containers, mock_external_service_dependencies ) dataset = self._create_test_dataset( db_session_with_containers, mock_external_service_dependencies, account, tenant ) # Setup mocks mock_external_service_dependencies["current_user"].current_tenant_id = tenant.id mock_external_service_dependencies["current_user"].id = account.id long_name = "a" * 256 # 256 characters, exceeding 255 limit metadata_args = MetadataArgs(type="string", name=long_name) # Act & Assert: Verify proper error handling with pytest.raises(ValueError, match="Metadata name cannot exceed 255 characters."): MetadataService.create_metadata(dataset.id, metadata_args) def test_create_metadata_name_already_exists(self, db_session_with_containers, mock_external_service_dependencies): """ Test metadata creation fails when name already exists in the same dataset. """ # Arrange: Create test data account, tenant = self._create_test_account_and_tenant( db_session_with_containers, mock_external_service_dependencies ) dataset = self._create_test_dataset( db_session_with_containers, mock_external_service_dependencies, account, tenant ) # Setup mocks mock_external_service_dependencies["current_user"].current_tenant_id = tenant.id mock_external_service_dependencies["current_user"].id = account.id # Create first metadata first_metadata_args = MetadataArgs(type="string", name="duplicate_name") MetadataService.create_metadata(dataset.id, first_metadata_args) # Try to create second metadata with same name second_metadata_args = MetadataArgs(type="number", name="duplicate_name") # Act & Assert: Verify proper error handling with pytest.raises(ValueError, match="Metadata name already exists."): MetadataService.create_metadata(dataset.id, second_metadata_args) def test_create_metadata_name_conflicts_with_built_in_field( self, db_session_with_containers, mock_external_service_dependencies ): """ Test metadata creation fails when name conflicts with built-in field names. """ # Arrange: Create test data account, tenant = self._create_test_account_and_tenant( db_session_with_containers, mock_external_service_dependencies ) dataset = self._create_test_dataset( db_session_with_containers, mock_external_service_dependencies, account, tenant ) # Setup mocks mock_external_service_dependencies["current_user"].current_tenant_id = tenant.id mock_external_service_dependencies["current_user"].id = account.id # Try to create metadata with built-in field name built_in_field_name = BuiltInField.document_name.value metadata_args = MetadataArgs(type="string", name=built_in_field_name) # Act & Assert: Verify proper error handling with pytest.raises(ValueError, match="Metadata name already exists in Built-in fields."): MetadataService.create_metadata(dataset.id, metadata_args) def test_update_metadata_name_success(self, db_session_with_containers, mock_external_service_dependencies): """ Test successful metadata name update with valid parameters. """ # Arrange: Create test data account, tenant = self._create_test_account_and_tenant( db_session_with_containers, mock_external_service_dependencies ) dataset = self._create_test_dataset( db_session_with_containers, mock_external_service_dependencies, account, tenant ) # Setup mocks mock_external_service_dependencies["current_user"].current_tenant_id = tenant.id mock_external_service_dependencies["current_user"].id = account.id # Create metadata first metadata_args = MetadataArgs(type="string", name="old_name") metadata = MetadataService.create_metadata(dataset.id, metadata_args) # Act: Execute the method under test new_name = "new_name" result = MetadataService.update_metadata_name(dataset.id, metadata.id, new_name) # Assert: Verify the expected outcomes assert result is not None assert result.name == new_name assert result.updated_by == account.id assert result.updated_at is not None # Verify database state from extensions.ext_database import db db.session.refresh(result) assert result.name == new_name def test_update_metadata_name_too_long(self, db_session_with_containers, mock_external_service_dependencies): """ Test metadata name update fails when new name exceeds 255 characters. """ # Arrange: Create test data account, tenant = self._create_test_account_and_tenant( db_session_with_containers, mock_external_service_dependencies ) dataset = self._create_test_dataset( db_session_with_containers, mock_external_service_dependencies, account, tenant ) # Setup mocks mock_external_service_dependencies["current_user"].current_tenant_id = tenant.id mock_external_service_dependencies["current_user"].id = account.id # Create metadata first metadata_args = MetadataArgs(type="string", name="old_name") metadata = MetadataService.create_metadata(dataset.id, metadata_args) # Try to update with too long name long_name = "a" * 256 # 256 characters, exceeding 255 limit # Act & Assert: Verify proper error handling with pytest.raises(ValueError, match="Metadata name cannot exceed 255 characters."): MetadataService.update_metadata_name(dataset.id, metadata.id, long_name) def test_update_metadata_name_already_exists(self, db_session_with_containers, mock_external_service_dependencies): """ Test metadata name update fails when new name already exists in the same dataset. """ # Arrange: Create test data account, tenant = self._create_test_account_and_tenant( db_session_with_containers, mock_external_service_dependencies ) dataset = self._create_test_dataset( db_session_with_containers, mock_external_service_dependencies, account, tenant ) # Setup mocks mock_external_service_dependencies["current_user"].current_tenant_id = tenant.id mock_external_service_dependencies["current_user"].id = account.id # Create two metadata entries first_metadata_args = MetadataArgs(type="string", name="first_metadata") first_metadata = MetadataService.create_metadata(dataset.id, first_metadata_args) second_metadata_args = MetadataArgs(type="number", name="second_metadata") second_metadata = MetadataService.create_metadata(dataset.id, second_metadata_args) # Try to update first metadata with second metadata's name with pytest.raises(ValueError, match="Metadata name already exists."): MetadataService.update_metadata_name(dataset.id, first_metadata.id, "second_metadata") def test_update_metadata_name_conflicts_with_built_in_field( self, db_session_with_containers, mock_external_service_dependencies ): """ Test metadata name update fails when new name conflicts with built-in field names. """ # Arrange: Create test data account, tenant = self._create_test_account_and_tenant( db_session_with_containers, mock_external_service_dependencies ) dataset = self._create_test_dataset( db_session_with_containers, mock_external_service_dependencies, account, tenant ) # Setup mocks mock_external_service_dependencies["current_user"].current_tenant_id = tenant.id mock_external_service_dependencies["current_user"].id = account.id # Create metadata first metadata_args = MetadataArgs(type="string", name="old_name") metadata = MetadataService.create_metadata(dataset.id, metadata_args) # Try to update with built-in field name built_in_field_name = BuiltInField.document_name.value with pytest.raises(ValueError, match="Metadata name already exists in Built-in fields."): MetadataService.update_metadata_name(dataset.id, metadata.id, built_in_field_name) def test_update_metadata_name_not_found(self, db_session_with_containers, mock_external_service_dependencies): """ Test metadata name update fails when metadata ID does not exist. """ # Arrange: Create test data account, tenant = self._create_test_account_and_tenant( db_session_with_containers, mock_external_service_dependencies ) dataset = self._create_test_dataset( db_session_with_containers, mock_external_service_dependencies, account, tenant ) # Setup mocks mock_external_service_dependencies["current_user"].current_tenant_id = tenant.id mock_external_service_dependencies["current_user"].id = account.id # Try to update non-existent metadata import uuid fake_metadata_id = str(uuid.uuid4()) # Use valid UUID format new_name = "new_name" # Act: Execute the method under test result = MetadataService.update_metadata_name(dataset.id, fake_metadata_id, new_name) # Assert: Verify the method returns None when metadata is not found assert result is None def test_delete_metadata_success(self, db_session_with_containers, mock_external_service_dependencies): """ Test successful metadata deletion with valid parameters. """ # Arrange: Create test data account, tenant = self._create_test_account_and_tenant( db_session_with_containers, mock_external_service_dependencies ) dataset = self._create_test_dataset( db_session_with_containers, mock_external_service_dependencies, account, tenant ) # Setup mocks mock_external_service_dependencies["current_user"].current_tenant_id = tenant.id mock_external_service_dependencies["current_user"].id = account.id # Create metadata first metadata_args = MetadataArgs(type="string", name="to_be_deleted") metadata = MetadataService.create_metadata(dataset.id, metadata_args) # Act: Execute the method under test result = MetadataService.delete_metadata(dataset.id, metadata.id) # Assert: Verify the expected outcomes assert result is not None assert result.id == metadata.id # Verify metadata was deleted from database from extensions.ext_database import db deleted_metadata = db.session.query(DatasetMetadata).filter_by(id=metadata.id).first() assert deleted_metadata is None def test_delete_metadata_not_found(self, db_session_with_containers, mock_external_service_dependencies): """ Test metadata deletion fails when metadata ID does not exist. """ # Arrange: Create test data account, tenant = self._create_test_account_and_tenant( db_session_with_containers, mock_external_service_dependencies ) dataset = self._create_test_dataset( db_session_with_containers, mock_external_service_dependencies, account, tenant ) # Setup mocks mock_external_service_dependencies["current_user"].current_tenant_id = tenant.id mock_external_service_dependencies["current_user"].id = account.id # Try to delete non-existent metadata import uuid fake_metadata_id = str(uuid.uuid4()) # Use valid UUID format # Act: Execute the method under test result = MetadataService.delete_metadata(dataset.id, fake_metadata_id) # Assert: Verify the method returns None when metadata is not found assert result is None def test_delete_metadata_with_document_bindings( self, db_session_with_containers, mock_external_service_dependencies ): """ Test metadata deletion successfully removes document metadata bindings. """ # Arrange: Create test data account, tenant = self._create_test_account_and_tenant( db_session_with_containers, mock_external_service_dependencies ) dataset = self._create_test_dataset( db_session_with_containers, mock_external_service_dependencies, account, tenant ) document = self._create_test_document( db_session_with_containers, mock_external_service_dependencies, dataset, account ) # Setup mocks mock_external_service_dependencies["current_user"].current_tenant_id = tenant.id mock_external_service_dependencies["current_user"].id = account.id # Create metadata metadata_args = MetadataArgs(type="string", name="test_metadata") metadata = MetadataService.create_metadata(dataset.id, metadata_args) # Create metadata binding binding = DatasetMetadataBinding( tenant_id=tenant.id, dataset_id=dataset.id, metadata_id=metadata.id, document_id=document.id, created_by=account.id, ) from extensions.ext_database import db db.session.add(binding) db.session.commit() # Set document metadata document.doc_metadata = {"test_metadata": "test_value"} db.session.add(document) db.session.commit() # Act: Execute the method under test result = MetadataService.delete_metadata(dataset.id, metadata.id) # Assert: Verify the expected outcomes assert result is not None # Verify metadata was deleted from database deleted_metadata = db.session.query(DatasetMetadata).filter_by(id=metadata.id).first() assert deleted_metadata is None # Note: The service attempts to update document metadata but may not succeed # due to mock configuration. The main functionality (metadata deletion) is verified. def test_get_built_in_fields_success(self, db_session_with_containers, mock_external_service_dependencies): """ Test successful retrieval of built-in metadata fields. """ # Act: Execute the method under test result = MetadataService.get_built_in_fields() # Assert: Verify the expected outcomes assert result is not None assert len(result) == 5 # Verify all expected built-in fields are present field_names = [field["name"] for field in result] field_types = [field["type"] for field in result] assert BuiltInField.document_name.value in field_names assert BuiltInField.uploader.value in field_names assert BuiltInField.upload_date.value in field_names assert BuiltInField.last_update_date.value in field_names assert BuiltInField.source.value in field_names # Verify field types assert "string" in field_types assert "time" in field_types def test_enable_built_in_field_success(self, db_session_with_containers, mock_external_service_dependencies): """ Test successful enabling of built-in fields for a dataset. """ # Arrange: Create test data account, tenant = self._create_test_account_and_tenant( db_session_with_containers, mock_external_service_dependencies ) dataset = self._create_test_dataset( db_session_with_containers, mock_external_service_dependencies, account, tenant ) document = self._create_test_document( db_session_with_containers, mock_external_service_dependencies, dataset, account ) # Setup mocks mock_external_service_dependencies["current_user"].current_tenant_id = tenant.id mock_external_service_dependencies["current_user"].id = account.id # Mock DocumentService.get_working_documents_by_dataset_id mock_external_service_dependencies["document_service"].get_working_documents_by_dataset_id.return_value = [ document ] # Verify dataset starts with built-in fields disabled assert dataset.built_in_field_enabled is False # Act: Execute the method under test MetadataService.enable_built_in_field(dataset) # Assert: Verify the expected outcomes from extensions.ext_database import db db.session.refresh(dataset) assert dataset.built_in_field_enabled is True # Note: Document metadata update depends on DocumentService mock working correctly # The main functionality (enabling built-in fields) is verified def test_enable_built_in_field_already_enabled( self, db_session_with_containers, mock_external_service_dependencies ): """ Test enabling built-in fields when they are already enabled. """ # Arrange: Create test data account, tenant = self._create_test_account_and_tenant( db_session_with_containers, mock_external_service_dependencies ) dataset = self._create_test_dataset( db_session_with_containers, mock_external_service_dependencies, account, tenant ) # Setup mocks mock_external_service_dependencies["current_user"].current_tenant_id = tenant.id mock_external_service_dependencies["current_user"].id = account.id # Enable built-in fields first dataset.built_in_field_enabled = True from extensions.ext_database import db db.session.add(dataset) db.session.commit() # Mock DocumentService.get_working_documents_by_dataset_id mock_external_service_dependencies["document_service"].get_working_documents_by_dataset_id.return_value = [] # Act: Execute the method under test MetadataService.enable_built_in_field(dataset) # Assert: Verify the method returns early without changes db.session.refresh(dataset) assert dataset.built_in_field_enabled is True def test_enable_built_in_field_with_no_documents( self, db_session_with_containers, mock_external_service_dependencies ): """ Test enabling built-in fields for a dataset with no documents. """ # Arrange: Create test data account, tenant = self._create_test_account_and_tenant( db_session_with_containers, mock_external_service_dependencies ) dataset = self._create_test_dataset( db_session_with_containers, mock_external_service_dependencies, account, tenant ) # Setup mocks mock_external_service_dependencies["current_user"].current_tenant_id = tenant.id mock_external_service_dependencies["current_user"].id = account.id # Mock DocumentService.get_working_documents_by_dataset_id to return empty list mock_external_service_dependencies["document_service"].get_working_documents_by_dataset_id.return_value = [] # Act: Execute the method under test MetadataService.enable_built_in_field(dataset) # Assert: Verify the expected outcomes from extensions.ext_database import db db.session.refresh(dataset) assert dataset.built_in_field_enabled is True def test_disable_built_in_field_success(self, db_session_with_containers, mock_external_service_dependencies): """ Test successful disabling of built-in fields for a dataset. """ # Arrange: Create test data account, tenant = self._create_test_account_and_tenant( db_session_with_containers, mock_external_service_dependencies ) dataset = self._create_test_dataset( db_session_with_containers, mock_external_service_dependencies, account, tenant ) document = self._create_test_document( db_session_with_containers, mock_external_service_dependencies, dataset, account ) # Setup mocks mock_external_service_dependencies["current_user"].current_tenant_id = tenant.id mock_external_service_dependencies["current_user"].id = account.id # Enable built-in fields first dataset.built_in_field_enabled = True from extensions.ext_database import db db.session.add(dataset) db.session.commit() # Set document metadata with built-in fields document.doc_metadata = { BuiltInField.document_name.value: document.name, BuiltInField.uploader.value: "test_uploader", BuiltInField.upload_date.value: 1234567890.0, BuiltInField.last_update_date.value: 1234567890.0, BuiltInField.source.value: "test_source", } db.session.add(document) db.session.commit() # Mock DocumentService.get_working_documents_by_dataset_id mock_external_service_dependencies["document_service"].get_working_documents_by_dataset_id.return_value = [ document ] # Act: Execute the method under test MetadataService.disable_built_in_field(dataset) # Assert: Verify the expected outcomes db.session.refresh(dataset) assert dataset.built_in_field_enabled is False # Note: Document metadata update depends on DocumentService mock working correctly # The main functionality (disabling built-in fields) is verified def test_disable_built_in_field_already_disabled( self, db_session_with_containers, mock_external_service_dependencies ): """ Test disabling built-in fields when they are already disabled. """ # Arrange: Create test data account, tenant = self._create_test_account_and_tenant( db_session_with_containers, mock_external_service_dependencies ) dataset = self._create_test_dataset( db_session_with_containers, mock_external_service_dependencies, account, tenant ) # Setup mocks mock_external_service_dependencies["current_user"].current_tenant_id = tenant.id mock_external_service_dependencies["current_user"].id = account.id # Verify dataset starts with built-in fields disabled assert dataset.built_in_field_enabled is False # Mock DocumentService.get_working_documents_by_dataset_id mock_external_service_dependencies["document_service"].get_working_documents_by_dataset_id.return_value = [] # Act: Execute the method under test MetadataService.disable_built_in_field(dataset) # Assert: Verify the method returns early without changes from extensions.ext_database import db db.session.refresh(dataset) assert dataset.built_in_field_enabled is False def test_disable_built_in_field_with_no_documents( self, db_session_with_containers, mock_external_service_dependencies ): """ Test disabling built-in fields for a dataset with no documents. """ # Arrange: Create test data account, tenant = self._create_test_account_and_tenant( db_session_with_containers, mock_external_service_dependencies ) dataset = self._create_test_dataset( db_session_with_containers, mock_external_service_dependencies, account, tenant ) # Setup mocks mock_external_service_dependencies["current_user"].current_tenant_id = tenant.id mock_external_service_dependencies["current_user"].id = account.id # Enable built-in fields first dataset.built_in_field_enabled = True from extensions.ext_database import db db.session.add(dataset) db.session.commit() # Mock DocumentService.get_working_documents_by_dataset_id to return empty list mock_external_service_dependencies["document_service"].get_working_documents_by_dataset_id.return_value = [] # Act: Execute the method under test MetadataService.disable_built_in_field(dataset) # Assert: Verify the expected outcomes db.session.refresh(dataset) assert dataset.built_in_field_enabled is False def test_update_documents_metadata_success(self, db_session_with_containers, mock_external_service_dependencies): """ Test successful update of documents metadata. """ # Arrange: Create test data account, tenant = self._create_test_account_and_tenant( db_session_with_containers, mock_external_service_dependencies ) dataset = self._create_test_dataset( db_session_with_containers, mock_external_service_dependencies, account, tenant ) document = self._create_test_document( db_session_with_containers, mock_external_service_dependencies, dataset, account ) # Setup mocks mock_external_service_dependencies["current_user"].current_tenant_id = tenant.id mock_external_service_dependencies["current_user"].id = account.id # Create metadata metadata_args = MetadataArgs(type="string", name="test_metadata") metadata = MetadataService.create_metadata(dataset.id, metadata_args) # Mock DocumentService.get_document mock_external_service_dependencies["document_service"].get_document.return_value = document # Create metadata operation data from services.entities.knowledge_entities.knowledge_entities import ( DocumentMetadataOperation, MetadataDetail, MetadataOperationData, ) metadata_detail = MetadataDetail(id=metadata.id, name=metadata.name, value="test_value") operation = DocumentMetadataOperation(document_id=document.id, metadata_list=[metadata_detail]) operation_data = MetadataOperationData(operation_data=[operation]) # Act: Execute the method under test MetadataService.update_documents_metadata(dataset, operation_data) # Assert: Verify the expected outcomes from extensions.ext_database import db # Verify document metadata was updated db.session.refresh(document) assert document.doc_metadata is not None assert "test_metadata" in document.doc_metadata assert document.doc_metadata["test_metadata"] == "test_value" # Verify metadata binding was created binding = ( db.session.query(DatasetMetadataBinding).filter_by(metadata_id=metadata.id, document_id=document.id).first() ) assert binding is not None assert binding.tenant_id == tenant.id assert binding.dataset_id == dataset.id def test_update_documents_metadata_with_built_in_fields_enabled( self, db_session_with_containers, mock_external_service_dependencies ): """ Test update of documents metadata when built-in fields are enabled. """ # Arrange: Create test data account, tenant = self._create_test_account_and_tenant( db_session_with_containers, mock_external_service_dependencies ) dataset = self._create_test_dataset( db_session_with_containers, mock_external_service_dependencies, account, tenant ) document = self._create_test_document( db_session_with_containers, mock_external_service_dependencies, dataset, account ) # Enable built-in fields dataset.built_in_field_enabled = True from extensions.ext_database import db db.session.add(dataset) db.session.commit() # Setup mocks mock_external_service_dependencies["current_user"].current_tenant_id = tenant.id mock_external_service_dependencies["current_user"].id = account.id # Create metadata metadata_args = MetadataArgs(type="string", name="test_metadata") metadata = MetadataService.create_metadata(dataset.id, metadata_args) # Mock DocumentService.get_document mock_external_service_dependencies["document_service"].get_document.return_value = document # Create metadata operation data from services.entities.knowledge_entities.knowledge_entities import ( DocumentMetadataOperation, MetadataDetail, MetadataOperationData, ) metadata_detail = MetadataDetail(id=metadata.id, name=metadata.name, value="test_value") operation = DocumentMetadataOperation(document_id=document.id, metadata_list=[metadata_detail]) operation_data = MetadataOperationData(operation_data=[operation]) # Act: Execute the method under test MetadataService.update_documents_metadata(dataset, operation_data) # Assert: Verify the expected outcomes # Verify document metadata was updated with both custom and built-in fields db.session.refresh(document) assert document.doc_metadata is not None assert "test_metadata" in document.doc_metadata assert document.doc_metadata["test_metadata"] == "test_value" # Note: Built-in fields would be added if DocumentService mock works correctly # The main functionality (custom metadata update) is verified def test_update_documents_metadata_document_not_found( self, db_session_with_containers, mock_external_service_dependencies ): """ Test update of documents metadata when document is not found. """ # Arrange: Create test data account, tenant = self._create_test_account_and_tenant( db_session_with_containers, mock_external_service_dependencies ) dataset = self._create_test_dataset( db_session_with_containers, mock_external_service_dependencies, account, tenant ) # Setup mocks mock_external_service_dependencies["current_user"].current_tenant_id = tenant.id mock_external_service_dependencies["current_user"].id = account.id # Create metadata metadata_args = MetadataArgs(type="string", name="test_metadata") metadata = MetadataService.create_metadata(dataset.id, metadata_args) # Mock DocumentService.get_document to return None (document not found) mock_external_service_dependencies["document_service"].get_document.return_value = None # Create metadata operation data from services.entities.knowledge_entities.knowledge_entities import ( DocumentMetadataOperation, MetadataDetail, MetadataOperationData, ) metadata_detail = MetadataDetail(id=metadata.id, name=metadata.name, value="test_value") operation = DocumentMetadataOperation(document_id="non-existent-document-id", metadata_list=[metadata_detail]) operation_data = MetadataOperationData(operation_data=[operation]) # Act: Execute the method under test # The method should handle the error gracefully and continue MetadataService.update_documents_metadata(dataset, operation_data) # Assert: Verify the method completes without raising exceptions # The main functionality (error handling) is verified def test_knowledge_base_metadata_lock_check_dataset_id( self, db_session_with_containers, mock_external_service_dependencies ): """ Test metadata lock check for dataset operations. """ # Arrange: Setup mocks mock_external_service_dependencies["redis_client"].get.return_value = None mock_external_service_dependencies["redis_client"].set.return_value = True dataset_id = "test-dataset-id" # Act: Execute the method under test MetadataService.knowledge_base_metadata_lock_check(dataset_id, None) # Assert: Verify the expected outcomes # Verify Redis lock was set mock_external_service_dependencies["redis_client"].set.assert_called_once() # Verify lock key format call_args = mock_external_service_dependencies["redis_client"].set.call_args assert call_args[0][0] == f"dataset_metadata_lock_{dataset_id}" def test_knowledge_base_metadata_lock_check_document_id( self, db_session_with_containers, mock_external_service_dependencies ): """ Test metadata lock check for document operations. """ # Arrange: Setup mocks mock_external_service_dependencies["redis_client"].get.return_value = None mock_external_service_dependencies["redis_client"].set.return_value = True document_id = "test-document-id" # Act: Execute the method under test MetadataService.knowledge_base_metadata_lock_check(None, document_id) # Assert: Verify the expected outcomes # Verify Redis lock was set mock_external_service_dependencies["redis_client"].set.assert_called_once() # Verify lock key format call_args = mock_external_service_dependencies["redis_client"].set.call_args assert call_args[0][0] == f"document_metadata_lock_{document_id}" def test_knowledge_base_metadata_lock_check_lock_exists( self, db_session_with_containers, mock_external_service_dependencies ): """ Test metadata lock check when lock already exists. """ # Arrange: Setup mocks to simulate existing lock mock_external_service_dependencies["redis_client"].get.return_value = "1" # Lock exists dataset_id = "test-dataset-id" # Act & Assert: Verify proper error handling with pytest.raises( ValueError, match="Another knowledge base metadata operation is running, please wait a moment." ): MetadataService.knowledge_base_metadata_lock_check(dataset_id, None) def test_knowledge_base_metadata_lock_check_document_lock_exists( self, db_session_with_containers, mock_external_service_dependencies ): """ Test metadata lock check when document lock already exists. """ # Arrange: Setup mocks to simulate existing lock mock_external_service_dependencies["redis_client"].get.return_value = "1" # Lock exists document_id = "test-document-id" # Act & Assert: Verify proper error handling with pytest.raises(ValueError, match="Another document metadata operation is running, please wait a moment."): MetadataService.knowledge_base_metadata_lock_check(None, document_id) def test_get_dataset_metadatas_success(self, db_session_with_containers, mock_external_service_dependencies): """ Test successful retrieval of dataset metadata information. """ # Arrange: Create test data account, tenant = self._create_test_account_and_tenant( db_session_with_containers, mock_external_service_dependencies ) dataset = self._create_test_dataset( db_session_with_containers, mock_external_service_dependencies, account, tenant ) # Setup mocks mock_external_service_dependencies["current_user"].current_tenant_id = tenant.id mock_external_service_dependencies["current_user"].id = account.id # Create metadata metadata_args = MetadataArgs(type="string", name="test_metadata") metadata = MetadataService.create_metadata(dataset.id, metadata_args) # Create document and metadata binding document = self._create_test_document( db_session_with_containers, mock_external_service_dependencies, dataset, account ) binding = DatasetMetadataBinding( tenant_id=tenant.id, dataset_id=dataset.id, metadata_id=metadata.id, document_id=document.id, created_by=account.id, ) from extensions.ext_database import db db.session.add(binding) db.session.commit() # Act: Execute the method under test result = MetadataService.get_dataset_metadatas(dataset) # Assert: Verify the expected outcomes assert result is not None assert "doc_metadata" in result assert "built_in_field_enabled" in result # Verify metadata information doc_metadata = result["doc_metadata"] assert len(doc_metadata) == 1 assert doc_metadata[0]["id"] == metadata.id assert doc_metadata[0]["name"] == metadata.name assert doc_metadata[0]["type"] == metadata.type assert doc_metadata[0]["count"] == 1 # One document bound to this metadata # Verify built-in field status assert result["built_in_field_enabled"] is False def test_get_dataset_metadatas_with_built_in_fields_enabled( self, db_session_with_containers, mock_external_service_dependencies ): """ Test retrieval of dataset metadata when built-in fields are enabled. """ # Arrange: Create test data account, tenant = self._create_test_account_and_tenant( db_session_with_containers, mock_external_service_dependencies ) dataset = self._create_test_dataset( db_session_with_containers, mock_external_service_dependencies, account, tenant ) # Enable built-in fields dataset.built_in_field_enabled = True from extensions.ext_database import db db.session.add(dataset) db.session.commit() # Setup mocks mock_external_service_dependencies["current_user"].current_tenant_id = tenant.id mock_external_service_dependencies["current_user"].id = account.id # Create metadata metadata_args = MetadataArgs(type="string", name="test_metadata") metadata = MetadataService.create_metadata(dataset.id, metadata_args) # Act: Execute the method under test result = MetadataService.get_dataset_metadatas(dataset) # Assert: Verify the expected outcomes assert result is not None assert "doc_metadata" in result assert "built_in_field_enabled" in result # Verify metadata information doc_metadata = result["doc_metadata"] assert len(doc_metadata) == 1 # Only custom metadata, built-in fields are not included in this list # Verify built-in field status assert result["built_in_field_enabled"] is True def test_get_dataset_metadatas_no_metadata(self, db_session_with_containers, mock_external_service_dependencies): """ Test retrieval of dataset metadata when no metadata exists. """ # Arrange: Create test data account, tenant = self._create_test_account_and_tenant( db_session_with_containers, mock_external_service_dependencies ) dataset = self._create_test_dataset( db_session_with_containers, mock_external_service_dependencies, account, tenant ) # Act: Execute the method under test result = MetadataService.get_dataset_metadatas(dataset) # Assert: Verify the expected outcomes assert result is not None assert "doc_metadata" in result assert "built_in_field_enabled" in result # Verify metadata information doc_metadata = result["doc_metadata"] assert len(doc_metadata) == 0 # No metadata exists # Verify built-in field status assert result["built_in_field_enabled"] is False