youngkingdom/dify
1
0
Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-05-01 07:58:02 +08:00
Code Issues Packages Projects Releases Wiki Activity

test: migrate dataset service update-delete SQL tests to testcontainers (#32548)

Browse Source 
Co-authored-by: KinomotoMio <200703522+KinomotoMio@users.noreply.github.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
木之本澪
2026-02-25 14:07:28 +08:00
committed by GitHub
parent 99cc98320a
commit 6ff420cd03
2 changed files with 359 additions and 416 deletions
Download Patch File Download Diff File Expand all files Collapse all files

416
api/tests/unit_tests/services/dataset_service_update_delete.py
View File

@ -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
# ============================================================================