fix migration resume and release workflow

backend/tests/api/test_git_repo_lifecycle_routes.py

@@ -314,4 +314,18 @@ class TestDeployDashboard:
             client = _make_client()
             resp = client.post("/repositories/bad-ref/deploy", json={"stage": "prod"})
         assert resp.status_code == 404
+
+    def test_prod_requires_named_release(self, mock_db_repo):
+        """The legacy deploy endpoint cannot bypass the named-release publication gate."""
+        mock_db = MagicMock()
+        mock_db.query.return_value.filter.return_value.first.return_value = mock_db_repo
+        from src.core.database import get_db
+        with (
+            patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)),
+            patch("src.api.routes.git._repo_lifecycle_routes._resolve_stage_environment", return_value=self._target_environment()),
+        ):
+            client = _make_client({get_db: lambda: mock_db})
+            resp = client.post("/repositories/42/deploy", json={"stage": "prod"})
+        assert resp.status_code == 409
+        assert "named dashboard release" in resp.json()["detail"]
 # #endregion Test.Api.GitRepoLifecycleRoutes

backend/tests/core/task_manager/test_graph.py

@@ -182,10 +182,11 @@ class TestCreateFuture:
 class TestResolveFuture:
     def test_resolves(self, graph):
         future = MagicMock()
+        future.done.return_value = False
         graph.task_futures["t1"] = future
         graph.resolve_future("t1", result=True)
         future.set_result.assert_called_once_with(True)
-        assert "t1" not in graph.task_futures
+        assert graph.task_futures["t1"] is future
 
     def test_resolve_nonexistent(self, graph):
         graph.resolve_future("nonexistent")  # should not raise

backend/tests/core/task_manager/test_lifecycle.py

@@ -40,6 +40,16 @@ def mock_graph():
     graph = MagicMock()
     graph.get_task.return_value = None  # default: not found
     graph.tasks = {}
+    graph.task_futures = {}
+
+    def create_future(task_id, future):
+        graph.task_futures[task_id] = future
+
+    def remove_future(task_id):
+        graph.task_futures.pop(task_id, None)
+
+    graph.create_future.side_effect = create_future
+    graph.remove_future.side_effect = remove_future
     return graph
 
 

backend/tests/core/test_migration_engine.py

@@ -331,6 +331,53 @@ def test_transform_zip_end_to_end():
 # #endregion test_transform_zip_end_to_end
 
 
+# #region test_transform_zip_keeps_only_referenced_mapped_database [C:2] [TYPE Function]
+# @BRIEF An exported archive must not import unused database resources or request their passwords.
+# @TEST_EDGE unreferenced_database_with_password -> excluded from transformed archive.
+def test_transform_zip_keeps_only_referenced_mapped_database():
+    engine = MigrationEngine()
+    with tempfile.TemporaryDirectory() as td:
+        root = Path(td)
+        source = root / "source.zip"
+        target = root / "target.zip"
+        archive = root / "archive"
+        (archive / "datasets").mkdir(parents=True)
+        (archive / "databases").mkdir()
+        (archive / "datasets" / "orders.yaml").write_text(
+            "database_uuid: source-used\ntable_name: orders\n"
+        )
+        (archive / "databases" / "used.yaml").write_text("uuid: source-used\n")
+        (archive / "databases" / "unrelated.yaml").write_text("uuid: source-unrelated\n")
+        with zipfile.ZipFile(source, "w") as zf:
+            for path in archive.rglob("*.yaml"):
+                zf.write(path, path.relative_to(archive))
+
+        assert engine.transform_zip(
+            str(source), str(target), {"source-used": "target-used"},
+            strip_databases=False,
+        )
+        with zipfile.ZipFile(target) as zf:
+            assert "databases/used.yaml" in zf.namelist()
+            assert "databases/unrelated.yaml" not in zf.namelist()
+            dataset = yaml.safe_load(zf.read("datasets/orders.yaml"))
+            assert dataset["database_uuid"] == "target-used"
+
+
+def test_transform_zip_rejects_dataset_without_database_mapping():
+    """@TEST_EDGE An unmapped dataset dependency cannot fall through to a password prompt."""
+    engine = MigrationEngine()
+    with tempfile.TemporaryDirectory() as td:
+        source = Path(td) / "source.zip"
+        target = Path(td) / "target.zip"
+        with zipfile.ZipFile(source, "w") as zf:
+            zf.writestr("datasets/orders.yaml", "database_uuid: source-unmapped\n")
+            zf.writestr("databases/unmapped.yaml", "uuid: source-unmapped\n")
+
+        assert not engine.transform_zip(str(source), str(target), {}, strip_databases=False)
+        assert not target.exists()
+# #endregion test_transform_zip_keeps_only_referenced_mapped_database
+
+
 # #region test_transform_zip_invalid_path [C:2] [TYPE Function]
 # @RELATION  BINDS_TO -> [TestMigrationEngine]
 # @PURPOSE: Verify transform_zip returns False when source archive path does not exist.

backend/tests/models/test_dashboard_release_model.py

