fix(workflow): resolve CI coverage and type failures

api/controllers/service_api/app/legacy_system_files.py

@@ -14,10 +14,10 @@ from services.app_generate_service import AppGenerateService
 def normalize_legacy_system_file_args_for_service_api(
     *,
     app_model: App,
-    args: Mapping[str, Any],
+    args: dict[str, Any],
     raw_payload: Mapping[str, Any] | None,
     workflow_id: str | None = None,
-) -> tuple[Mapping[str, Any], LegacySysFilesCompatVariable | None]:
+) -> tuple[dict[str, Any], LegacySysFilesCompatVariable | None]:
     # TODO: Remove this hidden Service API compatibility path after all persisted workflows are migrated.
     args_with_hidden_system = _copy_hidden_system_files_arg(args=args, raw_payload=raw_payload)
     if not _has_legacy_file_arg(args_with_hidden_system):
@@ -37,9 +37,9 @@ def attach_legacy_system_file_warning_for_service_api(
 
 def _copy_hidden_system_files_arg(
     *,
-    args: Mapping[str, Any],
+    args: dict[str, Any],
     raw_payload: Mapping[str, Any] | None,
-) -> Mapping[str, Any]:
+) -> dict[str, Any]:
     system = raw_payload.get("system") if isinstance(raw_payload, Mapping) else None
     if not isinstance(system, Mapping) or "files" not in system or system["files"] is None:
         return args

api/core/workflow/legacy_system_files.py

