From 687e3583e828fa2ffe6d6f15f7cd136f6f60effd Mon Sep 17 00:00:00 2001 From: yetuge <182123929+yetuge@users.noreply.github.com> Date: Fri, 6 Nov 2026 11:06:18 +0000 Subject: [PATCH] test(api): bring three indexing and timeout task tests under strict pyrefly (#45630) Co-authored-by: yetuge --- api/tests/unit_tests/pyrefly.toml | 3 -- .../tasks/test_document_indexing_task.py | 28 ++++++++++++++----- .../test_document_indexing_update_task.py | 18 +++++++----- .../tasks/test_human_input_timeout_tasks.py | 26 ++++++++++------- 4 files changed, 48 insertions(+), 27 deletions(-) diff --git a/api/tests/unit_tests/pyrefly.toml b/api/tests/unit_tests/pyrefly.toml index ca265642bcc..87f4cd83b80 100644 --- a/api/tests/unit_tests/pyrefly.toml +++ b/api/tests/unit_tests/pyrefly.toml @@ -309,9 +309,6 @@ project-excludes = [ "services/test_workflow_draft_variable_service.py", "services/workflow/test_workflow_converter.py", "services/test_workflow_event_snapshot_service.py", - "tasks/test_document_indexing_task.py", - "tasks/test_document_indexing_update_task.py", - "tasks/test_human_input_timeout_tasks.py", "tasks/app_generate/test_workflow_execute_task.py", "../e2e_tests/contracts/test_pytest_dify.py", ] diff --git a/api/tests/unit_tests/tasks/test_document_indexing_task.py b/api/tests/unit_tests/tasks/test_document_indexing_task.py index 49edd1d96b5..e776fb88de4 100644 --- a/api/tests/unit_tests/tasks/test_document_indexing_task.py +++ b/api/tests/unit_tests/tasks/test_document_indexing_task.py @@ -44,10 +44,16 @@ from tests.helpers.unit.config_override import apply_config_overrides, config_ov ], ) def test_recovery_dispatches_detached_owner_reference( - sqlite_session, tenant_id, dataset_id, document_ids, status, entry -): + sqlite_session: Session, + tenant_id: str, + dataset_id: str, + document_ids: list[str], + status: IndexingStatus, + entry: str | None, +) -> None: _persist_indexing_rows(sqlite_session, tenant_id=tenant_id, dataset_id=dataset_id, document_ids=document_ids[:1]) row = sqlite_session.get(Document, document_ids[0]) + assert row is not None row.indexing_status = status sqlite_session.commit() with patch("tasks.recover_document_indexing_task.build_document_indexing_service") as factory: @@ -90,6 +96,7 @@ def document_ids() -> list[str]: @pytest.fixture def mock_redis() -> MagicMock: """Reset the external Redis boundary used by tenant-isolated queues.""" + assert isinstance(redis_client, MagicMock) redis_client.reset_mock() redis_client.get.return_value = None redis_client.setex.return_value = True @@ -257,7 +264,8 @@ class TestDocumentIndexing: def assert_committed_parsing(refs: list[DocumentRef]) -> None: with Session(sqlite_session.get_bind()) as observer: - documents = [observer.get(Document, ref.document_id) for ref in refs] + fetched = [observer.get(Document, ref.document_id) for ref in refs] + documents = [document for document in fetched if document is not None] assert all(document.indexing_status == IndexingStatus.PARSING for document in documents) assert all(document.processing_started_at is not None for document in documents) @@ -437,7 +445,9 @@ class TestSummaryDispatch: for ref, status in zip( refs, [IndexingStatus.COMPLETED, IndexingStatus.COMPLETED, IndexingStatus.INDEXING], strict=True ): - writer.get(Document, ref.document_id).indexing_status = status + document = writer.get(Document, ref.document_id) + assert document is not None + document.indexing_status = status indexing_runner.run.side_effect = finish_documents summary_delay = MagicMock() @@ -468,7 +478,9 @@ class TestSummaryDispatch: def complete_documents(refs: list[DocumentRef]) -> None: with Session(sqlite_session.get_bind()) as writer, writer.begin(): - writer.get(Document, refs[0].document_id).indexing_status = IndexingStatus.COMPLETED + document = writer.get(Document, refs[0].document_id) + assert document is not None + document.indexing_status = IndexingStatus.COMPLETED indexing_runner.run.side_effect = complete_documents summary_delay = MagicMock(side_effect=RuntimeError("queue unavailable")) @@ -502,7 +514,9 @@ class TestSummaryDispatch: def complete_documents(refs: list[DocumentRef]) -> None: with Session(sqlite_session.get_bind()) as writer, writer.begin(): - writer.get(Document, refs[0].document_id).indexing_status = IndexingStatus.COMPLETED + document = writer.get(Document, refs[0].document_id) + assert document is not None + document.indexing_status = IndexingStatus.COMPLETED indexing_runner.run.side_effect = complete_documents summary_delay = MagicMock() @@ -579,7 +593,7 @@ class TestTenantQueue: self, tenant_id: str, dataset_id: str, document_ids: list[str], monkeypatch: pytest.MonkeyPatch ) -> None: queue = MagicMock() - queue.pull_tasks.return_value = [] + queue.pull_tasks.return_value = list[str]() monkeypatch.setattr("tasks.document_indexing_task.TenantIsolatedTaskQueue", MagicMock(return_value=queue)) indexing = MagicMock(side_effect=RuntimeError("indexing failed")) monkeypatch.setattr("tasks.document_indexing_task._document_indexing", indexing) diff --git a/api/tests/unit_tests/tasks/test_document_indexing_update_task.py b/api/tests/unit_tests/tasks/test_document_indexing_update_task.py index c38b3440b64..144925b2169 100644 --- a/api/tests/unit_tests/tasks/test_document_indexing_update_task.py +++ b/api/tests/unit_tests/tasks/test_document_indexing_update_task.py @@ -10,6 +10,7 @@ from unittest.mock import MagicMock import pytest from sqlalchemy import select +from sqlalchemy.engine import Connection, Engine from sqlalchemy.orm import Session import tasks.document_indexing_update_task as task_module @@ -19,6 +20,7 @@ from models.dataset import Dataset, Document, DocumentSegment, SegmentAttachment from models.enums import CreatorUserRole, DataSourceType, DocumentCreatedFrom, IndexingStatus from models.model import UploadFile from services.knowledge.indexing.errors import DocumentIsPausedError +from services.knowledge.resource_scope import DocumentRef from tasks.document_indexing_update_task import document_indexing_update_task @@ -49,7 +51,7 @@ def _persist_rows( session: Session, *, indexing_technique: IndexTechniqueType = IndexTechniqueType.HIGH_QUALITY, - summary_index_setting: dict | None = None, + summary_index_setting: dict[str, object] | None = None, doc_form: IndexStructureType = IndexStructureType.PARAGRAPH_INDEX, need_summary: bool = True, with_segment: bool = False, @@ -101,10 +103,12 @@ def _persist_rows( return dataset, document -def _complete_indexing(refs, *, bind) -> None: +def _complete_indexing(refs: list[DocumentRef], *, bind: Engine | Connection) -> None: with Session(bind) as writer, writer.begin(): for ref in refs: - writer.get(Document, ref.document_id).indexing_status = IndexingStatus.COMPLETED + document = writer.get(Document, ref.document_id) + assert document is not None + document.indexing_status = IndexingStatus.COMPLETED def _persist_attachment( @@ -175,8 +179,8 @@ def test_skips_summary_when_persisted_eligibility_does_not_match( sqlite_session: Session, task_harness: tuple[MagicMock, MagicMock], monkeypatch: pytest.MonkeyPatch, - dataset_changes: dict, - document_changes: dict, + dataset_changes: dict[str, object], + document_changes: dict[str, object], ) -> None: runner, _processor = task_harness dataset, document = _persist_rows( @@ -240,8 +244,8 @@ def test_skips_summary_when_dataset_is_removed_after_indexing( runner, _processor = task_harness dataset, document = _persist_rows(sqlite_session, summary_index_setting={"enable": True}) - def complete_and_remove(documents: list[Document], session: Session) -> None: - _complete_indexing(documents, session) + def complete_and_remove(documents: list[DocumentRef], session: Session) -> None: + _complete_indexing(documents, bind=session.get_bind()) persisted_dataset = session.get(Dataset, dataset.id) assert persisted_dataset is not None session.delete(persisted_dataset) diff --git a/api/tests/unit_tests/tasks/test_human_input_timeout_tasks.py b/api/tests/unit_tests/tasks/test_human_input_timeout_tasks.py index d24c0f24c56..83556b20d02 100644 --- a/api/tests/unit_tests/tasks/test_human_input_timeout_tasks.py +++ b/api/tests/unit_tests/tasks/test_human_input_timeout_tasks.py @@ -18,7 +18,7 @@ from tests.helpers.unit.config_override import apply_config_overrides class _FakeService: - def __init__(self): + def __init__(self) -> None: self.enqueued: list[str] = [] self.agent_app_resumed: list[tuple[str, str]] = [] @@ -61,6 +61,12 @@ def _build_form( ) +def _require_form(session: Session, form_id: str) -> HumanInputForm: + form = session.get(HumanInputForm, form_id) + assert form is not None + return form + + @pytest.fixture def sqlite_task_database( sqlite_engine: Engine, @@ -72,7 +78,7 @@ def sqlite_task_database( monkeypatch.setattr(task_module, "db", SimpleNamespace(engine=sqlite_engine)) -def test_is_global_timeout_uses_created_at(): +def test_is_global_timeout_uses_created_at() -> None: now = datetime(2025, 1, 1, 12, 0, 0) form = _build_form( form_id="form-1", @@ -101,7 +107,7 @@ def test_check_and_handle_human_input_timeouts_marks_and_routes( sqlite_task_database: None, sqlite_engine: Engine, sqlite_session: Session, -): +) -> None: now = datetime(2025, 1, 1, 12, 0, 0) monkeypatch.setattr(task_module, "naive_utc_now", lambda: now) apply_config_overrides(monkeypatch, HUMAN_INPUT_GLOBAL_TIMEOUT_SECONDS=3600) @@ -168,9 +174,9 @@ def test_check_and_handle_human_input_timeouts_marks_and_routes( service_factory.assert_called_once_with(task_session_maker, form_repository=repo) sqlite_session.expire_all() - assert sqlite_session.get(HumanInputForm, "form-global").status == HumanInputFormStatus.EXPIRED - assert sqlite_session.get(HumanInputForm, "form-node").status == HumanInputFormStatus.TIMEOUT - assert sqlite_session.get(HumanInputForm, "form-delivery").status == HumanInputFormStatus.TIMEOUT + assert _require_form(sqlite_session, "form-global").status == HumanInputFormStatus.EXPIRED + assert _require_form(sqlite_session, "form-node").status == HumanInputFormStatus.TIMEOUT + assert _require_form(sqlite_session, "form-delivery").status == HumanInputFormStatus.TIMEOUT @pytest.mark.parametrize("sqlite_session", [(HumanInputForm,)], indirect=True) @@ -178,7 +184,7 @@ def test_check_and_handle_human_input_timeouts_orders_by_id_before_limit( monkeypatch: pytest.MonkeyPatch, sqlite_task_database: None, sqlite_session: Session, -): +) -> None: now = datetime(2025, 1, 1, 12, 0, 0) monkeypatch.setattr(task_module, "naive_utc_now", lambda: now) apply_config_overrides(monkeypatch, HUMAN_INPUT_GLOBAL_TIMEOUT_SECONDS=0) @@ -211,8 +217,8 @@ def test_check_and_handle_human_input_timeouts_orders_by_id_before_limit( reason="delivery_test_timeout", ) sqlite_session.expire_all() - assert sqlite_session.get(HumanInputForm, "form-a").status == HumanInputFormStatus.TIMEOUT - assert sqlite_session.get(HumanInputForm, "form-b").status == HumanInputFormStatus.WAITING + assert _require_form(sqlite_session, "form-a").status == HumanInputFormStatus.TIMEOUT + assert _require_form(sqlite_session, "form-b").status == HumanInputFormStatus.WAITING @pytest.mark.parametrize("sqlite_session", [(HumanInputForm,)], indirect=True) @@ -220,7 +226,7 @@ def test_check_and_handle_human_input_timeouts_omits_global_filter_when_disabled monkeypatch: pytest.MonkeyPatch, sqlite_task_database: None, sqlite_session: Session, -): +) -> None: now = datetime(2025, 1, 1, 12, 0, 0) monkeypatch.setattr(task_module, "naive_utc_now", lambda: now) apply_config_overrides(monkeypatch, HUMAN_INPUT_GLOBAL_TIMEOUT_SECONDS=0)