@@ -0,0 +1,34 @@
+# #region Test.DashboardReleaseModel [C:3] [TYPE Module] [SEMANTICS test,git,release,model]
+# @BRIEF Verify immutable dashboard release ledger fields.
+# @RELATION BINDS_TO -> [DashboardRelease]
+# @TEST_CONTRACT: Valid release input -> immutable repository-scoped release record.
+# @TEST_EDGE: duplicate_version -> database unique constraint owns rejection.
+# @TEST_EDGE: missing_notes -> API schema rejects before persistence.
+# @TEST_EDGE: changed_preprod -> release route marks the active release superseded.
+# @TEST_INVARIANT: SourceHashesImmutable -> VERIFIED_BY: stores_pinned_candidate_hashes.
+
+from src.models.dashboard_release import DashboardRelease
+
+
+class TestDashboardRelease:
+    # #region test_stores_pinned_candidate_hashes [C:2] [TYPE Function]
+    # @BRIEF A release retains the exact commit and semantic content hash selected in PREPROD.
+    def test_stores_pinned_candidate_hashes(self):
+        release = DashboardRelease(
+            repository_id="repo-1",
+            deployment_id=17,
+            name="July dashboard",
+            version="2026.07.16",
+            notes="Новый фильтр региона",
+            commit_hash="a" * 40,
+            content_hash="b" * 64,
+            status="awaiting_approval",
+            created_by="analyst",
+        )
+        assert release.commit_hash == "a" * 40
+        assert release.content_hash == "b" * 64
+        assert release.status == "awaiting_approval"
+    # #endregion test_stores_pinned_candidate_hashes
+
+
+# #endregion Test.DashboardReleaseModel

backend/tests/plugins/test_migration_plugin_password.py