@@ -114,7 +114,7 @@ def normalize_legacy_sys_files_args(
     *,
     graph: Mapping[str, Any],
     args: Mapping[str, Any],
-) -> tuple[Mapping[str, Any], LegacySysFilesCompatVariable | None]:
+) -> tuple[dict[str, Any], LegacySysFilesCompatVariable | None]:
     """Map legacy Service/Web API file arguments onto the generated Start-node variable.
 
     The top-level `files` argument is the existing API surface for system files.
@@ -125,7 +125,7 @@ def normalize_legacy_sys_files_args(
     compat_variable = resolve_legacy_sys_files_compat_variable(graph)
     files, legacy_files_used = _extract_legacy_files(args)
     if compat_variable is None or not legacy_files_used:
-        return args, None
+        return dict(args), None
 
     normalized_args = dict(args)
     if "files" not in normalized_args:

api/tests/unit_tests/commands/test_migrate_legacy_sys_files_workflows.py

@@ -1,3 +1,4 @@
+from types import SimpleNamespace
 from unittest.mock import MagicMock
 
 import click
@@ -136,3 +137,99 @@ def test_migrate_legacy_sys_files_workflow_batch_commits_and_counts_failures(cap
     assert "Failed to migrate legacy" in caplog.text
     session.commit.assert_called_once()
     session.rollback.assert_not_called()
+
+
+def test_run_legacy_sys_files_workflow_migration_uses_keyset_batches(mocker):
+    session_maker = MagicMock()
+    sessions = [MagicMock(), MagicMock()]
+    session_maker.side_effect = sessions
+    mocker.patch.object(workflow_migration_commands, "sessionmaker", return_value=session_maker)
+    mocker.patch.object(workflow_migration_commands, "db", SimpleNamespace(engine=object()))
+    migrate_batch = mocker.patch.object(
+        workflow_migration_commands,
+        "_migrate_legacy_sys_files_workflow_batch",
+        side_effect=[
+            workflow_migration_commands.LegacySysFilesWorkflowMigrationStats(
+                scanned=2,
+                migrated=1,
+                failed=0,
+                last_id="workflow-2",
+            ),
+            workflow_migration_commands.LegacySysFilesWorkflowMigrationStats(
+                scanned=1,
+                migrated=1,
+                failed=0,
+                last_id="workflow-3",
+            ),
+        ],
+    )
+
+    stats = workflow_migration_commands.run_legacy_sys_files_workflow_migration(
+        batch_size=2,
+        limit=3,
+        start_after_id="workflow-0",
+        tenant_id="tenant-1",
+        app_id="app-1",
+        dry_run=True,
+    )
+
+    assert stats.scanned == 3
+    assert stats.migrated == 2
+    assert stats.batches == 2
+    assert stats.last_id == "workflow-3"
+    assert migrate_batch.call_args_list[0].kwargs["start_after_id"] == "workflow-0"
+    assert migrate_batch.call_args_list[0].kwargs["batch_size"] == 2
+    assert migrate_batch.call_args_list[1].kwargs["start_after_id"] == "workflow-2"
+    assert migrate_batch.call_args_list[1].kwargs["batch_size"] == 1
+
+
+def test_run_legacy_sys_files_workflow_migration_stops_on_empty_batch(mocker):
+    session_maker = MagicMock(return_value=MagicMock())
+    mocker.patch.object(workflow_migration_commands, "sessionmaker", return_value=session_maker)
+    mocker.patch.object(workflow_migration_commands, "db", SimpleNamespace(engine=object()))
+    mocker.patch.object(
+        workflow_migration_commands,
+        "_migrate_legacy_sys_files_workflow_batch",
+        return_value=workflow_migration_commands.LegacySysFilesWorkflowMigrationStats(scanned=0),
+    )
+
+    stats = workflow_migration_commands.run_legacy_sys_files_workflow_migration(
+        batch_size=2,
+        limit=None,
+        start_after_id=None,
+        tenant_id=None,
+        app_id=None,
+        dry_run=False,
+    )
+
+    assert stats.scanned == 0
+    assert stats.batches == 0
+
+
+def test_run_legacy_sys_files_workflow_migration_stops_on_short_batch(mocker):
+    session_maker = MagicMock(return_value=MagicMock())
+    mocker.patch.object(workflow_migration_commands, "sessionmaker", return_value=session_maker)
+    mocker.patch.object(workflow_migration_commands, "db", SimpleNamespace(engine=object()))
+    migrate_batch = mocker.patch.object(
+        workflow_migration_commands,
+        "_migrate_legacy_sys_files_workflow_batch",
+        return_value=workflow_migration_commands.LegacySysFilesWorkflowMigrationStats(
+            scanned=1,
+            migrated=1,
+            failed=0,
+            last_id="workflow-1",
+        ),
+    )
+
+    stats = workflow_migration_commands.run_legacy_sys_files_workflow_migration(
+        batch_size=2,
+        limit=None,
+        start_after_id=None,
+        tenant_id=None,
+        app_id=None,
+        dry_run=False,
+    )
+
+    assert stats.scanned == 1
+    assert stats.batches == 1
+    migrate_batch.assert_called_once()

api/tests/unit_tests/commands/test_workspace.py

@@ -31,3 +31,19 @@ def test_reset_encrypt_key_pair_rotates_keys_and_removes_custom_provider_data(mo
     assert session.execute.call_count == 2
     captured = capsys.readouterr()
     assert "tenant-1 has been reset" in captured.out
+
+
+def test_reset_encrypt_key_pair_stops_when_workspace_record_is_missing(monkeypatch, capsys):
+    monkeypatch.setattr(workspace_commands.dify_config, "EDITION", "SELF_HOSTED")
+    session = MagicMock()
+    session.scalars.return_value.all.return_value = [None]
+    session_manager = MagicMock()
+    session_manager.begin.return_value.__enter__.return_value = session
+    monkeypatch.setattr(workspace_commands, "sessionmaker", lambda *args, **kwargs: session_manager)
+    monkeypatch.setattr(workspace_commands, "db", MagicMock(engine=object()))
+
+    reset_encrypt_key_pair.callback()
+
+    session.execute.assert_not_called()
+    captured = capsys.readouterr()
+    assert "No workspaces found" in captured.out

api/tests/unit_tests/controllers/service_api/app/test_legacy_system_files.py

@@ -38,6 +38,40 @@ def test_hidden_service_api_file_payload_maps_to_generated_start_input(mocker):
     assert args["inputs"][compat_variable.variable_name] == files
 
 
+def test_service_api_file_payload_is_ignored_when_absent(mocker):
+    get_workflow = mocker.patch.object(AppGenerateService, "get_workflow")
+    app_model = MagicMock()
+    original_args = {"inputs": {}}
+
+    args, compat_variable = normalize_legacy_system_file_args_for_service_api(
+        app_model=app_model,
+        args=original_args,
+        raw_payload={},
+    )
+
+    assert args is original_args
+    assert compat_variable is None
+    get_workflow.assert_not_called()
+
+
+def test_top_level_service_api_file_payload_still_checks_workflow_graph(mocker):
+    workflow = MagicMock()
+    workflow.graph_dict = {"nodes": []}
+    get_workflow = mocker.patch.object(AppGenerateService, "get_workflow", return_value=workflow)
+    app_model = MagicMock()
+    files = [{"id": "file-1"}]
+
+    args, compat_variable = normalize_legacy_system_file_args_for_service_api(
+        app_model=app_model,
+        args={"inputs": {}, "files": files},
+        raw_payload={},
+    )
+
+    get_workflow.assert_called_once()
+    assert args["files"] == files
+    assert compat_variable is None
+
+
 def test_service_api_warning_is_attached_only_when_compatibility_was_used():
     compat_variable = MagicMock(variable_name="generated_files_input")
 

api/tests/unit_tests/core/workflow/test_legacy_system_files.py

@@ -0,0 +1,118 @@
+from core.workflow.legacy_system_files import (
+    LegacySysFilesCompatVariable,
+    attach_legacy_sys_files_warning,
+    migrate_legacy_sys_files_graph_with_result,
+    normalize_legacy_sys_files_args,
+    resolve_legacy_sys_files_compat_variable,
+)
+
+_LEGACY_NODE_ID = "sys"
+_LEGACY_VARIABLE_NAME = "files"
+_LEGACY_SELECTOR = [_LEGACY_NODE_ID, _LEGACY_VARIABLE_NAME]
+_LEGACY_TEMPLATE = "{{#" + ".".join((_LEGACY_NODE_ID, _LEGACY_VARIABLE_NAME)) + "#}}"
+
+
+def test_migrate_legacy_sys_files_graph_ignores_invalid_or_unrelated_graphs():
+    assert not migrate_legacy_sys_files_graph_with_result({}).changed
+    assert not migrate_legacy_sys_files_graph_with_result({"nodes": [], "edges": [_LEGACY_SELECTOR]}).changed
+    assert not migrate_legacy_sys_files_graph_with_result({"nodes": [{"data": {"value": _LEGACY_SELECTOR}}]}).changed
+    assert not migrate_legacy_sys_files_graph_with_result(
+        {"nodes": [{"id": 1, "data": {"type": "start"}}, {"data": {"value": _LEGACY_SELECTOR}}]}
+    ).changed
+
+
+def test_migrate_legacy_sys_files_graph_creates_collision_free_file_input_from_features():
+    graph = {
+        "nodes": [
+            {"id": "start", "data": {"type": "start", "variables": [{"variable": "sys_files"}]}},
+            {"id": "answer", "data": {"type": "answer", "answer": _LEGACY_SELECTOR}},
+        ],
+    }
+
+    result = migrate_legacy_sys_files_graph_with_result(
+        graph,
+        features={
+            "file_upload": {
+                "enabled": True,
+                "allowed_file_upload_methods": ["remote_url"],
+                "allowed_file_types": ["image"],
+                "allowed_file_extensions": [".png"],
+                "number_limits": 9,
+            }
+        },
+    )
+
+    assert result.changed
+    start_data = result.graph["nodes"][0]["data"]
+    created_variable = start_data["variables"][1]
+    assert created_variable["variable"] == "sys_files_1"
+    assert created_variable["allowed_file_upload_methods"] == ["remote_url"]
+    assert created_variable["allowed_file_types"] == ["image"]
+    assert created_variable["allowed_file_extensions"] == [".png"]
+    assert created_variable["max_length"] == 9
+    assert result.graph["nodes"][1]["data"]["answer"] == ["start", "sys_files_1"]
+
+
+def test_resolve_legacy_sys_files_compat_variable_handles_missing_start_variable():
+    assert resolve_legacy_sys_files_compat_variable({}) is None
+    assert resolve_legacy_sys_files_compat_variable({"nodes": [1, {"data": {"value": _LEGACY_SELECTOR}}]}) is None
+    assert (
+        resolve_legacy_sys_files_compat_variable(
+            {"nodes": [{"id": 1, "data": {"type": "start"}}, {"data": {"value": _LEGACY_SELECTOR}}]}
+        )
+        is None
+    )
+    assert (
+        resolve_legacy_sys_files_compat_variable(
+            {"nodes": [{"id": "start", "data": {"type": "start", "variables": []}}]}
+        )
+        is None
+    )
+
+
+def test_normalize_legacy_sys_files_args_handles_no_compat_and_top_level_files():
+    args_without_legacy, compat_without_legacy = normalize_legacy_sys_files_args(
+        graph={"nodes": []},
+        args={"inputs": {}},
+    )
+    assert args_without_legacy == {"inputs": {}}
+    assert compat_without_legacy is None
+
+    files = [{"id": "file-1"}]
+    graph = {
+        "nodes": [
+            {"id": "start", "data": {"type": "start", "variables": []}},
+            {"id": "answer", "data": {"type": "answer", "answer": _LEGACY_TEMPLATE}},
+        ],
+    }
+    normalized_args, compat_variable = normalize_legacy_sys_files_args(
+        graph=graph,
+        args={"inputs": {}, "files": files},
+    )
+
+    assert compat_variable is not None
+    assert normalized_args["files"] == files
+    assert normalized_args["inputs"][compat_variable.variable_name] == files
+
+
+def test_attach_legacy_sys_files_warning_wraps_stream_and_closes_source():
+    class CloseableStream:
+        closed = False
+
+        def __iter__(self):
+            yield "data: payload\n\n"
+
+        def close(self):
+            self.closed = True
+
+    stream = CloseableStream()
+    wrapped = attach_legacy_sys_files_warning(
+        stream,
+        LegacySysFilesCompatVariable(start_node_id="start", variable_name="generated_files_input"),
+    )
+
+    chunks = list(wrapped)
+
+    assert "warning" in chunks[0]
+    assert chunks[1] == "data: payload\n\n"
+    assert stream.closed