"""Integration tests for dataset indexing task SQL behaviors using testcontainers.""" import uuid from collections.abc import Sequence from unittest.mock import MagicMock, patch import pytest from faker import Faker from sqlalchemy import select from sqlalchemy.orm import Session from core.indexing_runner import DocumentIsPausedError from core.rag.index_processor.constant.index_type import IndexTechniqueType from enums.cloud_plan import CloudPlan from models import Account, AccountStatus, Tenant, TenantAccountJoin, TenantAccountRole, TenantStatus from models.dataset import Dataset, Document from models.enums import DataSourceType, DocumentCreatedFrom, IndexingStatus from tasks.document_indexing_task import ( _document_indexing, _document_indexing_with_tenant_queue, document_indexing_task, normal_document_indexing_task, priority_document_indexing_task, ) class _TrackedSessionContext: def __init__(self, original_context_manager, opened_sessions: list, closed_sessions: list): self._original_context_manager = original_context_manager self._opened_sessions = opened_sessions self._closed_sessions = closed_sessions self._close_patcher = None self._session = None def __enter__(self): self._session = self._original_context_manager.__enter__() self._opened_sessions.append(self._session) original_close = self._session.close def _tracked_close(*args, **kwargs): self._closed_sessions.append(self._session) return original_close(*args, **kwargs) self._close_patcher = patch.object(self._session, "close", side_effect=_tracked_close, autospec=True) self._close_patcher.start() return self._session def __exit__(self, exc_type, exc_val, exc_tb): try: return self._original_context_manager.__exit__(exc_type, exc_val, exc_tb) finally: if self._close_patcher is not None: self._close_patcher.stop() @pytest.fixture(autouse=True) def _ensure_testcontainers_db(db_session_with_containers: Session): """Ensure this suite always runs on testcontainers infrastructure.""" return db_session_with_containers @pytest.fixture def session_close_tracker(): """Track all sessions opened by session_factory and which were closed.""" opened_sessions = [] closed_sessions = [] from tasks import document_indexing_task as task_module original_create_session = task_module.session_factory.create_session def _tracked_create_session(*args, **kwargs): original_context_manager = original_create_session(*args, **kwargs) return _TrackedSessionContext(original_context_manager, opened_sessions, closed_sessions) with patch.object( task_module.session_factory, "create_session", side_effect=_tracked_create_session, autospec=True ): yield {"opened_sessions": opened_sessions, "closed_sessions": closed_sessions} @pytest.fixture def patched_external_dependencies(): """Patch non-DB collaborators while keeping database behavior real.""" with ( patch("tasks.document_indexing_task.IndexingRunner", autospec=True) as mock_indexing_runner, patch("tasks.document_indexing_task.FeatureService", autospec=True) as mock_feature_service, patch("tasks.document_indexing_task.generate_summary_index_task", autospec=True) as mock_summary_task, ): mock_runner_instance = mock_indexing_runner.return_value mock_features = MagicMock() mock_features.billing.enabled = False mock_features.billing.subscription.plan = CloudPlan.PROFESSIONAL mock_features.vector_space.limit = 100 mock_features.vector_space.size = 0 mock_feature_service.get_features.return_value = mock_features yield { "indexing_runner": mock_indexing_runner, "indexing_runner_instance": mock_runner_instance, "feature_service": mock_feature_service, "features": mock_features, "summary_task": mock_summary_task, } class TestDatasetIndexingTaskIntegration: """1:1 SQL test migration from unit tests to testcontainers integration tests.""" def _create_test_dataset_and_documents( self, db_session_with_containers, *, document_count: int = 3, document_ids: Sequence[str] | None = None, ) -> tuple[Dataset, list[Document]]: """Create a tenant dataset and waiting documents used by indexing tests.""" fake = Faker() account = Account( email=fake.email(), name=fake.name(), interface_language="en-US", status=AccountStatus.ACTIVE, ) db_session_with_containers.add(account) db_session_with_containers.flush() tenant = Tenant(name=fake.company(), status=TenantStatus.NORMAL) db_session_with_containers.add(tenant) db_session_with_containers.flush() join = TenantAccountJoin( tenant_id=tenant.id, account_id=account.id, role=TenantAccountRole.OWNER, current=True, ) db_session_with_containers.add(join) dataset = Dataset( id=fake.uuid4(), tenant_id=tenant.id, name=fake.company(), description=fake.text(max_nb_chars=100), data_source_type=DataSourceType.UPLOAD_FILE, indexing_technique=IndexTechniqueType.HIGH_QUALITY, created_by=account.id, ) db_session_with_containers.add(dataset) if document_ids is None: document_ids = [str(uuid.uuid4()) for _ in range(document_count)] documents = [] for position, document_id in enumerate(document_ids): document = Document( id=document_id, tenant_id=tenant.id, dataset_id=dataset.id, position=position, data_source_type=DataSourceType.UPLOAD_FILE, batch="test_batch", name=f"doc-{position}.txt", created_from=DocumentCreatedFrom.WEB, created_by=account.id, indexing_status=IndexingStatus.WAITING, enabled=True, ) db_session_with_containers.add(document) documents.append(document) db_session_with_containers.commit() db_session_with_containers.refresh(dataset) return dataset, documents def _query_document(self, db_session_with_containers: Session, document_id: str) -> Document | None: """Return the latest persisted document state.""" return db_session_with_containers.scalar(select(Document).where(Document.id == document_id).limit(1)) def _assert_documents_parsing(self, db_session_with_containers: Session, document_ids: Sequence[str]) -> None: """Assert all target documents are persisted in parsing status.""" db_session_with_containers.expire_all() for document_id in document_ids: updated = self._query_document(db_session_with_containers, document_id) assert updated is not None assert updated.indexing_status == IndexingStatus.PARSING assert updated.processing_started_at is not None def _assert_documents_error_contains( self, db_session_with_containers, document_ids: Sequence[str], expected_error_substring: str, ) -> None: """Assert all target documents are persisted in error status with message.""" db_session_with_containers.expire_all() for document_id in document_ids: updated = self._query_document(db_session_with_containers, document_id) assert updated is not None assert updated.indexing_status == IndexingStatus.ERROR assert updated.error is not None assert expected_error_substring in updated.error assert updated.stopped_at is not None def _assert_all_opened_sessions_closed(self, session_close_tracker: dict) -> None: """Assert that every opened session is eventually closed.""" opened = session_close_tracker["opened_sessions"] closed = session_close_tracker["closed_sessions"] opened_ids = {id(session) for session in opened} closed_ids = {id(session) for session in closed} assert len(opened) >= 2 assert opened_ids <= closed_ids def test_legacy_document_indexing_task_still_works( self, db_session_with_containers: Session, patched_external_dependencies ): """Ensure the legacy task entrypoint still updates parsing status.""" # Arrange dataset, documents = self._create_test_dataset_and_documents(db_session_with_containers, document_count=2) document_ids = [doc.id for doc in documents] # Act document_indexing_task(dataset.id, document_ids) # Assert patched_external_dependencies["indexing_runner_instance"].run.assert_called_once() self._assert_documents_parsing(db_session_with_containers, document_ids) def test_batch_processing_multiple_documents( self, db_session_with_containers: Session, patched_external_dependencies ): """Process multiple documents in one batch.""" # Arrange dataset, documents = self._create_test_dataset_and_documents(db_session_with_containers, document_count=3) document_ids = [doc.id for doc in documents] # Act _document_indexing(dataset.id, document_ids) # Assert patched_external_dependencies["indexing_runner_instance"].run.assert_called_once() run_args = patched_external_dependencies["indexing_runner_instance"].run.call_args[0][0] assert len(run_args) == len(document_ids) self._assert_documents_parsing(db_session_with_containers, document_ids) def test_batch_processing_with_limit_check( self, db_session_with_containers: Session, patched_external_dependencies ): """Reject batches larger than configured upload limit. This test patches config only to force a deterministic limit branch while keeping SQL writes real. """ # Arrange dataset, documents = self._create_test_dataset_and_documents(db_session_with_containers, document_count=3) document_ids = [doc.id for doc in documents] features = patched_external_dependencies["features"] features.billing.enabled = True features.billing.subscription.plan = CloudPlan.PROFESSIONAL features.vector_space.limit = 100 features.vector_space.size = 50 # Act with patch("tasks.document_indexing_task.dify_config.BATCH_UPLOAD_LIMIT", "2"): _document_indexing(dataset.id, document_ids) # Assert patched_external_dependencies["indexing_runner_instance"].run.assert_not_called() self._assert_documents_error_contains(db_session_with_containers, document_ids, "batch upload limit") def test_batch_processing_sandbox_plan_single_document_only( self, db_session_with_containers: Session, patched_external_dependencies ): """Reject multi-document upload under sandbox plan.""" # Arrange dataset, documents = self._create_test_dataset_and_documents(db_session_with_containers, document_count=2) document_ids = [doc.id for doc in documents] features = patched_external_dependencies["features"] features.billing.enabled = True features.billing.subscription.plan = CloudPlan.SANDBOX # Act _document_indexing(dataset.id, document_ids) # Assert patched_external_dependencies["indexing_runner_instance"].run.assert_not_called() self._assert_documents_error_contains(db_session_with_containers, document_ids, "does not support batch upload") def test_batch_processing_empty_document_list( self, db_session_with_containers: Session, patched_external_dependencies ): """Handle empty list input without failing.""" # Arrange dataset, _ = self._create_test_dataset_and_documents(db_session_with_containers, document_count=0) # Act _document_indexing(dataset.id, []) # Assert patched_external_dependencies["indexing_runner_instance"].run.assert_called_once_with([]) def test_tenant_queue_dispatches_next_task_after_completion( self, db_session_with_containers: Session, patched_external_dependencies ): """Dispatch the next queued task after current tenant task completes. Queue APIs are patched to isolate dispatch side effects while preserving DB assertions. """ # Arrange dataset, documents = self._create_test_dataset_and_documents(db_session_with_containers, document_count=1) document_ids = [doc.id for doc in documents] next_task = { "tenant_id": dataset.tenant_id, "dataset_id": dataset.id, "document_ids": [str(uuid.uuid4())], } task_dispatch_spy = MagicMock() # Act with ( patch( "tasks.document_indexing_task.TenantIsolatedTaskQueue.pull_tasks", return_value=[next_task], autospec=True, ), patch( "tasks.document_indexing_task.TenantIsolatedTaskQueue.set_task_waiting_time", autospec=True ) as set_waiting_spy, patch( "tasks.document_indexing_task.TenantIsolatedTaskQueue.delete_task_key", autospec=True ) as delete_key_spy, ): _document_indexing_with_tenant_queue(dataset.tenant_id, dataset.id, document_ids, task_dispatch_spy) # Assert # apply_async is used by implementation; assert it was called once with expected kwargs assert task_dispatch_spy.apply_async.call_count == 1 call_kwargs = task_dispatch_spy.apply_async.call_args.kwargs.get("kwargs", {}) assert call_kwargs == { "tenant_id": next_task["tenant_id"], "dataset_id": next_task["dataset_id"], "document_ids": next_task["document_ids"], } set_waiting_spy.assert_called_once() delete_key_spy.assert_not_called() def test_tenant_queue_deletes_running_key_when_no_follow_up_tasks( self, db_session_with_containers: Session, patched_external_dependencies ): """Delete tenant running flag when queue has no pending tasks. Queue APIs are patched to isolate dispatch side effects while preserving DB assertions. """ # Arrange dataset, documents = self._create_test_dataset_and_documents(db_session_with_containers, document_count=1) document_ids = [doc.id for doc in documents] task_dispatch_spy = MagicMock() # Act with ( patch("tasks.document_indexing_task.TenantIsolatedTaskQueue.pull_tasks", return_value=[], autospec=True), patch( "tasks.document_indexing_task.TenantIsolatedTaskQueue.delete_task_key", autospec=True ) as delete_key_spy, ): _document_indexing_with_tenant_queue(dataset.tenant_id, dataset.id, document_ids, task_dispatch_spy) # Assert task_dispatch_spy.apply_async.assert_not_called() delete_key_spy.assert_called_once() def test_validation_failure_sets_error_status_when_vector_space_at_limit( self, db_session_with_containers: Session, patched_external_dependencies ): """Set error status when vector space validation fails before runner phase.""" # Arrange dataset, documents = self._create_test_dataset_and_documents(db_session_with_containers, document_count=3) document_ids = [doc.id for doc in documents] features = patched_external_dependencies["features"] features.billing.enabled = True features.billing.subscription.plan = CloudPlan.PROFESSIONAL features.vector_space.limit = 100 features.vector_space.size = 100 # Act _document_indexing(dataset.id, document_ids) # Assert patched_external_dependencies["indexing_runner_instance"].run.assert_not_called() self._assert_documents_error_contains(db_session_with_containers, document_ids, "over the limit") def test_runner_exception_does_not_crash_indexing_task( self, db_session_with_containers: Session, patched_external_dependencies ): """Catch generic runner exceptions without crashing the task.""" # Arrange dataset, documents = self._create_test_dataset_and_documents(db_session_with_containers, document_count=2) document_ids = [doc.id for doc in documents] patched_external_dependencies["indexing_runner_instance"].run.side_effect = Exception("runner failed") # Act _document_indexing(dataset.id, document_ids) # Assert patched_external_dependencies["indexing_runner_instance"].run.assert_called_once() self._assert_documents_parsing(db_session_with_containers, document_ids) def test_document_paused_error_handling(self, db_session_with_containers: Session, patched_external_dependencies): """Handle DocumentIsPausedError and keep persisted state consistent.""" # Arrange dataset, documents = self._create_test_dataset_and_documents(db_session_with_containers, document_count=2) document_ids = [doc.id for doc in documents] patched_external_dependencies["indexing_runner_instance"].run.side_effect = DocumentIsPausedError("paused") # Act _document_indexing(dataset.id, document_ids) # Assert patched_external_dependencies["indexing_runner_instance"].run.assert_called_once() self._assert_documents_parsing(db_session_with_containers, document_ids) def test_dataset_not_found_error_handling(self, patched_external_dependencies): """Exit gracefully when dataset does not exist.""" # Arrange missing_dataset_id = str(uuid.uuid4()) missing_document_id = str(uuid.uuid4()) # Act _document_indexing(missing_dataset_id, [missing_document_id]) # Assert patched_external_dependencies["indexing_runner_instance"].run.assert_not_called() def test_tenant_queue_error_handling_still_processes_next_task( self, db_session_with_containers: Session, patched_external_dependencies ): """Even on current task failure, enqueue the next waiting tenant task. Queue APIs are patched to isolate dispatch side effects while preserving DB assertions. """ # Arrange dataset, documents = self._create_test_dataset_and_documents(db_session_with_containers, document_count=1) document_ids = [doc.id for doc in documents] next_task = { "tenant_id": dataset.tenant_id, "dataset_id": dataset.id, "document_ids": [str(uuid.uuid4())], } task_dispatch_spy = MagicMock() # Act with ( patch("tasks.document_indexing_task._document_indexing", side_effect=Exception("failed"), autospec=True), patch( "tasks.document_indexing_task.TenantIsolatedTaskQueue.pull_tasks", return_value=[next_task], autospec=True, ), patch("tasks.document_indexing_task.TenantIsolatedTaskQueue.set_task_waiting_time", autospec=True), ): _document_indexing_with_tenant_queue(dataset.tenant_id, dataset.id, document_ids, task_dispatch_spy) # Assert task_dispatch_spy.apply_async.assert_called_once() def test_sessions_close_on_successful_indexing( self, db_session_with_containers, patched_external_dependencies, session_close_tracker, ): """Close all opened sessions in successful indexing path.""" # Arrange dataset, documents = self._create_test_dataset_and_documents(db_session_with_containers, document_count=2) document_ids = [doc.id for doc in documents] # Act _document_indexing(dataset.id, document_ids) # Assert self._assert_all_opened_sessions_closed(session_close_tracker) def test_sessions_close_when_runner_raises( self, db_session_with_containers, patched_external_dependencies, session_close_tracker, ): """Close opened sessions even when runner fails.""" # Arrange dataset, documents = self._create_test_dataset_and_documents(db_session_with_containers, document_count=2) document_ids = [doc.id for doc in documents] patched_external_dependencies["indexing_runner_instance"].run.side_effect = Exception("boom") # Act _document_indexing(dataset.id, document_ids) # Assert self._assert_all_opened_sessions_closed(session_close_tracker) def test_multiple_documents_with_mixed_success_and_failure( self, db_session_with_containers: Session, patched_external_dependencies ): """Process only existing documents when request includes missing ids.""" # Arrange dataset, documents = self._create_test_dataset_and_documents(db_session_with_containers, document_count=2) existing_ids = [doc.id for doc in documents] mixed_ids = [existing_ids[0], str(uuid.uuid4()), existing_ids[1]] # Act _document_indexing(dataset.id, mixed_ids) # Assert run_args = patched_external_dependencies["indexing_runner_instance"].run.call_args[0][0] assert len(run_args) == 2 self._assert_documents_parsing(db_session_with_containers, existing_ids) def test_tenant_queue_dispatches_up_to_concurrency_limit( self, db_session_with_containers: Session, patched_external_dependencies ): """Dispatch only up to configured concurrency under queued backlog burst. Queue APIs are patched to isolate dispatch side effects while preserving DB assertions. """ # Arrange dataset, documents = self._create_test_dataset_and_documents(db_session_with_containers, document_count=1) document_ids = [doc.id for doc in documents] concurrency_limit = 3 backlog_size = 20 pending_tasks = [ {"tenant_id": dataset.tenant_id, "dataset_id": dataset.id, "document_ids": [f"doc_{idx}"]} for idx in range(backlog_size) ] task_dispatch_spy = MagicMock() # Act with ( patch("tasks.document_indexing_task.dify_config.TENANT_ISOLATED_TASK_CONCURRENCY", concurrency_limit), patch( "tasks.document_indexing_task.TenantIsolatedTaskQueue.pull_tasks", return_value=pending_tasks[:concurrency_limit], autospec=True, ), patch( "tasks.document_indexing_task.TenantIsolatedTaskQueue.set_task_waiting_time", autospec=True ) as set_waiting_spy, ): _document_indexing_with_tenant_queue(dataset.tenant_id, dataset.id, document_ids, task_dispatch_spy) # Assert assert task_dispatch_spy.apply_async.call_count == concurrency_limit assert set_waiting_spy.call_count == concurrency_limit def test_task_queue_fifo_ordering(self, db_session_with_containers: Session, patched_external_dependencies): """Keep FIFO ordering when dispatching next queued tasks. Queue APIs are patched to isolate dispatch side effects while preserving DB assertions. """ # Arrange dataset, documents = self._create_test_dataset_and_documents(db_session_with_containers, document_count=1) document_ids = [doc.id for doc in documents] ordered_tasks = [ {"tenant_id": dataset.tenant_id, "dataset_id": dataset.id, "document_ids": ["task_A"]}, {"tenant_id": dataset.tenant_id, "dataset_id": dataset.id, "document_ids": ["task_B"]}, {"tenant_id": dataset.tenant_id, "dataset_id": dataset.id, "document_ids": ["task_C"]}, ] task_dispatch_spy = MagicMock() # Act with ( patch("tasks.document_indexing_task.dify_config.TENANT_ISOLATED_TASK_CONCURRENCY", 3), patch( "tasks.document_indexing_task.TenantIsolatedTaskQueue.pull_tasks", return_value=ordered_tasks, autospec=True, ), patch("tasks.document_indexing_task.TenantIsolatedTaskQueue.set_task_waiting_time", autospec=True), ): _document_indexing_with_tenant_queue(dataset.tenant_id, dataset.id, document_ids, task_dispatch_spy) # Assert assert task_dispatch_spy.apply_async.call_count == 3 for index, expected_task in enumerate(ordered_tasks): call_kwargs = task_dispatch_spy.apply_async.call_args_list[index].kwargs.get("kwargs", {}) assert call_kwargs.get("document_ids") == expected_task["document_ids"] def test_billing_disabled_skips_limit_checks( self, db_session_with_containers: Session, patched_external_dependencies ): """Skip limit checks when billing feature is disabled.""" # Arrange large_document_ids = [str(uuid.uuid4()) for _ in range(100)] dataset, _ = self._create_test_dataset_and_documents( db_session_with_containers, document_ids=large_document_ids, ) features = patched_external_dependencies["features"] features.billing.enabled = False # Act _document_indexing(dataset.id, large_document_ids) # Assert run_args = patched_external_dependencies["indexing_runner_instance"].run.call_args[0][0] assert len(run_args) == 100 self._assert_documents_parsing(db_session_with_containers, large_document_ids) def test_complete_workflow_normal_task(self, db_session_with_containers: Session, patched_external_dependencies): """Run end-to-end normal queue workflow with tenant queue cleanup. Queue APIs are patched to isolate dispatch side effects while preserving DB assertions. """ # Arrange dataset, documents = self._create_test_dataset_and_documents(db_session_with_containers, document_count=2) document_ids = [doc.id for doc in documents] # Act with ( patch("tasks.document_indexing_task.TenantIsolatedTaskQueue.pull_tasks", return_value=[], autospec=True), patch( "tasks.document_indexing_task.TenantIsolatedTaskQueue.delete_task_key", autospec=True ) as delete_key_spy, ): normal_document_indexing_task(dataset.tenant_id, dataset.id, document_ids) # Assert patched_external_dependencies["indexing_runner_instance"].run.assert_called_once() self._assert_documents_parsing(db_session_with_containers, document_ids) delete_key_spy.assert_called_once() def test_complete_workflow_priority_task(self, db_session_with_containers: Session, patched_external_dependencies): """Run end-to-end priority queue workflow with tenant queue cleanup. Queue APIs are patched to isolate dispatch side effects while preserving DB assertions. """ # Arrange dataset, documents = self._create_test_dataset_and_documents(db_session_with_containers, document_count=2) document_ids = [doc.id for doc in documents] # Act with ( patch("tasks.document_indexing_task.TenantIsolatedTaskQueue.pull_tasks", return_value=[], autospec=True), patch( "tasks.document_indexing_task.TenantIsolatedTaskQueue.delete_task_key", autospec=True ) as delete_key_spy, ): priority_document_indexing_task(dataset.tenant_id, dataset.id, document_ids) # Assert patched_external_dependencies["indexing_runner_instance"].run.assert_called_once() self._assert_documents_parsing(db_session_with_containers, document_ids) delete_key_spy.assert_called_once() def test_single_document_processing(self, db_session_with_containers: Session, patched_external_dependencies): """Process the minimum batch size (single document).""" # Arrange dataset, documents = self._create_test_dataset_and_documents(db_session_with_containers, document_count=1) document_id = documents[0].id # Act _document_indexing(dataset.id, [document_id]) # Assert run_args = patched_external_dependencies["indexing_runner_instance"].run.call_args[0][0] assert len(run_args) == 1 self._assert_documents_parsing(db_session_with_containers, [document_id]) def test_document_with_special_characters_in_id( self, db_session_with_containers: Session, patched_external_dependencies ): """Handle standard UUID ids with hyphen characters safely.""" # Arrange special_document_id = str(uuid.uuid4()) dataset, _ = self._create_test_dataset_and_documents( db_session_with_containers, document_ids=[special_document_id], ) # Act _document_indexing(dataset.id, [special_document_id]) # Assert self._assert_documents_parsing(db_session_with_containers, [special_document_id]) def test_zero_vector_space_limit_allows_unlimited( self, db_session_with_containers: Session, patched_external_dependencies ): """Treat vector limit 0 as unlimited and continue indexing.""" # Arrange dataset, documents = self._create_test_dataset_and_documents(db_session_with_containers, document_count=3) document_ids = [doc.id for doc in documents] features = patched_external_dependencies["features"] features.billing.enabled = True features.billing.subscription.plan = CloudPlan.PROFESSIONAL features.vector_space.limit = 0 features.vector_space.size = 1000 # Act _document_indexing(dataset.id, document_ids) # Assert patched_external_dependencies["indexing_runner_instance"].run.assert_called_once() self._assert_documents_parsing(db_session_with_containers, document_ids) def test_negative_vector_space_values_handled_gracefully( self, db_session_with_containers: Session, patched_external_dependencies ): """Treat negative vector limits as non-blocking and continue indexing.""" # Arrange dataset, documents = self._create_test_dataset_and_documents(db_session_with_containers, document_count=3) document_ids = [doc.id for doc in documents] features = patched_external_dependencies["features"] features.billing.enabled = True features.billing.subscription.plan = CloudPlan.PROFESSIONAL features.vector_space.limit = -1 features.vector_space.size = 100 # Act _document_indexing(dataset.id, document_ids) # Assert patched_external_dependencies["indexing_runner_instance"].run.assert_called_once() self._assert_documents_parsing(db_session_with_containers, document_ids) def test_large_document_batch_processing(self, db_session_with_containers: Session, patched_external_dependencies): """Process a batch exactly at configured upload limit. This test patches config only to force a deterministic limit branch while keeping SQL writes real. """ # Arrange batch_limit = 50 document_ids = [str(uuid.uuid4()) for _ in range(batch_limit)] dataset, _ = self._create_test_dataset_and_documents( db_session_with_containers, document_ids=document_ids, ) features = patched_external_dependencies["features"] features.billing.enabled = True features.billing.subscription.plan = CloudPlan.PROFESSIONAL features.vector_space.limit = 10000 features.vector_space.size = 0 # Act with patch("tasks.document_indexing_task.dify_config.BATCH_UPLOAD_LIMIT", str(batch_limit)): _document_indexing(dataset.id, document_ids) # Assert run_args = patched_external_dependencies["indexing_runner_instance"].run.call_args[0][0] assert len(run_args) == batch_limit self._assert_documents_parsing(db_session_with_containers, document_ids)