From 6ff420cd03490a8e944e0484db7edf9c663b6b5e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9C=A8=E4=B9=8B=E6=9C=AC=E6=BE=AA?= Date: Wed, 25 Feb 2026 14:07:28 +0800 Subject: [PATCH] test: migrate dataset service update-delete SQL tests to testcontainers (#32548) Co-authored-by: KinomotoMio <200703522+KinomotoMio@users.noreply.github.com> Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- .../services/dataset_service_update_delete.py | 359 +++++++++++++++ .../services/dataset_service_update_delete.py | 416 ------------------ 2 files changed, 359 insertions(+), 416 deletions(-) create mode 100644 api/tests/test_containers_integration_tests/services/dataset_service_update_delete.py diff --git a/api/tests/test_containers_integration_tests/services/dataset_service_update_delete.py b/api/tests/test_containers_integration_tests/services/dataset_service_update_delete.py new file mode 100644 index 0000000000..9871ef37e6 --- /dev/null +++ b/api/tests/test_containers_integration_tests/services/dataset_service_update_delete.py @@ -0,0 +1,359 @@ +""" +Integration tests for DatasetService update and delete operations using a real database. + +This module contains comprehensive integration tests for the DatasetService class, +specifically focusing on update and delete operations for datasets backed by Testcontainers. +""" + +import datetime +from unittest.mock import patch +from uuid import uuid4 + +import pytest +from werkzeug.exceptions import NotFound + +from extensions.ext_database import db +from models import Account, Tenant, TenantAccountJoin, TenantAccountRole +from models.dataset import AppDatasetJoin, Dataset, DatasetPermissionEnum +from models.model import App +from services.dataset_service import DatasetService +from services.errors.account import NoPermissionError + + +class DatasetUpdateDeleteTestDataFactory: + """ + Factory class for creating test data and mock objects for dataset update/delete tests. + """ + + @staticmethod + def create_account_with_tenant( + role: TenantAccountRole = TenantAccountRole.NORMAL, + tenant: Tenant | None = None, + ) -> tuple[Account, Tenant]: + """Create a real account and tenant with specified role.""" + account = Account( + email=f"{uuid4()}@example.com", + name=f"user-{uuid4()}", + interface_language="en-US", + status="active", + ) + db.session.add(account) + db.session.commit() + + if tenant is None: + tenant = Tenant(name=f"tenant-{uuid4()}", status="normal") + db.session.add(tenant) + db.session.commit() + + join = TenantAccountJoin( + tenant_id=tenant.id, + account_id=account.id, + role=role, + current=True, + ) + db.session.add(join) + db.session.commit() + + account.current_tenant = tenant + return account, tenant + + @staticmethod + def create_dataset( + tenant_id: str, + created_by: str, + name: str = "Test Dataset", + enable_api: bool = True, + permission: DatasetPermissionEnum = DatasetPermissionEnum.ONLY_ME, + ) -> Dataset: + """Create a real dataset with specified attributes.""" + dataset = Dataset( + tenant_id=tenant_id, + name=name, + description="Test description", + data_source_type="upload_file", + indexing_technique="high_quality", + created_by=created_by, + permission=permission, + provider="vendor", + retrieval_model={"top_k": 2}, + enable_api=enable_api, + ) + db.session.add(dataset) + db.session.commit() + return dataset + + @staticmethod + def create_app(tenant_id: str, created_by: str, name: str = "Test App") -> App: + """Create a real app for AppDatasetJoin.""" + app = App( + tenant_id=tenant_id, + name=name, + mode="chat", + icon_type="emoji", + icon="icon", + icon_background="#FFFFFF", + enable_site=True, + enable_api=True, + created_by=created_by, + ) + db.session.add(app) + db.session.commit() + return app + + @staticmethod + def create_app_dataset_join(app_id: str, dataset_id: str) -> AppDatasetJoin: + """Create a real AppDatasetJoin record.""" + join = AppDatasetJoin(app_id=app_id, dataset_id=dataset_id) + db.session.add(join) + db.session.commit() + return join + + +class TestDatasetServiceDeleteDataset: + """ + Comprehensive integration tests for DatasetService.delete_dataset method. + """ + + def test_delete_dataset_success(self, db_session_with_containers): + """ + Test successful deletion of a dataset. + + Verifies that when all validation passes, a dataset is deleted + correctly with proper event signaling and database cleanup. + + This test ensures: + - Dataset is retrieved correctly + - Permission is checked + - Event is sent for cleanup + - Dataset is deleted from database + - Transaction is committed + - Method returns True + """ + # Arrange + owner, tenant = DatasetUpdateDeleteTestDataFactory.create_account_with_tenant(role=TenantAccountRole.OWNER) + dataset = DatasetUpdateDeleteTestDataFactory.create_dataset(tenant.id, owner.id) + + # Act + with patch("services.dataset_service.dataset_was_deleted") as mock_dataset_was_deleted: + result = DatasetService.delete_dataset(dataset.id, owner) + + # Assert + assert result is True + assert db.session.get(Dataset, dataset.id) is None + mock_dataset_was_deleted.send.assert_called_once_with(dataset) + + def test_delete_dataset_not_found(self, db_session_with_containers): + """ + Test handling when dataset is not found. + + Verifies that when the dataset ID doesn't exist, the method + returns False without performing any operations. + + This test ensures: + - Method returns False when dataset not found + - No permission checks are performed + - No events are sent + - No database operations are performed + """ + # Arrange + owner, _ = DatasetUpdateDeleteTestDataFactory.create_account_with_tenant(role=TenantAccountRole.OWNER) + dataset_id = str(uuid4()) + + # Act + result = DatasetService.delete_dataset(dataset_id, owner) + + # Assert + assert result is False + + def test_delete_dataset_permission_denied_error(self, db_session_with_containers): + """ + Test error handling when user lacks permission. + + Verifies that when the user doesn't have permission to delete + the dataset, a NoPermissionError is raised. + + This test ensures: + - Permission validation works correctly + - Error is raised before deletion + - No database operations are performed + """ + # Arrange + owner, tenant = DatasetUpdateDeleteTestDataFactory.create_account_with_tenant(role=TenantAccountRole.OWNER) + normal_user, _ = DatasetUpdateDeleteTestDataFactory.create_account_with_tenant( + role=TenantAccountRole.NORMAL, + tenant=tenant, + ) + dataset = DatasetUpdateDeleteTestDataFactory.create_dataset(tenant.id, owner.id) + + # Act & Assert + with pytest.raises(NoPermissionError): + DatasetService.delete_dataset(dataset.id, normal_user) + + # Verify no deletion was attempted + assert db.session.get(Dataset, dataset.id) is not None + + +class TestDatasetServiceDatasetUseCheck: + """ + Comprehensive integration tests for DatasetService.dataset_use_check method. + """ + + def test_dataset_use_check_in_use(self, db_session_with_containers): + """ + Test detection when dataset is in use. + + Verifies that when a dataset has associated AppDatasetJoin records, + the method returns True. + + This test ensures: + - Query is constructed correctly + - True is returned when dataset is in use + - Database query is executed + """ + # Arrange + owner, tenant = DatasetUpdateDeleteTestDataFactory.create_account_with_tenant(role=TenantAccountRole.OWNER) + dataset = DatasetUpdateDeleteTestDataFactory.create_dataset(tenant.id, owner.id) + app = DatasetUpdateDeleteTestDataFactory.create_app(tenant.id, owner.id) + DatasetUpdateDeleteTestDataFactory.create_app_dataset_join(app.id, dataset.id) + + # Act + result = DatasetService.dataset_use_check(dataset.id) + + # Assert + assert result is True + + def test_dataset_use_check_not_in_use(self, db_session_with_containers): + """ + Test detection when dataset is not in use. + + Verifies that when a dataset has no associated AppDatasetJoin records, + the method returns False. + + This test ensures: + - Query is constructed correctly + - False is returned when dataset is not in use + - Database query is executed + """ + # Arrange + owner, tenant = DatasetUpdateDeleteTestDataFactory.create_account_with_tenant(role=TenantAccountRole.OWNER) + dataset = DatasetUpdateDeleteTestDataFactory.create_dataset(tenant.id, owner.id) + + # Act + result = DatasetService.dataset_use_check(dataset.id) + + # Assert + assert result is False + + +class TestDatasetServiceUpdateDatasetApiStatus: + """ + Comprehensive integration tests for DatasetService.update_dataset_api_status method. + """ + + def test_update_dataset_api_status_enable_success(self, db_session_with_containers): + """ + Test successful enabling of dataset API access. + + Verifies that when all validation passes, the dataset's API + access is enabled and the update is committed. + + This test ensures: + - Dataset is retrieved correctly + - enable_api is set to True + - updated_by and updated_at are set + - Transaction is committed + """ + # Arrange + owner, tenant = DatasetUpdateDeleteTestDataFactory.create_account_with_tenant(role=TenantAccountRole.OWNER) + dataset = DatasetUpdateDeleteTestDataFactory.create_dataset(tenant.id, owner.id, enable_api=False) + current_time = datetime.datetime(2023, 1, 1, 12, 0, 0) + + # Act + with ( + patch("services.dataset_service.current_user", owner), + patch("services.dataset_service.naive_utc_now", return_value=current_time), + ): + DatasetService.update_dataset_api_status(dataset.id, True) + + # Assert + db.session.refresh(dataset) + assert dataset.enable_api is True + assert dataset.updated_by == owner.id + assert dataset.updated_at == current_time + + def test_update_dataset_api_status_disable_success(self, db_session_with_containers): + """ + Test successful disabling of dataset API access. + + Verifies that when all validation passes, the dataset's API + access is disabled and the update is committed. + + This test ensures: + - Dataset is retrieved correctly + - enable_api is set to False + - updated_by and updated_at are set + - Transaction is committed + """ + # Arrange + owner, tenant = DatasetUpdateDeleteTestDataFactory.create_account_with_tenant(role=TenantAccountRole.OWNER) + dataset = DatasetUpdateDeleteTestDataFactory.create_dataset(tenant.id, owner.id, enable_api=True) + current_time = datetime.datetime(2023, 1, 1, 12, 0, 0) + + # Act + with ( + patch("services.dataset_service.current_user", owner), + patch("services.dataset_service.naive_utc_now", return_value=current_time), + ): + DatasetService.update_dataset_api_status(dataset.id, False) + + # Assert + db.session.refresh(dataset) + assert dataset.enable_api is False + assert dataset.updated_by == owner.id + + def test_update_dataset_api_status_not_found_error(self, db_session_with_containers): + """ + Test error handling when dataset is not found. + + Verifies that when the dataset ID doesn't exist, a NotFound + exception is raised. + + This test ensures: + - NotFound exception is raised + - No updates are performed + - Error message is appropriate + """ + # Arrange + dataset_id = str(uuid4()) + + # Act & Assert + with pytest.raises(NotFound, match="Dataset not found"): + DatasetService.update_dataset_api_status(dataset_id, True) + + def test_update_dataset_api_status_missing_current_user_error(self, db_session_with_containers): + """ + Test error handling when current_user is missing. + + Verifies that when current_user is None or has no ID, a ValueError + is raised. + + This test ensures: + - ValueError is raised when current_user is None + - Error message is clear + - No updates are committed + """ + # Arrange + owner, tenant = DatasetUpdateDeleteTestDataFactory.create_account_with_tenant(role=TenantAccountRole.OWNER) + dataset = DatasetUpdateDeleteTestDataFactory.create_dataset(tenant.id, owner.id, enable_api=False) + + # Act & Assert + with ( + patch("services.dataset_service.current_user", None), + pytest.raises(ValueError, match="Current user or current user id not found"), + ): + DatasetService.update_dataset_api_status(dataset.id, True) + + # Verify no commit was attempted + db.session.rollback() + db.session.refresh(dataset) + assert dataset.enable_api is False diff --git a/api/tests/unit_tests/services/dataset_service_update_delete.py b/api/tests/unit_tests/services/dataset_service_update_delete.py index 3715aadfdc..5deec10d5e 100644 --- a/api/tests/unit_tests/services/dataset_service_update_delete.py +++ b/api/tests/unit_tests/services/dataset_service_update_delete.py @@ -96,7 +96,6 @@ from unittest.mock import Mock, create_autospec, patch import pytest from sqlalchemy.orm import Session -from werkzeug.exceptions import NotFound from models import Account, TenantAccountRole from models.dataset import ( @@ -536,421 +535,6 @@ class TestDatasetServiceUpdateDataset: DatasetService.update_dataset(dataset_id, update_data, user) -# ============================================================================ -# Tests for delete_dataset -# ============================================================================ - - -class TestDatasetServiceDeleteDataset: - """ - Comprehensive unit tests for DatasetService.delete_dataset method. - - This test class covers the dataset deletion functionality, including - permission validation, event signaling, and database cleanup. - - The delete_dataset method: - 1. Retrieves the dataset by ID - 2. Returns False if dataset not found - 3. Validates user permissions - 4. Sends dataset_was_deleted event - 5. Deletes dataset from database - 6. Commits transaction - 7. Returns True on success - - Test scenarios include: - - Successful dataset deletion - - Permission validation - - Event signaling - - Database cleanup - - Not found handling - """ - - @pytest.fixture - def mock_dataset_service_dependencies(self): - """ - Mock dataset service dependencies for testing. - - Provides mocked dependencies including: - - get_dataset method - - check_dataset_permission method - - dataset_was_deleted event signal - - Database session - """ - with ( - patch("services.dataset_service.DatasetService.get_dataset") as mock_get_dataset, - patch("services.dataset_service.DatasetService.check_dataset_permission") as mock_check_perm, - patch("services.dataset_service.dataset_was_deleted") as mock_event, - patch("extensions.ext_database.db.session") as mock_db, - ): - yield { - "get_dataset": mock_get_dataset, - "check_permission": mock_check_perm, - "dataset_was_deleted": mock_event, - "db_session": mock_db, - } - - def test_delete_dataset_success(self, mock_dataset_service_dependencies): - """ - Test successful deletion of a dataset. - - Verifies that when all validation passes, a dataset is deleted - correctly with proper event signaling and database cleanup. - - This test ensures: - - Dataset is retrieved correctly - - Permission is checked - - Event is sent for cleanup - - Dataset is deleted from database - - Transaction is committed - - Method returns True - """ - # Arrange - dataset_id = "dataset-123" - dataset = DatasetUpdateDeleteTestDataFactory.create_dataset_mock(dataset_id=dataset_id) - user = DatasetUpdateDeleteTestDataFactory.create_user_mock() - - mock_dataset_service_dependencies["get_dataset"].return_value = dataset - - # Act - result = DatasetService.delete_dataset(dataset_id, user) - - # Assert - assert result is True - - # Verify dataset was retrieved - mock_dataset_service_dependencies["get_dataset"].assert_called_once_with(dataset_id) - - # Verify permission was checked - mock_dataset_service_dependencies["check_permission"].assert_called_once_with(dataset, user) - - # Verify event was sent for cleanup - mock_dataset_service_dependencies["dataset_was_deleted"].send.assert_called_once_with(dataset) - - # Verify dataset was deleted and committed - mock_dataset_service_dependencies["db_session"].delete.assert_called_once_with(dataset) - mock_dataset_service_dependencies["db_session"].commit.assert_called_once() - - def test_delete_dataset_not_found(self, mock_dataset_service_dependencies): - """ - Test handling when dataset is not found. - - Verifies that when the dataset ID doesn't exist, the method - returns False without performing any operations. - - This test ensures: - - Method returns False when dataset not found - - No permission checks are performed - - No events are sent - - No database operations are performed - """ - # Arrange - dataset_id = "non-existent-dataset" - user = DatasetUpdateDeleteTestDataFactory.create_user_mock() - - mock_dataset_service_dependencies["get_dataset"].return_value = None - - # Act - result = DatasetService.delete_dataset(dataset_id, user) - - # Assert - assert result is False - - # Verify no operations were performed - mock_dataset_service_dependencies["check_permission"].assert_not_called() - mock_dataset_service_dependencies["dataset_was_deleted"].send.assert_not_called() - mock_dataset_service_dependencies["db_session"].delete.assert_not_called() - - def test_delete_dataset_permission_denied_error(self, mock_dataset_service_dependencies): - """ - Test error handling when user lacks permission. - - Verifies that when the user doesn't have permission to delete - the dataset, a NoPermissionError is raised. - - This test ensures: - - Permission validation works correctly - - Error is raised before deletion - - No database operations are performed - """ - # Arrange - dataset_id = "dataset-123" - dataset = DatasetUpdateDeleteTestDataFactory.create_dataset_mock(dataset_id=dataset_id) - user = DatasetUpdateDeleteTestDataFactory.create_user_mock() - - mock_dataset_service_dependencies["get_dataset"].return_value = dataset - mock_dataset_service_dependencies["check_permission"].side_effect = NoPermissionError("No permission") - - # Act & Assert - with pytest.raises(NoPermissionError): - DatasetService.delete_dataset(dataset_id, user) - - # Verify no deletion was attempted - mock_dataset_service_dependencies["db_session"].delete.assert_not_called() - - -# ============================================================================ -# Tests for dataset_use_check -# ============================================================================ - - -class TestDatasetServiceDatasetUseCheck: - """ - Comprehensive unit tests for DatasetService.dataset_use_check method. - - This test class covers the dataset use checking functionality, which - determines if a dataset is currently being used by any applications. - - The dataset_use_check method: - 1. Queries AppDatasetJoin table for the dataset ID - 2. Returns True if dataset is in use - 3. Returns False if dataset is not in use - - Test scenarios include: - - Dataset in use (has AppDatasetJoin records) - - Dataset not in use (no AppDatasetJoin records) - - Database query validation - """ - - @pytest.fixture - def mock_db_session(self): - """ - Mock database session for testing. - - Provides a mocked database session that can be used to verify - query construction and execution. - """ - with patch("services.dataset_service.db.session") as mock_db: - yield mock_db - - def test_dataset_use_check_in_use(self, mock_db_session): - """ - Test detection when dataset is in use. - - Verifies that when a dataset has associated AppDatasetJoin records, - the method returns True. - - This test ensures: - - Query is constructed correctly - - True is returned when dataset is in use - - Database query is executed - """ - # Arrange - dataset_id = "dataset-123" - - # Mock the exists() query to return True - mock_execute = Mock() - mock_execute.scalar_one.return_value = True - mock_db_session.execute.return_value = mock_execute - - # Act - result = DatasetService.dataset_use_check(dataset_id) - - # Assert - assert result is True - - # Verify query was executed - mock_db_session.execute.assert_called_once() - - def test_dataset_use_check_not_in_use(self, mock_db_session): - """ - Test detection when dataset is not in use. - - Verifies that when a dataset has no associated AppDatasetJoin records, - the method returns False. - - This test ensures: - - Query is constructed correctly - - False is returned when dataset is not in use - - Database query is executed - """ - # Arrange - dataset_id = "dataset-123" - - # Mock the exists() query to return False - mock_execute = Mock() - mock_execute.scalar_one.return_value = False - mock_db_session.execute.return_value = mock_execute - - # Act - result = DatasetService.dataset_use_check(dataset_id) - - # Assert - assert result is False - - # Verify query was executed - mock_db_session.execute.assert_called_once() - - -# ============================================================================ -# Tests for update_dataset_api_status -# ============================================================================ - - -class TestDatasetServiceUpdateDatasetApiStatus: - """ - Comprehensive unit tests for DatasetService.update_dataset_api_status method. - - This test class covers the dataset API status update functionality, - which enables or disables API access for a dataset. - - The update_dataset_api_status method: - 1. Retrieves the dataset by ID - 2. Validates dataset exists - 3. Updates enable_api field - 4. Updates updated_by and updated_at fields - 5. Commits transaction - - Test scenarios include: - - Successful API status enable - - Successful API status disable - - Dataset not found error - - Current user validation - """ - - @pytest.fixture - def mock_dataset_service_dependencies(self): - """ - Mock dataset service dependencies for testing. - - Provides mocked dependencies including: - - get_dataset method - - current_user context - - Database session - - Current time utilities - """ - with ( - patch("services.dataset_service.DatasetService.get_dataset") as mock_get_dataset, - patch( - "services.dataset_service.current_user", create_autospec(Account, instance=True) - ) as mock_current_user, - patch("extensions.ext_database.db.session") as mock_db, - patch("services.dataset_service.naive_utc_now") as mock_naive_utc_now, - ): - current_time = datetime.datetime(2023, 1, 1, 12, 0, 0) - mock_naive_utc_now.return_value = current_time - mock_current_user.id = "user-123" - - yield { - "get_dataset": mock_get_dataset, - "current_user": mock_current_user, - "db_session": mock_db, - "naive_utc_now": mock_naive_utc_now, - "current_time": current_time, - } - - def test_update_dataset_api_status_enable_success(self, mock_dataset_service_dependencies): - """ - Test successful enabling of dataset API access. - - Verifies that when all validation passes, the dataset's API - access is enabled and the update is committed. - - This test ensures: - - Dataset is retrieved correctly - - enable_api is set to True - - updated_by and updated_at are set - - Transaction is committed - """ - # Arrange - dataset_id = "dataset-123" - dataset = DatasetUpdateDeleteTestDataFactory.create_dataset_mock(dataset_id=dataset_id, enable_api=False) - - mock_dataset_service_dependencies["get_dataset"].return_value = dataset - - # Act - DatasetService.update_dataset_api_status(dataset_id, True) - - # Assert - assert dataset.enable_api is True - assert dataset.updated_by == "user-123" - assert dataset.updated_at == mock_dataset_service_dependencies["current_time"] - - # Verify dataset was retrieved - mock_dataset_service_dependencies["get_dataset"].assert_called_once_with(dataset_id) - - # Verify transaction was committed - mock_dataset_service_dependencies["db_session"].commit.assert_called_once() - - def test_update_dataset_api_status_disable_success(self, mock_dataset_service_dependencies): - """ - Test successful disabling of dataset API access. - - Verifies that when all validation passes, the dataset's API - access is disabled and the update is committed. - - This test ensures: - - Dataset is retrieved correctly - - enable_api is set to False - - updated_by and updated_at are set - - Transaction is committed - """ - # Arrange - dataset_id = "dataset-123" - dataset = DatasetUpdateDeleteTestDataFactory.create_dataset_mock(dataset_id=dataset_id, enable_api=True) - - mock_dataset_service_dependencies["get_dataset"].return_value = dataset - - # Act - DatasetService.update_dataset_api_status(dataset_id, False) - - # Assert - assert dataset.enable_api is False - assert dataset.updated_by == "user-123" - - # Verify transaction was committed - mock_dataset_service_dependencies["db_session"].commit.assert_called_once() - - def test_update_dataset_api_status_not_found_error(self, mock_dataset_service_dependencies): - """ - Test error handling when dataset is not found. - - Verifies that when the dataset ID doesn't exist, a NotFound - exception is raised. - - This test ensures: - - NotFound exception is raised - - No updates are performed - - Error message is appropriate - """ - # Arrange - dataset_id = "non-existent-dataset" - - mock_dataset_service_dependencies["get_dataset"].return_value = None - - # Act & Assert - with pytest.raises(NotFound, match="Dataset not found"): - DatasetService.update_dataset_api_status(dataset_id, True) - - # Verify no commit was attempted - mock_dataset_service_dependencies["db_session"].commit.assert_not_called() - - def test_update_dataset_api_status_missing_current_user_error(self, mock_dataset_service_dependencies): - """ - Test error handling when current_user is missing. - - Verifies that when current_user is None or has no ID, a ValueError - is raised. - - This test ensures: - - ValueError is raised when current_user is None - - Error message is clear - - No updates are committed - """ - # Arrange - dataset_id = "dataset-123" - dataset = DatasetUpdateDeleteTestDataFactory.create_dataset_mock(dataset_id=dataset_id) - - mock_dataset_service_dependencies["get_dataset"].return_value = dataset - mock_dataset_service_dependencies["current_user"].id = None # Missing user ID - - # Act & Assert - with pytest.raises(ValueError, match="Current user or current user id not found"): - DatasetService.update_dataset_api_status(dataset_id, True) - - # Verify no commit was attempted - mock_dataset_service_dependencies["db_session"].commit.assert_not_called() - - # ============================================================================ # Tests for update_rag_pipeline_dataset_settings # ============================================================================