@@ -0,0 +1,795 @@
+#region Test.MigrationPlugin.Password [C:3] [TYPE Module] [SEMANTICS test,migration,password,await_input,regression]
+# @BRIEF Regression & integration tests for password injection flow during dashboard migration.
+# @RELATION BINDS_TO -> [MigrationPlugin]
+# @TEST_EDGE: await_input_no_add_log_callback -> Regression: await_input called without add_log_callback kwarg
+# @TEST_EDGE: password_injection_full_flow -> await_input payload verified, wait_for_input called, retry with passwords
+# @TEST_EDGE: password_error_unknown_pattern -> Falls back to db_name="unknown" when no regex matches
+# @TEST_EDGE: multiple_dashboards_partial_password -> Some dashboards succeed, some fail on password
+import pytest
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from src.plugins.migration import MigrationPlugin
+
+
+# ── Helpers (mirror from test_migration_plugin.py) ──
+
+
+def _make_env(id_val="env-1", name="Source Env"):
+    env = MagicMock()
+    env.id = id_val
+    env.name = name
+    env.url = "https://superset.example.com"
+    env.username = "admin"
+    env.password = "secret"
+    return env
+
+
+def _make_dashboard(dash_id=1, title="Test Dash"):
+    return {"id": dash_id, "slug": f"slug-{dash_id}", "dashboard_title": title}
+
+
+def _make_mock_superset_client():
+    client = MagicMock()
+    client.aclose = AsyncMock()
+    return client
+
+
+def _make_mock_mapping_service():
+    svc = MagicMock()
+    svc.sync_environment = AsyncMock()
+    return svc
+
+
+def _make_mock_ctf(path="/tmp/test.zip"):
+    mock_ctf = MagicMock()
+    mock_ctf.return_value.__enter__ = MagicMock(return_value=path)
+    return mock_ctf
+
+
+# ══════════════════════════════════════════════════════════════════════════════
+# CORE REGRESSION: await_input called without add_log_callback
+# ══════════════════════════════════════════════════════════════════════════════
+
+
+class TestPasswordInjectionRegression:
+    """Verify the await_input call signature — the bug from production logs."""
+
+    # #region test_await_input_no_add_log_callback [C:2] [TYPE Function]
+    # @BRIEF Regression: tm.await_input() must NOT receive add_log_callback kwarg.
+    @pytest.mark.asyncio
+    async def test_await_input_no_add_log_callback(self):
+        """
+        Production bug (2026-07-16): migration plugin passed add_log_callback=add_log
+        to tm.await_input(), which raised TypeError because manager.await_input()
+        only accepts (task_id, input_request). Manager handles add_log_callback
+        internally via self._add_log.
+        """
+        plugin = MigrationPlugin()
+        src_env = _make_env("env-1", "Source")
+        tgt_env = _make_env("env-2", "Target")
+
+        mock_cm = MagicMock()
+        mock_cm.get_environments.return_value = [src_env, tgt_env]
+
+        mock_task_manager = MagicMock()
+        mock_task_manager.get_task.return_value = MagicMock(
+            params={"passwords": {"ClickHouse": "s3cret"}}
+        )
+        mock_task_manager.await_input = AsyncMock()
+        mock_task_manager.wait_for_input = AsyncMock()
+
+        mock_src_client = _make_mock_superset_client()
+        mock_src_client.get_dashboards = AsyncMock(
+            return_value=(True, [_make_dashboard(1, "Dash")])
+        )
+        mock_src_client.export_dashboard = AsyncMock(return_value=(b"zip", "meta"))
+        mock_tgt_client = _make_mock_superset_client()
+        mock_tgt_client.import_dashboard = AsyncMock(
+            side_effect=[
+                RuntimeError(
+                    "Must provide a password for the database databases/Dev_Clickhouse_Node_1.yaml"
+                ),
+                None,
+            ]
+        )
+
+        mock_engine = MagicMock()
+        mock_engine.transform_zip.return_value = True
+
+        with patch("src.plugins.migration.get_config_manager", return_value=mock_cm), \
+             patch("src.plugins.migration.SupersetClient") as MockSC, \
+             patch("src.plugins.migration.MigrationEngine", return_value=mock_engine), \
+             patch("src.plugins.migration.create_temp_file", return_value=_make_mock_ctf()), \
+             patch("src.dependencies.get_task_manager", return_value=mock_task_manager), \
+             patch("src.plugins.migration.IdMappingService", return_value=_make_mock_mapping_service()), \
+             patch("src.plugins.migration.SessionLocal"):
+
+            MockSC.side_effect = [mock_src_client, mock_tgt_client]
+
+            result = await plugin.execute({
+                "source_env_id": "env-1",
+                "target_env_id": "env-2",
+                "selected_ids": [1],
+                "replace_db_config": False,
+                "_task_id": "task-regress-1",
+            })
+
+            assert result["status"] == "SUCCESS"
+
+            # KEY ASSERTION: await_input was called WITHOUT add_log_callback
+            mock_task_manager.await_input.assert_called_once_with(
+                "task-regress-1",
+                {
+                    "type": "database_password",
+                    "databases": ["Dev_Clickhouse_Node_1"],
+                    "error_message": "A database password is required to continue this migration.",
+                },
+            )
+    # #endregion test_await_input_no_add_log_callback
+
+    # #region test_await_input_called_without_extra_kwargs [C:2] [TYPE Function]
+    # @BRIEF Regression: await_input call must not include any kwargs beyond task_id + input_request.
+    @pytest.mark.asyncio
+    async def test_await_input_called_without_extra_kwargs(self):
+        """Explicitly verify no extra keyword arguments leak into await_input call."""
+        plugin = MigrationPlugin()
+        src_env = _make_env("env-1", "Source")
+        tgt_env = _make_env("env-2", "Target")
+
+        mock_cm = MagicMock()
+        mock_cm.get_environments.return_value = [src_env, tgt_env]
+
+        mock_task_manager = MagicMock()
+        mock_task_manager.get_task.return_value = MagicMock(
+            params={"passwords": {"PG": "pwd"}}
+        )
+        mock_task_manager.await_input = AsyncMock()
+        mock_task_manager.wait_for_input = AsyncMock()
+
+        mock_src_client = _make_mock_superset_client()
+        mock_src_client.get_dashboards = AsyncMock(
+            return_value=(True, [_make_dashboard(1, "Dash")])
+        )
+        mock_src_client.export_dashboard = AsyncMock(return_value=(b"zip", "meta"))
+        mock_tgt_client = _make_mock_superset_client()
+        mock_tgt_client.import_dashboard = AsyncMock(
+            side_effect=[
+                RuntimeError("Must provide a password for the database 'PostgreSQL'"),
+                None,
+            ]
+        )
+
+        mock_engine = MagicMock()
+        mock_engine.transform_zip.return_value = True
+
+        with patch("src.plugins.migration.get_config_manager", return_value=mock_cm), \
+             patch("src.plugins.migration.SupersetClient") as MockSC, \
+             patch("src.plugins.migration.MigrationEngine", return_value=mock_engine), \
+             patch("src.plugins.migration.create_temp_file", return_value=_make_mock_ctf()), \
+             patch("src.dependencies.get_task_manager", return_value=mock_task_manager), \
+             patch("src.plugins.migration.IdMappingService", return_value=_make_mock_mapping_service()), \
+             patch("src.plugins.migration.SessionLocal"):
+
+            MockSC.side_effect = [mock_src_client, mock_tgt_client]
+
+            await plugin.execute({
+                "source_env_id": "env-1",
+                "target_env_id": "env-2",
+                "selected_ids": [1],
+                "replace_db_config": False,
+                "_task_id": "task-extra-kwargs-1",
+            })
+
+            # Verify call signature: exactly 2 positional args, no extra kwargs
+            call_args = mock_task_manager.await_input.call_args
+            assert len(call_args[0]) == 2  # task_id, input_request (positional)
+            assert call_args[1] == {}  # no keyword arguments
+            assert call_args[0][0] == "task-extra-kwargs-1"
+            assert call_args[0][1]["type"] == "database_password"
+    # #endregion test_await_input_called_without_extra_kwargs
+
+
+# ══════════════════════════════════════════════════════════════════════════════
+# FULL FLOW: password injection end-to-end verification
+# ══════════════════════════════════════════════════════════════════════════════
+
+
+class TestPasswordInjectionFullFlow:
+    """Verify the complete password injection lifecycle."""
+
+    # #region test_full_flow_wait_for_input_called [C:2] [TYPE Function]
+    # @BRIEF After await_input, wait_for_input is called before reading passwords.
+    @pytest.mark.asyncio
+    async def test_full_flow_wait_for_input_called(self):
+        """Verify wait_for_input is called after await_input, before get_task."""
+        plugin = MigrationPlugin()
+        src_env = _make_env("env-1", "Source")
+        tgt_env = _make_env("env-2", "Target")
+
+        mock_cm = MagicMock()
+        mock_cm.get_environments.return_value = [src_env, tgt_env]
+
+        call_order = []
+
+        mock_task_manager = MagicMock()
+        mock_task_manager.get_task.return_value = MagicMock(
+            params={"passwords": {"ClickHouse": "secret"}}
+        )
+
+        async def _await_input(task_id, input_request):
+            call_order.append("await_input")
+
+        async def _wait_for_input(task_id):
+            call_order.append("wait_for_input")
+
+        mock_task_manager.await_input = _await_input
+        mock_task_manager.wait_for_input = _wait_for_input
+
+        mock_src_client = _make_mock_superset_client()
+        mock_src_client.get_dashboards = AsyncMock(
+            return_value=(True, [_make_dashboard(1, "Dash")])
+        )
+        mock_src_client.export_dashboard = AsyncMock(return_value=(b"zip", "meta"))
+        mock_tgt_client = _make_mock_superset_client()
+        mock_tgt_client.import_dashboard = AsyncMock(
+            side_effect=[
+                RuntimeError("Must provide a password for the database 'ClickHouse'"),
+                None,
+            ]
+        )
+
+        mock_engine = MagicMock()
+        mock_engine.transform_zip.return_value = True
+
+        with patch("src.plugins.migration.get_config_manager", return_value=mock_cm), \
+             patch("src.plugins.migration.SupersetClient") as MockSC, \
+             patch("src.plugins.migration.MigrationEngine", return_value=mock_engine), \
+             patch("src.plugins.migration.create_temp_file", return_value=_make_mock_ctf()), \
+             patch("src.dependencies.get_task_manager", return_value=mock_task_manager), \
+             patch("src.plugins.migration.IdMappingService", return_value=_make_mock_mapping_service()), \
+             patch("src.plugins.migration.SessionLocal"):
+
+            MockSC.side_effect = [mock_src_client, mock_tgt_client]
+
+            result = await plugin.execute({
+                "source_env_id": "env-1",
+                "target_env_id": "env-2",
+                "selected_ids": [1],
+                "replace_db_config": False,
+                "_task_id": "task-order-1",
+            })
+
+            assert result["status"] == "SUCCESS"
+            # Call order must be: await_input → wait_for_input → get_task
+            assert call_order == ["await_input", "wait_for_input"]
+    # #endregion test_full_flow_wait_for_input_called
+
+    # #region test_full_flow_import_retry_with_passwords [C:2] [TYPE Function]
+    # @BRIEF After password injection, import is retried with passwords parameter.
+    @pytest.mark.asyncio
+    async def test_full_flow_import_retry_with_passwords(self):
+        """Verify the retry import_dashboard call includes the passwords kwarg."""
+        plugin = MigrationPlugin()
+        src_env = _make_env("env-1", "Source")
+        tgt_env = _make_env("env-2", "Target")
+
+        mock_cm = MagicMock()
+        mock_cm.get_environments.return_value = [src_env, tgt_env]
+
+        mock_task_manager = MagicMock()
+        mock_task_manager.get_task.return_value = MagicMock(
+            params={"passwords": {"Dev_Clickhouse": "p@ssw0rd"}}
+        )
+        mock_task_manager.await_input = AsyncMock()
+        mock_task_manager.wait_for_input = AsyncMock()
+
+        mock_src_client = _make_mock_superset_client()
+        mock_src_client.get_dashboards = AsyncMock(
+            return_value=(True, [_make_dashboard(1, "Dash")])
+        )
+        mock_src_client.export_dashboard = AsyncMock(return_value=(b"zip", "meta"))
+        mock_tgt_client = _make_mock_superset_client()
+        mock_tgt_client.import_dashboard = AsyncMock(
+            side_effect=[
+                RuntimeError("Must provide a password for the database 'Dev_Clickhouse'"),
+                None,
+            ]
+        )
+
+        mock_engine = MagicMock()
+        mock_engine.transform_zip.return_value = True
+
+        with patch("src.plugins.migration.get_config_manager", return_value=mock_cm), \
+             patch("src.plugins.migration.SupersetClient") as MockSC, \
+             patch("src.plugins.migration.MigrationEngine", return_value=mock_engine), \
+             patch("src.plugins.migration.create_temp_file", return_value=_make_mock_ctf()), \
+             patch("src.dependencies.get_task_manager", return_value=mock_task_manager), \
+             patch("src.plugins.migration.IdMappingService", return_value=_make_mock_mapping_service()), \
+             patch("src.plugins.migration.SessionLocal"):
+
+            MockSC.side_effect = [mock_src_client, mock_tgt_client]
+
+            await plugin.execute({
+                "source_env_id": "env-1",
+                "target_env_id": "env-2",
+                "selected_ids": [1],
+                "replace_db_config": False,
+                "_task_id": "task-retry-1",
+            })
+
+            # Verify import_dashboard was called twice (fail + retry)
+            assert mock_tgt_client.import_dashboard.call_count == 2
+
+            # Second call (retry) must include passwords kwarg
+            retry_kwargs = mock_tgt_client.import_dashboard.call_args_list[1][1]
+            assert "passwords" in retry_kwargs
+            assert retry_kwargs["passwords"] == {"Dev_Clickhouse": "p@ssw0rd"}
+    # #endregion test_full_flow_import_retry_with_passwords
+
+    # #region test_full_flow_passwords_cleaned_after_retry [C:2] [TYPE Function]
+    # @BRIEF Password params must be deleted from task after successful retry.
+    @pytest.mark.asyncio
+    async def test_full_flow_passwords_cleaned_after_retry(self):
+        """Verify passwords are popped from task.params after retry (security)."""
+        plugin = MigrationPlugin()
+        src_env = _make_env("env-1", "Source")
+        tgt_env = _make_env("env-2", "Target")
+
+        mock_cm = MagicMock()
+        mock_cm.get_environments.return_value = [src_env, tgt_env]
+
+        task_mock_params = {"passwords": {"PG": "secret123"}}
+
+        mock_task_manager = MagicMock()
+        mock_task_manager.get_task.return_value = MagicMock(
+            params=task_mock_params
+        )
+        mock_task_manager.await_input = AsyncMock()
+        mock_task_manager.wait_for_input = AsyncMock()
+
+        mock_src_client = _make_mock_superset_client()
+        mock_src_client.get_dashboards = AsyncMock(
+            return_value=(True, [_make_dashboard(1, "Dash")])
+        )
+        mock_src_client.export_dashboard = AsyncMock(return_value=(b"zip", "meta"))
+        mock_tgt_client = _make_mock_superset_client()
+        mock_tgt_client.import_dashboard = AsyncMock(
+            side_effect=[
+                RuntimeError("Must provide a password for the database 'PG'"),
+                None,
+            ]
+        )
+
+        mock_engine = MagicMock()
+        mock_engine.transform_zip.return_value = True
+
+        with patch("src.plugins.migration.get_config_manager", return_value=mock_cm), \
+             patch("src.plugins.migration.SupersetClient") as MockSC, \
+             patch("src.plugins.migration.MigrationEngine", return_value=mock_engine), \
+             patch("src.plugins.migration.create_temp_file", return_value=_make_mock_ctf()), \
+             patch("src.dependencies.get_task_manager", return_value=mock_task_manager), \
+             patch("src.plugins.migration.IdMappingService", return_value=_make_mock_mapping_service()), \
+             patch("src.plugins.migration.SessionLocal"):
+
+            MockSC.side_effect = [mock_src_client, mock_tgt_client]
+
+            await plugin.execute({
+                "source_env_id": "env-1",
+                "target_env_id": "env-2",
+                "selected_ids": [1],
+                "replace_db_config": False,
+                "_task_id": "task-clean-1",
+            })
+
+            # After retry, passwords must be removed from task params
+            assert "passwords" not in task_mock_params
+    # #endregion test_full_flow_passwords_cleaned_after_retry
+
+
+# ══════════════════════════════════════════════════════════════════════════════
+# EDGE CASES: unknown db_name patterns, multi-dashboard
+# ══════════════════════════════════════════════════════════════════════════════
+
+
+class TestPasswordInjectionEdgeCases:
+    """Cover edge cases for the password error handling."""
+
+    # #region test_password_error_unknown_pattern [C:2] [TYPE Function]
+    # @BRIEF Edge: password error message with format not matching known regex → db_name="unknown".
+    @pytest.mark.asyncio
+    async def test_password_error_unknown_pattern(self):
+        """When the error message doesn't match databases/*.yaml or 'db_name', fallback to 'unknown'."""
+        plugin = MigrationPlugin()
+        src_env = _make_env("env-1", "Source")
+        tgt_env = _make_env("env-2", "Target")
+
+        mock_cm = MagicMock()
+        mock_cm.get_environments.return_value = [src_env, tgt_env]
+
+        mock_task_manager = MagicMock()
+        mock_task_manager.get_task.return_value = MagicMock(
+            params={"passwords": {}}  # empty passwords → retry fails → partial
+        )
+        mock_task_manager.await_input = AsyncMock()
+        mock_task_manager.wait_for_input = AsyncMock()
+
+        mock_src_client = _make_mock_superset_client()
+        mock_src_client.get_dashboards = AsyncMock(
+            return_value=(True, [_make_dashboard(1, "Broken Import")])
+        )
+        mock_src_client.export_dashboard = AsyncMock(return_value=(b"zip", "meta"))
+        mock_tgt_client = _make_mock_superset_client()
+        # Error message with no recognizable db name pattern at all
+        mock_tgt_client.import_dashboard = AsyncMock(
+            side_effect=[
+                RuntimeError("Must provide a password for the database"),
+                None,
+            ]
+        )
+
+        mock_engine = MagicMock()
+        mock_engine.transform_zip.return_value = True
+
+        with patch("src.plugins.migration.get_config_manager", return_value=mock_cm), \
+             patch("src.plugins.migration.SupersetClient") as MockSC, \
+             patch("src.plugins.migration.MigrationEngine", return_value=mock_engine), \
+             patch("src.plugins.migration.create_temp_file", return_value=_make_mock_ctf()), \
+             patch("src.dependencies.get_task_manager", return_value=mock_task_manager), \
+             patch("src.plugins.migration.IdMappingService", return_value=_make_mock_mapping_service()), \
+             patch("src.plugins.migration.SessionLocal"):
+
+            MockSC.side_effect = [mock_src_client, mock_tgt_client]
+
+            await plugin.execute({
+                "source_env_id": "env-1",
+                "target_env_id": "env-2",
+                "selected_ids": [1],
+                "replace_db_config": False,
+                "_task_id": "task-unknown-1",
+            })
+
+            # Fallback to "unknown" in the input_request
+            mock_task_manager.await_input.assert_called_once_with(
+                "task-unknown-1",
+                {
+                    "type": "database_password",
+                    "databases": ["unknown"],
+                    "error_message": "A database password is required to continue this migration.",
+                },
+            )
+    # #endregion test_password_error_unknown_pattern
+
+    # #region test_partial_password_multiple_dashboards [C:2] [TYPE Function]
+    # @BRIEF Edge: 3 dashboards — 1st succeeds, 2nd password error → recovered, 3rd succeeds.
+    @pytest.mark.asyncio
+    async def test_partial_password_multiple_dashboards(self):
+        """Multi-dashboard migration where one hits a password error and recovers."""
+        plugin = MigrationPlugin()
+        src_env = _make_env("env-1", "Source")
+        tgt_env = _make_env("env-2", "Target")
+
+        mock_cm = MagicMock()
+        mock_cm.get_environments.return_value = [src_env, tgt_env]
+
+        mock_task_manager = MagicMock()
+        mock_task_manager.get_task.return_value = MagicMock(
+            params={"passwords": {"ClickHouse": "p@ss"}}
+        )
+        mock_task_manager.await_input = AsyncMock()
+        mock_task_manager.wait_for_input = AsyncMock()
+
+        mock_src_client = _make_mock_superset_client()
+        dashboards = [
+            _make_dashboard(1, "Dash A"),
+            _make_dashboard(2, "Dash B (needs password)"),
+            _make_dashboard(3, "Dash C"),
+        ]
+        mock_src_client.get_dashboards = AsyncMock(return_value=(True, dashboards))
+        mock_src_client.export_dashboard = AsyncMock(return_value=(b"zip", "meta"))
+        mock_tgt_client = _make_mock_superset_client()
+        # Dash 1 → success, Dash 2 → password error → retry success, Dash 3 → success
+        mock_tgt_client.import_dashboard = AsyncMock(
+            side_effect=[
+                None,  # dash 1: success
+                RuntimeError(
+                    "Must provide a password for the database databases/ClickHouse.yaml"
+                ),  # dash 2: fail
+                None,  # dash 2: retry success
+                None,  # dash 3: success
+            ]
+        )
+
+        mock_engine = MagicMock()
+        mock_engine.transform_zip.return_value = True
+
+        with patch("src.plugins.migration.get_config_manager", return_value=mock_cm), \
+             patch("src.plugins.migration.SupersetClient") as MockSC, \
+             patch("src.plugins.migration.MigrationEngine", return_value=mock_engine), \
+             patch("src.plugins.migration.create_temp_file", return_value=_make_mock_ctf()), \
+             patch("src.dependencies.get_task_manager", return_value=mock_task_manager), \
+             patch("src.plugins.migration.IdMappingService", return_value=_make_mock_mapping_service()), \
+             patch("src.plugins.migration.SessionLocal"):
+
+            MockSC.side_effect = [mock_src_client, mock_tgt_client]
+
+            result = await plugin.execute({
+                "source_env_id": "env-1",
+                "target_env_id": "env-2",
+                "selected_ids": [1, 2, 3],
+                "replace_db_config": False,
+                "_task_id": "task-multi-1",
+            })
+
+            # All 3 dashboards should succeed (2nd via password injection)
+            assert result["status"] == "SUCCESS"
+            assert len(result["migrated_dashboards"]) == 3
+            assert len(result["failed_dashboards"]) == 0
+
+            # await_input should be called exactly once (for dash 2)
+            assert mock_task_manager.await_input.call_count == 1
+            assert mock_task_manager.wait_for_input.call_count == 1
+
+            # import_dashboard called 4 times (dash1, dash2-fail, dash2-retry, dash3)
+            assert mock_tgt_client.import_dashboard.call_count == 4
+    # #endregion test_partial_password_multiple_dashboards
+
+    # #region test_password_error_yaml_with_dots_in_name [C:2] [TYPE Function]
+    # @BRIEF Edge: YAML filename with dots (e.g. Dev_Clickhouse_Node_1.yaml) → correct extraction.
+    @pytest.mark.asyncio
+    async def test_password_error_yaml_with_dots_in_name(self):
+        """Path like databases/Dev_Clickhouse_Node_1.yaml with dots → extract correctly."""
+        plugin = MigrationPlugin()
+        src_env = _make_env("env-1", "Source")
+        tgt_env = _make_env("env-2", "Target")
+
+        mock_cm = MagicMock()
+        mock_cm.get_environments.return_value = [src_env, tgt_env]
+
+        mock_task_manager = MagicMock()
+        mock_task_manager.get_task.return_value = MagicMock(
+            params={"passwords": {"Dev_Clickhouse_Node_1": "pwd123"}}
+        )
+        mock_task_manager.await_input = AsyncMock()
+        mock_task_manager.wait_for_input = AsyncMock()
+
+        mock_src_client = _make_mock_superset_client()
+        mock_src_client.get_dashboards = AsyncMock(
+            return_value=(True, [_make_dashboard(1, "Dash")])
+        )
+        mock_src_client.export_dashboard = AsyncMock(return_value=(b"zip", "meta"))
+        mock_tgt_client = _make_mock_superset_client()
+        mock_tgt_client.import_dashboard = AsyncMock(
+            side_effect=[
+                RuntimeError(
+                    "Must provide a password for the database databases/Dev_Clickhouse_Node_1.yaml"
+                ),
+                None,
+            ]
+        )
+
+        mock_engine = MagicMock()
+        mock_engine.transform_zip.return_value = True
+
+        with patch("src.plugins.migration.get_config_manager", return_value=mock_cm), \
+             patch("src.plugins.migration.SupersetClient") as MockSC, \
+             patch("src.plugins.migration.MigrationEngine", return_value=mock_engine), \
+             patch("src.plugins.migration.create_temp_file", return_value=_make_mock_ctf()), \
+             patch("src.dependencies.get_task_manager", return_value=mock_task_manager), \
+             patch("src.plugins.migration.IdMappingService", return_value=_make_mock_mapping_service()), \
+             patch("src.plugins.migration.SessionLocal"):
+
+            MockSC.side_effect = [mock_src_client, mock_tgt_client]
+
+            await plugin.execute({
+                "source_env_id": "env-1",
+                "target_env_id": "env-2",
+                "selected_ids": [1],
+                "replace_db_config": False,
+                "_task_id": "task-dots-1",
+            })
+
+            # Regex r"databases/([^.]+)\.yaml" extracts up to first dot: "Dev_Clickhouse_Node_1"
+            mock_task_manager.await_input.assert_called_once_with(
+                "task-dots-1",
+                {
+                    "type": "database_password",
+                    "databases": ["Dev_Clickhouse_Node_1"],
+                    "error_message": "A database password is required to continue this migration.",
+                },
+            )
+    # #endregion test_password_error_yaml_with_dots_in_name
+
+    # #region test_password_error_no_task_id_skips_await [C:2] [TYPE Function]
+    # @BRIEF Edge: password error without task_id — await_input not called, dashboard marked failed.
+    @pytest.mark.asyncio
+    async def test_password_error_no_task_id_skips_await(self):
+        """When _task_id is not provided, skip await_input and mark dashboard as failed."""
+        plugin = MigrationPlugin()
+        src_env = _make_env("env-1", "Source")
+        tgt_env = _make_env("env-2", "Target")
+
+        mock_cm = MagicMock()
+        mock_cm.get_environments.return_value = [src_env, tgt_env]
+
+        mock_task_manager = MagicMock()
+        mock_task_manager.await_input = AsyncMock()
+        mock_task_manager.wait_for_input = AsyncMock()
+
+        mock_src_client = _make_mock_superset_client()
+        mock_src_client.get_dashboards = AsyncMock(
+            return_value=(True, [_make_dashboard(1, "Dash")])
+        )
+        mock_src_client.export_dashboard = AsyncMock(return_value=(b"zip", "meta"))
+        mock_tgt_client = _make_mock_superset_client()
+        mock_tgt_client.import_dashboard = AsyncMock(
+            side_effect=[
+                RuntimeError(
+                    "Must provide a password for the database databases/ClickHouse.yaml"
+                ),
+            ]
+        )
+
+        mock_engine = MagicMock()
+        mock_engine.transform_zip.return_value = True
+
+        with patch("src.plugins.migration.get_config_manager", return_value=mock_cm), \
+             patch("src.plugins.migration.SupersetClient") as MockSC, \
+             patch("src.plugins.migration.MigrationEngine", return_value=mock_engine), \
+             patch("src.plugins.migration.create_temp_file", return_value=_make_mock_ctf()), \
+             patch("src.dependencies.get_task_manager", return_value=mock_task_manager), \
+             patch("src.plugins.migration.IdMappingService", return_value=_make_mock_mapping_service()), \
+             patch("src.plugins.migration.SessionLocal"):
+
+            MockSC.side_effect = [mock_src_client, mock_tgt_client]
+
+            result = await plugin.execute({
+                "source_env_id": "env-1",
+                "target_env_id": "env-2",
+                "selected_ids": [1],
+                "replace_db_config": False,
+                # No _task_id — should skip await_input
+            })
+
+            # await_input must NOT be called without task_id
+            mock_task_manager.await_input.assert_not_called()
+
+            # Dashboard must be marked as failed
+            assert len(result["failed_dashboards"]) == 1
+            assert result["failed_dashboards"][0]["title"] == "Dash"
+    # #endregion test_password_error_no_task_id_skips_await
+
+
+# ══════════════════════════════════════════════════════════════════════════════
+# INTEGRATION-STYLE: cross-filter + password injection interplay
+# ══════════════════════════════════════════════════════════════════════════════
+
+
+class TestPasswordInjectionIntegration:
+    """Integration-style tests combining password injection with other migration features."""
+
+    # #region test_fix_cross_filters_passed_to_transform_zip [C:2] [TYPE Function]
+    # @BRIEF Integration: fix_cross_filters param flows through to transform_zip even during password errors.
+    @pytest.mark.asyncio
+    async def test_fix_cross_filters_passed_to_transform_zip(self):
+        """Verify fix_cross_filters=True reaches transform_zip during password injection flow."""
+        plugin = MigrationPlugin()
+        src_env = _make_env("env-1", "Source")
+        tgt_env = _make_env("env-2", "Target")
+
+        mock_cm = MagicMock()
+        mock_cm.get_environments.return_value = [src_env, tgt_env]
+
+        mock_task_manager = MagicMock()
+        mock_task_manager.get_task.return_value = MagicMock(
+            params={"passwords": {"PG": "s3cret"}}
+        )
+        mock_task_manager.await_input = AsyncMock()
+        mock_task_manager.wait_for_input = AsyncMock()
+
+        mock_src_client = _make_mock_superset_client()
+        mock_src_client.get_dashboards = AsyncMock(
+            return_value=(True, [_make_dashboard(1, "Dash")])
+        )
+        mock_src_client.export_dashboard = AsyncMock(return_value=(b"zip", "meta"))
+        mock_tgt_client = _make_mock_superset_client()
+        mock_tgt_client.import_dashboard = AsyncMock(
+            side_effect=[
+                RuntimeError("Must provide a password for the database 'PG'"),
+                None,
+            ]
+        )
+
+        mock_engine = MagicMock()
+        mock_engine.transform_zip.return_value = True
+
+        with patch("src.plugins.migration.get_config_manager", return_value=mock_cm), \
+             patch("src.plugins.migration.SupersetClient") as MockSC, \
+             patch("src.plugins.migration.MigrationEngine", return_value=mock_engine), \
+             patch("src.plugins.migration.create_temp_file", return_value=_make_mock_ctf()), \
+             patch("src.dependencies.get_task_manager", return_value=mock_task_manager), \
+             patch("src.plugins.migration.IdMappingService", return_value=_make_mock_mapping_service()), \
+             patch("src.plugins.migration.SessionLocal"):
+
+            MockSC.side_effect = [mock_src_client, mock_tgt_client]
+
+            await plugin.execute({
+                "source_env_id": "env-1",
+                "target_env_id": "env-2",
+                "selected_ids": [1],
+                "replace_db_config": False,
+                "fix_cross_filters": True,
+                "_task_id": "task-xfilt-1",
+            })
+
+            # transform_zip must be called with fix_cross_filters=True
+            assert mock_engine.transform_zip.call_count == 1
+            transform_kwargs = mock_engine.transform_zip.call_args[1]
+            assert transform_kwargs["fix_cross_filters"] is True
+    # #endregion test_fix_cross_filters_passed_to_transform_zip
+
+    # #region test_password_injection_with_task_context [C:2] [TYPE Function]
+    # @BRIEF Integration: password injection flow works correctly with TaskContext logger.
+    @pytest.mark.asyncio
+    async def test_password_injection_with_task_context(self):
+        """Ensure password injection works when TaskContext is provided (production path)."""
+        plugin = MigrationPlugin()
+        src_env = _make_env("env-1", "Source")
+        tgt_env = _make_env("env-2", "Target")
+
+        mock_cm = MagicMock()
+        mock_cm.get_environments.return_value = [src_env, tgt_env]
+
+        mock_task_manager = MagicMock()
+        mock_task_manager.get_task.return_value = MagicMock(
+            params={"passwords": {"PG": "pwd"}}
+        )
+        mock_task_manager.await_input = AsyncMock()
+        mock_task_manager.wait_for_input = AsyncMock()
+
+        mock_src_client = _make_mock_superset_client()
+        mock_src_client.get_dashboards = AsyncMock(
+            return_value=(True, [_make_dashboard(1, "Dash")])
+        )
+        mock_src_client.export_dashboard = AsyncMock(return_value=(b"zip", "meta"))
+        mock_tgt_client = _make_mock_superset_client()
+        mock_tgt_client.import_dashboard = AsyncMock(
+            side_effect=[
+                RuntimeError("Must provide a password for the database 'PG'"),
+                None,
+            ]
+        )
+
+        mock_engine = MagicMock()
+        mock_engine.transform_zip.return_value = True
+
+        # Create a TaskContext with a mock logger
+        ctx = MagicMock()
+        ctx.logger = MagicMock()
+        ctx.logger.with_source.return_value = MagicMock()
+
+        with patch("src.plugins.migration.get_config_manager", return_value=mock_cm), \
+             patch("src.plugins.migration.SupersetClient") as MockSC, \
+             patch("src.plugins.migration.MigrationEngine", return_value=mock_engine), \
+             patch("src.plugins.migration.create_temp_file", return_value=_make_mock_ctf()), \
+             patch("src.dependencies.get_task_manager", return_value=mock_task_manager), \
+             patch("src.plugins.migration.IdMappingService", return_value=_make_mock_mapping_service()), \
+             patch("src.plugins.migration.SessionLocal"):
+
+            MockSC.side_effect = [mock_src_client, mock_tgt_client]
+
+            result = await plugin.execute(
+                {
+                    "source_env_id": "env-1",
+                    "target_env_id": "env-2",
+                    "selected_ids": [1],
+                    "replace_db_config": False,
+                    "_task_id": "task-ctx-1",
+                },
+                context=ctx,
+            )
+
+            assert result["status"] == "SUCCESS"
+            assert len(result["migrated_dashboards"]) == 1
+    # #endregion test_password_injection_with_task_context
+
+# #endregion Test.MigrationPlugin.Password

backend/tests/test_task_manager.py

@@ -568,6 +568,27 @@ class TestTaskManagerInput:
         finally:
             _cleanup_manager(mgr)
 
+    @pytest.mark.asyncio
+    async def test_immediate_password_resume_is_not_lost_before_wait(self):
+        """@TEST_EDGE Resume arriving immediately after the input prompt must unblock migration."""
+        mgr, _, _, _ = _make_manager()
+        try:
+            from src.core.task_manager.models import Task, TaskStatus
+
+            task = Task(plugin_id="p1", params={})
+            task.status = TaskStatus.RUNNING
+            mgr.tasks[task.id] = task
+            mgr._add_log = AsyncMock()
+
+            await mgr.await_input(task.id, {"type": "database_password"})
+            await mgr.resume_task_with_password(task.id, {"db1": "secret"})
+
+            await asyncio.wait_for(mgr.wait_for_input(task.id), timeout=0.1)
+            assert task.status == TaskStatus.RUNNING
+            assert task.id not in mgr.graph.task_futures
+        finally:
+            _cleanup_manager(mgr)
+
     @pytest.mark.asyncio
     async def test_await_input_not_running_raises(self):
         mgr, _, _, _ = _make_manager()