"""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 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, Tenant, TenantAccountJoin, TenantAccountRole
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):
    """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="active",
        )
        db_session_with_containers.add(account)
        db_session_with_containers.flush()

        tenant = Tenant(name=fake.company(), status="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, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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)
