From 81ba94684fdd0d439e1b02bbd36cff353ab7caec Mon Sep 17 00:00:00 2001 From: busya Date: Wed, 19 Aug 2026 10:29:28 +0300 Subject: [PATCH] =?UTF-8?q?feat:=20sed-=D0=BC=D1=83=D1=82=D0=B0=D1=86?= =?UTF-8?q?=D0=B8=D0=B8=20=D0=B4=D0=B0=D1=82=D0=B0=D1=81=D0=B5=D1=82=D0=BE?= =?UTF-8?q?=D0=B2,=20=D0=BF=D1=80=D0=B0=D0=B2=D0=BA=D0=B8=20=D0=BC=D0=B8?= =?UTF-8?q?=D0=B3=D1=80=D0=B0=D1=86=D0=B8=D0=B8=20=D0=B8=20=D1=84=D0=B8?= =?UTF-8?q?=D0=BA=D1=81=D1=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Dataset viewer: sed-замена (find→replace) по всем/выбранным/текущим датасетам с обязательным визуальным предпросмотром и целями (sql/metrics/yaml); сохранение и выбор именованных правил (sedRules store + SedRuleEditor). - Migration: literal-замена в dataset YAML при переносе; рескан ID чартов/датасетов перед миграцией + метрика средней длительности синка (GET /migration/sync-stats); логическая группировка опций (маппинги БД, сервер изменений, rescan) под галочками. - Fix: дашборды хаба скрывались дефолтным профиль-фильтром show_only_slug_dashboards=true (теперь false); мастер миграции не грузил дашборды предзаполненного источника; потеря терминального task_status из-за утечки pending-корутин в /ws/logs (панель результата не появлялась); горизонтальный скролл в MappingTable; чистая per-env ошибка в mapping coverage. - Backend: record_sync_duration + alembic-миграция sync-duration; literal_replace модуль. --- .gitignore | 4 + ...4y5z6_add_sync_duration_to_environments.py | 44 +++ .../routes/__tests__/test_datasets_routes.py | 189 +++++++++++ .../routes/__tests__/test_migration_routes.py | 49 +++ .../api/routes/dashboards/_listing_routes.py | 2 +- backend/src/api/routes/datasets.py | 304 +++++++++++++++++- backend/src/api/routes/migration.py | 34 +- backend/src/app.py | 9 +- backend/src/core/database.py | 2 +- backend/src/core/literal_replace.py | 30 ++ backend/src/core/mapping_service.py | 23 +- backend/src/core/migration_engine.py | 33 ++ backend/src/models/dashboard.py | 3 + backend/src/models/mapping.py | 6 +- backend/src/models/profile.py | 2 +- backend/src/plugins/migration.py | 36 ++- .../translate/__tests__/test_batch_insert.py | 43 ++- .../translate/__tests__/test_executor.py | 33 +- .../src/plugins/translate/_batch_insert.py | 62 +++- backend/src/plugins/translate/executor.py | 67 +++- .../plugins/translate/orchestrator_retry.py | 26 +- backend/src/schemas/profile.py | 4 +- backend/src/services/mapping_analysis.py | 12 + .../services/profile_preference_service.py | 4 +- backend/src/services/profile_utils.py | 2 +- backend/tests/api/test_mappings.py | 18 ++ backend/tests/core/test_migration_engine.py | 57 ++++ .../tests/plugins/translate/test_executor.py | 4 +- .../translate/test_orchestrator_retry.py | 8 +- backend/tests/schemas/test_profile.py | 2 +- .../test_profile_preference_service.py | 2 +- backend/tests/services/test_profile_utils.py | 2 +- backend/tests/test_app_ws_events.py | 80 +++++ docs/adr-sed-dataset-mutations.md | 36 +++ .../src/lib/components/SedRuleEditor.svelte | 105 ++++++ .../migration/MappingCoverageOverview.svelte | 13 +- .../__tests__/MappingCoverageOverview.test.ts | 16 + .../src/lib/components/ui/MappingTable.svelte | 18 +- frontend/src/lib/i18n/index.svelte.ts | 4 + .../src/lib/i18n/locales/en/migration.json | 13 +- frontend/src/lib/i18n/locales/en/sed.json | 36 +++ .../src/lib/i18n/locales/ru/migration.json | 13 +- frontend/src/lib/i18n/locales/ru/sed.json | 36 +++ .../src/lib/models/DatasetsHubModel.svelte.ts | 167 ++++++++++ .../src/lib/models/MigrationModel.svelte.ts | 54 +++- .../models/__tests__/DatasetsHubModel.test.ts | 150 ++++++++- .../models/__tests__/MigrationModel.test.ts | 41 +++ .../src/lib/stores/__tests__/test_sedRules.ts | 41 +++ frontend/src/lib/stores/sedRules.svelte.ts | 74 +++++ frontend/src/routes/datasets/+page.svelte | 137 ++++++++ .../src/routes/datasets/DatasetList.svelte | 7 + frontend/src/routes/migration/+page.svelte | 252 +++++++++------ frontend/src/types/dashboard.ts | 3 + 53 files changed, 2224 insertions(+), 188 deletions(-) create mode 100644 backend/alembic/versions/u1v2w3x4y5z6_add_sync_duration_to_environments.py create mode 100644 backend/src/core/literal_replace.py create mode 100644 docs/adr-sed-dataset-mutations.md create mode 100644 frontend/src/lib/components/SedRuleEditor.svelte create mode 100644 frontend/src/lib/i18n/locales/en/sed.json create mode 100644 frontend/src/lib/i18n/locales/ru/sed.json create mode 100644 frontend/src/lib/stores/__tests__/test_sedRules.ts create mode 100644 frontend/src/lib/stores/sedRules.svelte.ts diff --git a/.gitignore b/.gitignore index c68d28b82..ba1299815 100755 --- a/.gitignore +++ b/.gitignore @@ -132,8 +132,12 @@ artifacts/ .npmrc # Binary blobs / PDFs (research material, not source) +research *.pdf +# SQLite in-memory test artifacts +:memory:* + # Client-specific certs (not secrets, but not part of the source tree) /RUSAL_ROOT.cer diff --git a/backend/alembic/versions/u1v2w3x4y5z6_add_sync_duration_to_environments.py b/backend/alembic/versions/u1v2w3x4y5z6_add_sync_duration_to_environments.py new file mode 100644 index 000000000..4cd9d8d72 --- /dev/null +++ b/backend/alembic/versions/u1v2w3x4y5z6_add_sync_duration_to_environments.py @@ -0,0 +1,44 @@ +"""add ID-sync duration metrics to environments + +Revision ID: u1v2w3x4y5z6 +Revises: t1u2v3w4x5y6 +Create Date: 2026-08-18 + +""" +from collections.abc import Sequence + +import sqlalchemy as sa +from sqlalchemy import inspect + +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = "u1v2w3x4y5z6" +down_revision: str | Sequence[str] | None = "t1u2v3w4x5y6" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def _table_exists(table_name: str) -> bool: + conn = op.get_bind() + inspector = inspect(conn) + return table_name in inspector.get_table_names() + + +def upgrade() -> None: + """Add last/avg sync duration + run count to environments for the pre-migration ID rescan reference.""" + if not _table_exists("environments"): + return + op.add_column("environments", sa.Column("last_sync_duration_seconds", sa.Float(), nullable=True)) + op.add_column("environments", sa.Column("sync_duration_avg_seconds", sa.Float(), nullable=True)) + op.add_column( + "environments", + sa.Column("sync_run_count", sa.Integer(), nullable=False, server_default=sa.text("0")), + ) + + +def downgrade() -> None: + """Remove the ID-sync duration metrics columns.""" + op.drop_column("environments", "sync_run_count") + op.drop_column("environments", "sync_duration_avg_seconds") + op.drop_column("environments", "last_sync_duration_seconds") diff --git a/backend/src/api/routes/__tests__/test_datasets_routes.py b/backend/src/api/routes/__tests__/test_datasets_routes.py index b649a343d..a529541c9 100644 --- a/backend/src/api/routes/__tests__/test_datasets_routes.py +++ b/backend/src/api/routes/__tests__/test_datasets_routes.py @@ -269,4 +269,193 @@ def test_get_datasets_superset_failure(mock_deps): assert response.status_code == 503 assert "Failed to fetch datasets" in response.json()["detail"] # #endregion Test.Tests.TestGetDatasetsSupersetFailure + + +# #region Test.Tests.TestMutateDatasets [C:3] [TYPE Class] [SEMANTICS datasets,api,mutation,sed,preview] +# @BRIEF Unit tests for POST /api/datasets/mutate and /mutate/preview — sed-style literal mutation with target scoping. +# @TEST_INVARIANT target-scoping -> sql/metrics/yaml targets mutate exactly their declared fields; preview never writes. +class _FakeSupersetClient: + def __init__(self, env, updates): + self.env = env + self.updates = updates + + async def get_dataset(self, dataset_id): + return { + "result": { + "id": dataset_id, + "table_name": "dm_view.account_debt", + "sql": "SELECT * FROM dm_view.account_debt", + "description": "keep dm_view.account_debt as is", + "metrics": [ + {"id": 1, "expression": "COUNT(dm_view.account_debt)", "verbose_name": "debt dm_view.account_debt", "metric_name": "debt_m"} + ], + } + } + + async def update_dataset(self, dataset_id, data, override_columns=False): + self.updates.append((dataset_id, data)) + return {"id": dataset_id} + + +def _monkeypatch_client(monkeypatch, updates): + import src.api.routes.datasets as route_mod + + def _factory(env): + return _FakeSupersetClient(env, updates) + + monkeypatch.setattr(route_mod, "AsyncSupersetClient", _factory) + + +def _mutate_json(overrides=None): + payload = { + "env_id": "prod", + "dataset_ids": [7], + "find": "dm_view.account_debt", + "replace": "dm_view.account_debt_final", + "targets": ["sql"], + } + if overrides: + payload.update(overrides) + return payload + + +def test_mutate_datasets_sql_target_only(mock_deps, monkeypatch): + mock_env = MagicMock() + mock_env.id = "prod" + mock_deps["config"].get_environments.return_value = [mock_env] + updates = [] + _monkeypatch_client(monkeypatch, updates) + + response = client.post("/api/datasets/mutate", json=_mutate_json({"targets": ["sql"]})) + assert response.status_code == 200 + assert response.json()["mutated_count"] == 1 + + dataset_id, data = updates[0] + assert data["sql"] == "SELECT * FROM dm_view.account_debt_final" + # non-targeted fields untouched + assert data["table_name"] == "dm_view.account_debt" + assert data["description"] == "keep dm_view.account_debt as is" + assert data["metrics"][0]["expression"] == "COUNT(dm_view.account_debt)" + + +def test_mutate_datasets_sql_metrics_targets(mock_deps, monkeypatch): + mock_env = MagicMock() + mock_env.id = "prod" + mock_deps["config"].get_environments.return_value = [mock_env] + updates = [] + _monkeypatch_client(monkeypatch, updates) + + response = client.post("/api/datasets/mutate", json=_mutate_json({"targets": ["sql", "metrics"]})) + assert response.status_code == 200 + + _, data = updates[0] + assert data["sql"] == "SELECT * FROM dm_view.account_debt_final" + assert data["metrics"][0]["expression"] == "COUNT(dm_view.account_debt_final)" + assert data["metrics"][0]["verbose_name"] == "debt dm_view.account_debt_final" + # metric_name not targeted; table_name/description untouched + assert data["metrics"][0]["metric_name"] == "debt_m" + assert data["table_name"] == "dm_view.account_debt" + assert data["description"] == "keep dm_view.account_debt as is" + + +def test_mutate_datasets_yaml_target_replaces_everything(mock_deps, monkeypatch): + mock_env = MagicMock() + mock_env.id = "prod" + mock_deps["config"].get_environments.return_value = [mock_env] + updates = [] + _monkeypatch_client(monkeypatch, updates) + + response = client.post("/api/datasets/mutate", json=_mutate_json({"targets": ["yaml"]})) + assert response.status_code == 200 + + _, data = updates[0] + assert data["sql"] == "SELECT * FROM dm_view.account_debt_final" + assert data["table_name"] == "dm_view.account_debt_final" + assert data["description"] == "keep dm_view.account_debt_final as is" + assert data["metrics"][0]["expression"] == "COUNT(dm_view.account_debt_final)" + + +def test_mutate_datasets_all(mock_deps, monkeypatch): + mock_env = MagicMock() + mock_env.id = "prod" + mock_deps["config"].get_environments.return_value = [mock_env] + mock_deps["task"].get_all_tasks.return_value = [] + mock_deps["resource"].get_datasets_with_status = AsyncMock( + return_value=[{"id": 11}, {"id": 12}] + ) + updates = [] + _monkeypatch_client(monkeypatch, updates) + + response = client.post( + "/api/datasets/mutate", + json={"env_id": "prod", "dataset_ids": [], "find": "dm_view.account_debt", "replace": "x", "targets": ["sql"]}, + ) + assert response.status_code == 200 + assert response.json()["mutated_count"] == 2 + + +def test_mutate_datasets_empty_find_rejected(mock_deps): + mock_env = MagicMock() + mock_env.id = "prod" + mock_deps["config"].get_environments.return_value = [mock_env] + response = client.post( + "/api/datasets/mutate", + json={"env_id": "prod", "dataset_ids": [1], "find": "", "replace": "x", "targets": ["sql"]}, + ) + assert response.status_code == 422 + + +def test_mutate_datasets_invalid_targets_rejected(mock_deps): + mock_env = MagicMock() + mock_env.id = "prod" + mock_deps["config"].get_environments.return_value = [mock_env] + # yaml cannot combine with others + assert client.post("/api/datasets/mutate", json=_mutate_json({"targets": ["sql", "yaml"]})).status_code == 422 + # empty targets + assert client.post("/api/datasets/mutate", json=_mutate_json({"targets": []})).status_code == 422 + # unknown target + assert client.post("/api/datasets/mutate", json=_mutate_json({"targets": ["bogus"]})).status_code == 422 + + +def test_preview_mutate_datasets_readonly(mock_deps, monkeypatch): + mock_env = MagicMock() + mock_env.id = "prod" + mock_deps["config"].get_environments.return_value = [mock_env] + updates = [] + _monkeypatch_client(monkeypatch, updates) + + response = client.post("/api/datasets/mutate/preview", json=_mutate_json({"targets": ["sql", "metrics"]})) + assert response.status_code == 200 + body = response.json() + assert body["affected"] == 1 + assert body["scanned"] == 1 + assert len(body["previews"]) == 1 + + preview = body["previews"][0] + assert preview["dataset_id"] == 7 + assert preview["table_name"] == "dm_view.account_debt" + fields = {e["field"] for e in preview["edits"]} + assert "sql" in fields + assert "metrics.expression" in fields + + # preview must never write + assert updates == [] + + +def test_preview_mutate_datasets_no_changes(mock_deps, monkeypatch): + mock_env = MagicMock() + mock_env.id = "prod" + mock_deps["config"].get_environments.return_value = [mock_env] + updates = [] + _monkeypatch_client(monkeypatch, updates) + + response = client.post("/api/datasets/mutate/preview", json=_mutate_json({"find": "no_such_needle"})) + assert response.status_code == 200 + body = response.json() + assert body["affected"] == 0 + assert body["previews"] == [] + assert updates == [] + + +# #endregion Test.Tests.TestMutateDatasets # #endregion Test.Tests.DatasetsApiTests diff --git a/backend/src/api/routes/__tests__/test_migration_routes.py b/backend/src/api/routes/__tests__/test_migration_routes.py index d1ffbce46..71d957547 100644 --- a/backend/src/api/routes/__tests__/test_migration_routes.py +++ b/backend/src/api/routes/__tests__/test_migration_routes.py @@ -410,6 +410,55 @@ async def test_trigger_sync_now_idempotent_env_upsert(db_session, _mock_env): await trigger_sync_now(config_manager=cm, db=db_session, _=None) env_count = db_session.query(EnvironmentModel).filter_by(id="test-env-1").count() assert env_count == 1 + + +# --- record_sync_duration / sync-stats tests --- +def test_record_sync_duration_running_average(db_session): + """Running average and run count accumulate across recorded syncs.""" + from src.core.mapping_service import record_sync_duration + from src.models.mapping import Environment as EnvironmentModel + + env = EnvironmentModel(id="env-x", name="X", url="http://x", credentials_id="env-x") + db_session.add(env) + db_session.commit() + + record_sync_duration(db_session, "env-x", 10.0) + record_sync_duration(db_session, "env-x", 20.0) + + db_session.refresh(env) + assert env.sync_run_count == 2 + assert env.last_sync_duration_seconds == 20.0 + assert env.sync_duration_avg_seconds == 15.0 + + +@pytest.mark.asyncio +async def test_get_sync_stats_returns_duration(db_session): + """sync-stats surfaces last/avg duration and run count for an environment.""" + from src.api.routes.migration import get_sync_stats + from src.models.mapping import Environment as EnvironmentModel + + env = EnvironmentModel(id="env-x", name="X", url="http://x", credentials_id="env-x") + env.sync_duration_avg_seconds = 12.5 + env.last_sync_duration_seconds = 14.0 + env.sync_run_count = 3 + db_session.add(env) + db_session.commit() + + result = await get_sync_stats(env_id="env-x", db=db_session, _=None) + assert result["avg_duration_seconds"] == 12.5 + assert result["last_duration_seconds"] == 14.0 + assert result["run_count"] == 3 + + +@pytest.mark.asyncio +async def test_get_sync_stats_unknown_env_returns_nulls(db_session): + from src.api.routes.migration import get_sync_stats + + result = await get_sync_stats(env_id="missing", db=db_session, _=None) + assert result["avg_duration_seconds"] is None + assert result["run_count"] == 0 + + # --- get_dashboards tests --- @pytest.mark.asyncio async def test_get_dashboards_success(_mock_env): diff --git a/backend/src/api/routes/dashboards/_listing_routes.py b/backend/src/api/routes/dashboards/_listing_routes.py index 403be01a8..d6637587c 100644 --- a/backend/src/api/routes/dashboards/_listing_routes.py +++ b/backend/src/api/routes/dashboards/_listing_routes.py @@ -132,7 +132,7 @@ async def get_dashboards( page_context == "dashboards_main" and bool(apply_profile_default) and not bool(override_show_all) - and bool(profile_preference.get("show_only_slug_dashboards", True)) + and bool(profile_preference.get("show_only_slug_dashboards", False)) ) profile_match_logic = None diff --git a/backend/src/api/routes/datasets.py b/backend/src/api/routes/datasets.py index d000faa82..5a650a295 100644 --- a/backend/src/api/routes/datasets.py +++ b/backend/src/api/routes/datasets.py @@ -17,10 +17,11 @@ import re from fastapi import APIRouter, Depends, HTTPException, Query -from pydantic import BaseModel, ConfigDict, Field +from pydantic import BaseModel, ConfigDict, Field, field_validator from ...core.logger import belief_scope, logger from ...core.async_superset_client import AsyncSupersetClient +from ...core.literal_replace import replace_literals from ...dependencies import get_config_manager, get_resource_service, get_task_manager, has_permission router = APIRouter(prefix="/api/datasets", tags=["Datasets"]) @@ -453,6 +454,307 @@ async def generate_docs( raise HTTPException(status_code=503, detail=f"Failed to create documentation generation task: {e!s}") # #endregion Api.Datasets.GenerateDocs +# #region Api.Datasets.DatasetMutateRequest [C:2] [TYPE DataClass] +# @BRIEF Request DTO for sed-style literal mutation of datasets. +# @INVARIANT find must be non-empty (a no-op is rejected to avoid accidental full rewrites). +# @INVARIANT dataset_ids empty means "all datasets in the environment". +# @INVARIANT targets is a non-empty subset of {sql, metrics, yaml}; "yaml" (full replacement) is exclusive and dangerous. +class DatasetMutateRequest(BaseModel): + env_id: str = Field(..., description="Environment ID") + dataset_ids: list[int] = Field(default_factory=list, description="Dataset IDs to mutate; empty means all") + find: str = Field(..., min_length=1, max_length=10000, description="Literal substring to find") + replace: str = Field("", max_length=10000, description="Literal replacement (empty clears matches)") + targets: list[str] = Field(default_factory=lambda: ["sql"], description="Where to apply: 'sql' (dataset SQL), 'metrics' (metric expressions), 'yaml' (full replacement — dangerous, exclusive)") + + @field_validator("targets") + @classmethod + def _check_targets(cls, value: list[str]) -> list[str]: + allowed = {"sql", "metrics", "yaml"} + if not value: + raise ValueError("targets must not be empty") + if any(t not in allowed for t in value): + raise ValueError(f"invalid target; allowed values: {sorted(allowed)}") + if "yaml" in value and len(value) > 1: + raise ValueError("'yaml' (full replacement) cannot be combined with other targets") + return value +# #endregion Api.Datasets.DatasetMutateRequest + +# #region Api.Datasets.DatasetMutateEdit [C:1] [TYPE DataClass] +# @BRIEF A single identifier field that would change: field path, before, after. +class DatasetMutateEdit(BaseModel): + field: str + before: str + after: str +# #endregion Api.Datasets.DatasetMutateEdit + +# #region Api.Datasets.DatasetMutatePreviewItem [C:1] [TYPE DataClass] +# @BRIEF Per-dataset preview of the edits a literal mutation would apply. +class DatasetMutatePreviewItem(BaseModel): + dataset_id: int + table_name: str | None = None + edits: list[DatasetMutateEdit] +# #endregion Api.Datasets.DatasetMutatePreviewItem + + +# Literal-replacement target selectors shared by preview and apply. +_TARGET_SQL = "sql" +_TARGET_METRICS = "metrics" +_TARGET_YAML = "yaml" +_METRIC_LITERAL_FIELDS = ("expression", "verbose_name") + + +def _yaml_string_paths(obj: object): + """Yield (path, string) for every string value in the object tree (depth-first).""" + def _walk(node: object, path: str): + if isinstance(node, dict): + for key, val in node.items(): + _walk(val, f"{path}.{key}" if path else str(key)) + elif isinstance(node, list): + for idx, val in enumerate(node): + _walk(val, f"{path}[{idx}]") + elif isinstance(node, str): + yield path, node + yield from _walk(obj, "") + + +def _sql_edit(dataset: dict, find: str, replace: str) -> dict | None: + value = dataset.get("sql") + if not isinstance(value, str): + return None + updated = replace_literals(value, find, replace) + return {"field": "sql", "before": value, "after": updated} if updated != value else None + + +def _metric_edits(dataset: dict, find: str, replace: str) -> list[dict]: + edits: list[dict] = [] + for metric in dataset.get("metrics") or []: + if not isinstance(metric, dict): + continue + for key in _METRIC_LITERAL_FIELDS: + value = metric.get(key) + if not isinstance(value, str): + continue + updated = replace_literals(value, find, replace) + if updated != value: + edits.append({"field": f"metrics.{key}", "before": value, "after": updated}) + return edits + + +def _literal_edits(dataset: dict, find: str, replace: str, targets: list[str]) -> list[dict]: + """Read-only: return [{field, before, after}] for each targeted field that would change. + + - sql → dataset["sql"] only + - metrics → metric expression / verbose_name + - yaml → every string value in the whole dataset payload (dangerous, full replacement) + """ + edits: list[dict] = [] + if _TARGET_SQL in targets: + edit = _sql_edit(dataset, find, replace) + if edit: + edits.append(edit) + if _TARGET_METRICS in targets: + edits.extend(_metric_edits(dataset, find, replace)) + if _TARGET_YAML in targets: + for path, value in _yaml_string_paths(dataset): + updated = replace_literals(value, find, replace) + if updated != value: + edits.append({"field": path, "before": value, "after": updated}) + return edits + + +async def _resolve_mutate_dataset_ids( + request: DatasetMutateRequest, + env, + task_manager, + resource_service, +) -> list[int]: + """Resolve target ids: explicit list, or all datasets in the environment when empty.""" + if request.dataset_ids: + return list(request.dataset_ids) + all_tasks = task_manager.get_all_tasks() + datasets = await resource_service.get_datasets_with_status(env, all_tasks) + return [d["id"] for d in datasets if d.get("id") is not None] + + +def _mutate_env(config_manager, env_id: str): + environments = config_manager.get_environments() + env = next((e for e in environments if e.id == env_id), None) + if not env: + logger.explore("Environment not found for dataset mutation", payload={"env_id": env_id}, error=f"Environment not found: {env_id}") + raise HTTPException(status_code=404, detail="Environment not found") + return env + + +# #region Api.Datasets.PreviewMutateDatasets [C:4] [TYPE Function] +# @ingroup Api +# @BRIEF Read-only preview of a literal find→replace over all / selected / single datasets, WITHOUT mutating. +# @PRE User has plugin:migration READ permission; env_id exists; find is non-empty. +# @POST Returns per-dataset edits (before→after) so the caller can render a visual diff before applying. +# @SIDE_EFFECT Reads datasets from Superset; never writes. +# @INVARIANT The edit set (fields, before, after) is computed by the same `_literal_edits` used to apply. +@router.post("/mutate/preview") +async def preview_mutate_datasets( + request: DatasetMutateRequest, + config_manager=Depends(get_config_manager), + task_manager=Depends(get_task_manager), + resource_service=Depends(get_resource_service), + _ = Depends(has_permission("plugin:migration", "READ")), +): + with belief_scope("preview_mutate_datasets", f"env={request.env_id}, count={len(request.dataset_ids)}, find={request.find}"): + env = _mutate_env(config_manager, request.env_id) + dataset_ids = await _resolve_mutate_dataset_ids(request, env, task_manager, resource_service) + + client = AsyncSupersetClient(env) + previews: list[DatasetMutatePreviewItem] = [] + errors: list[dict] = [] + scanned = 0 + + for dataset_id in dataset_ids: + try: + response = await client.get_dataset(dataset_id) + except Exception as e: + errors.append({"dataset_id": dataset_id, "error": f"GET failed: {e!s}"}) + continue + + dataset = response.get("result") if isinstance(response, dict) and "result" in response else response + if not isinstance(dataset, dict) or not dataset.get("id"): + errors.append({"dataset_id": dataset_id, "error": "dataset not found"}) + continue + + scanned += 1 + edits = _literal_edits(dataset, request.find, request.replace, request.targets) + if edits: + previews.append( + DatasetMutatePreviewItem( + dataset_id=dataset_id, + table_name=dataset.get("table_name"), + edits=[DatasetMutateEdit(**e) for e in edits], + ) + ) + + logger.reflect("Dataset mutation preview complete", payload={"scanned": scanned, "affected": len(previews), "errors": len(errors)}) + return {"scanned": scanned, "affected": len(previews), "previews": previews, "errors": errors} +# #endregion Api.Datasets.PreviewMutateDatasets + + +# #region Api.Datasets.MutateDatasets [C:4] [TYPE Function] +# @ingroup Api +# @BRIEF Apply a literal find→replace to dataset sql / metrics / full yaml across all / selected / single datasets. +# @PRE User has plugin:migration WRITE permission; env_id exists; find is non-empty; targets valid. +# @POST Each target dataset is GET→modified→PUT back via Superset. Returns a per-dataset report. +# @SIDE_EFFECT Mutates dataset metadata in upstream Superset via PUT. +# @INVARIANT Replacement is scoped to the requested targets (sql, metrics, or full yaml); non-targeted data is left untouched. +# @INVARIANT The mutation is a pure function of the same `_literal_edits` the preview shows; caller MUST preview first. +# @RATIONALE Mirrors the migration engine's raw-text sed replacement, but over the live Superset API for in-place editing without an export/import round-trip. +@router.post("/mutate") +async def mutate_datasets( + request: DatasetMutateRequest, + config_manager=Depends(get_config_manager), + task_manager=Depends(get_task_manager), + resource_service=Depends(get_resource_service), + _ = Depends(has_permission("plugin:migration", "WRITE")), +): + with belief_scope("mutate_datasets", f"env={request.env_id}, count={len(request.dataset_ids)}, find={request.find}"): + env = _mutate_env(config_manager, request.env_id) + dataset_ids = await _resolve_mutate_dataset_ids(request, env, task_manager, resource_service) + + client = AsyncSupersetClient(env) + mutated: list[int] = [] + errors: list[dict] = [] + + for dataset_id in dataset_ids: + try: + response = await client.get_dataset(dataset_id) + except Exception as e: + logger.explore("Superset GET failed for mutate_datasets", payload={"dataset_id": dataset_id}, error=str(e)) + errors.append({"dataset_id": dataset_id, "error": f"GET failed: {e!s}"}) + continue + + dataset = response.get("result") if isinstance(response, dict) and "result" in response else response + if not isinstance(dataset, dict) or not dataset.get("id"): + errors.append({"dataset_id": dataset_id, "error": "dataset not found"}) + continue + + changed = _apply_literal_to_dataset(dataset, request.find, request.replace, request.targets) + if not changed: + continue + + try: + await client.update_dataset(dataset_id, dataset, override_columns=False) + except Exception as e: + logger.explore("Superset PUT failed for mutate_datasets", payload={"dataset_id": dataset_id}, error=str(e)) + errors.append({"dataset_id": dataset_id, "error": f"PUT failed: {e!s}"}) + continue + + mutated.append(dataset_id) + + logger.reflect("Dataset mutation complete", payload={"mutated": len(mutated), "errors": len(errors)}) + return {"mutated": mutated, "mutated_count": len(mutated), "errors": errors} + + +def _apply_yaml_replace(node: object, find: str, replace: str) -> bool: + """Recursively replace `find` in every string value of the tree; returns True if any changed.""" + if isinstance(node, dict): + items = node.items() + elif isinstance(node, list): + items = enumerate(node) + else: + return False + + changed = False + for key, val in items: + if isinstance(val, str): + updated = replace_literals(val, find, replace) + if updated != val: + node[key] = updated + changed = True + elif isinstance(val, (dict, list)): + changed = _apply_yaml_replace(val, find, replace) or changed + return changed + + +def _apply_sql(dataset: dict, find: str, replace: str) -> bool: + value = dataset.get("sql") + if not isinstance(value, str): + return False + updated = replace_literals(value, find, replace) + if updated != value: + dataset["sql"] = updated + return True + return False + + +def _apply_metrics(dataset: dict, find: str, replace: str) -> bool: + changed = False + for metric in dataset.get("metrics") or []: + if not isinstance(metric, dict): + continue + for key in _METRIC_LITERAL_FIELDS: + value = metric.get(key) + if not isinstance(value, str): + continue + updated = replace_literals(value, find, replace) + if updated != value: + metric[key] = updated + changed = True + return changed + + +def _apply_literal_to_dataset(dataset: dict, find: str, replace: str, targets: list[str]) -> bool: + """Apply literal `find`→`replace` to the targeted fields of a dataset dict; returns True if changed. + + Mirrors `_literal_edits` exactly: same targets, same fields. Never mutates non-targeted data. + """ + changed = False + if _TARGET_SQL in targets: + changed = _apply_sql(dataset, find, replace) or changed + if _TARGET_METRICS in targets: + changed = _apply_metrics(dataset, find, replace) or changed + if _TARGET_YAML in targets: + changed = _apply_yaml_replace(dataset, find, replace) or changed + return changed +# #endregion Api.Datasets.MutateDatasets + # #region Api.Datasets.GetDatasetDetail [C:4] [TYPE Function] # @ingroup Api # @BRIEF Get detailed dataset information including columns and linked dashboards diff --git a/backend/src/api/routes/migration.py b/backend/src/api/routes/migration.py index 0843773ac..6a60f1f9c 100644 --- a/backend/src/api/routes/migration.py +++ b/backend/src/api/routes/migration.py @@ -24,6 +24,7 @@ # @RATIONALE Separates API concerns (routing, permission checks, request/response serialization) from core migration business logic, enabling independent evolution of HTTP contract and domain logic. # @REJECTED Embedding API logic directly in the core migration layer was rejected — it would couple HTTP concerns with business logic, making both layers harder to test, version, and maintain independently. +import time from typing import Any, cast from fastapi import APIRouter, Depends, HTTPException, Query @@ -31,7 +32,7 @@ from sqlalchemy.orm import Session from ...core.database import get_db from ...core.logger import belief_scope, logger -from ...core.mapping_service import IdMappingService +from ...core.mapping_service import IdMappingService, record_sync_duration from ...core.migration.dry_run_orchestrator import MigrationDryRunService from ...core.async_superset_client import AsyncSupersetClient from ...dependencies import get_config_manager, get_task_manager, has_permission @@ -400,7 +401,9 @@ async def trigger_sync_now( for env in environments: try: client = AsyncSupersetClient(env) + start = time.monotonic() await service.sync_environment(env.id, client) + record_sync_duration(db, env.id, time.monotonic() - start) results["synced"].append(env.id) logger.reason(f"Synced environment {env.id}", extra={"src": "trigger_sync_now"}) except Exception as e: @@ -417,4 +420,33 @@ async def trigger_sync_now( # #endregion Api.Migration.TriggerSyncNow + +# #region Api.Migration.GetSyncStats [C:3] [TYPE Function] +# @ingroup Api +# @BRIEF Return the real average/last ID-sync duration (seconds) for one environment. +# @PRE env_id provided; requester has READ permission. +# @POST Returns {avg_duration_seconds, last_duration_seconds, run_count} (None when never synced). +# @SIDE_EFFECT Reads the Environment row's sync metrics. +# @RATIONALE The migration wizard surfaces "≈ N sec" as a trustworthy reference before opting +# into a pre-migration ID rescan. +@router.get("/migration/sync-stats", response_model=dict[str, Any]) +async def get_sync_stats( + env_id: str, + db: Session = Depends(get_db), + _=Depends(has_permission("plugin:migration", "READ")), +): + from ...models.mapping import Environment as EnvironmentModel + + env = db.query(EnvironmentModel).filter_by(id=env_id).first() + if env is None: + return {"avg_duration_seconds": None, "last_duration_seconds": None, "run_count": 0} + return { + "avg_duration_seconds": env.sync_duration_avg_seconds, + "last_duration_seconds": env.last_sync_duration_seconds, + "run_count": env.sync_run_count or 0, + } + + +# #endregion Api.Migration.GetSyncStats + # #endregion Api.Migration.MigrationApi diff --git a/backend/src/app.py b/backend/src/app.py index a1ab206a7..b4b936c35 100755 --- a/backend/src/app.py +++ b/backend/src/app.py @@ -996,10 +996,17 @@ async def websocket_endpoint(websocket: WebSocket, task_id: str, source: str = N while True: log_task = asyncio.create_task(log_queue.get()) status_task = asyncio.create_task(status_queue.get()) - done, _ = await asyncio.wait( + done, pending = await asyncio.wait( [log_task, status_task], return_when=asyncio.FIRST_COMPLETED, ) + # @INVARIANT Cancel the not-yet-completed waiter BEFORE the next iteration. + # Without this, each loop leaks a pending queue.get() coroutine that + # consumes the NEXT item (e.g. the terminal task_status event), so the + # completion broadcast is silently swallowed and the client never sees + # the structured result — the result panel stays hidden after a task ends. + for pending_task in pending: + pending_task.cancel() for coro in done: result = coro.result() diff --git a/backend/src/core/database.py b/backend/src/core/database.py index 72a917ac0..aed8af60e 100644 --- a/backend/src/core/database.py +++ b/backend/src/core/database.py @@ -137,7 +137,7 @@ def _ensure_user_dashboard_preferences_columns(bind_engine): if "dashboards_table_density" not in existing_columns: alter_statements.append("ALTER TABLE user_dashboard_preferences ADD COLUMN dashboards_table_density VARCHAR NOT NULL DEFAULT 'comfortable'") if "show_only_slug_dashboards" not in existing_columns: - alter_statements.append("ALTER TABLE user_dashboard_preferences ADD COLUMN show_only_slug_dashboards BOOLEAN NOT NULL DEFAULT TRUE") + alter_statements.append("ALTER TABLE user_dashboard_preferences ADD COLUMN show_only_slug_dashboards BOOLEAN NOT NULL DEFAULT FALSE") if not alter_statements: return diff --git a/backend/src/core/literal_replace.py b/backend/src/core/literal_replace.py new file mode 100644 index 000000000..bc4b74e3e --- /dev/null +++ b/backend/src/core/literal_replace.py @@ -0,0 +1,30 @@ +# #region Core.LiteralReplace.LiteralReplaceModule [C:3] [TYPE Module] [SEMANTICS migration,dataset,literal-replace,sed,transform] +# @defgroup Core Module group. +# +# @BRIEF Deterministic literal substring replacement for dataset mutations ("sed"-style). +# @LAYER Domain +# @PRE `find` is a plain substring (no regex); `replace` is the literal replacement. +# @POST Returns the input with every occurrence of `find` replaced by `replace`. +# @INVARIANT An empty or missing `find` is a no-op — the input is returned unchanged. +# @INVARIANT Replacement is literal: no regex metacharacters are interpreted. +# @RATIONALE A dedicated helper keeps the sed semantics (literal replace) in one place +# shared by the migration engine (ZIP dataset YAMLs) and the live dataset +# mutation endpoint, so both modules behave identically. +# @REJECTED Full `s///` regex semantics rejected — literal replace is deterministic and +# covers the requested use-case (rename a mart name). Regex can be layered on +# later without changing this module's contract. + + +def replace_literals(text: str, find: str | None, replace: str | None) -> str: + """Replace every occurrence of ``find`` in ``text`` with ``replace`` (literal). + + Empty/None ``find`` is a safe no-op so callers never divide-by-zero or + accidentally delete the whole document. + """ + if not text: + return text + if not find: + return text + replacement = replace or "" + return text.replace(find, replacement) +# #endregion Core.LiteralReplace.LiteralReplaceModule diff --git a/backend/src/core/mapping_service.py b/backend/src/core/mapping_service.py index 25389af44..224faad85 100644 --- a/backend/src/core/mapping_service.py +++ b/backend/src/core/mapping_service.py @@ -21,7 +21,7 @@ from sqlalchemy.orm import Session from ss_tools.shared.cot_logger import seed_trace_id from src.core.logger import belief_scope, logger -from src.models.mapping import ResourceMapping, ResourceType +from src.models.mapping import Environment, ResourceMapping, ResourceType # #region Core.MappingService.IdMappingService [C:5] [TYPE Class] @@ -290,4 +290,25 @@ class IdMappingService: return result # #endregion Core.MappingService.GetRemoteIdsBatch # #endregion Core.MappingService.IdMappingService + + +# #region Core.MappingService.RecordSyncDuration [C:3] [TYPE Function] [SEMANTICS mapping,sync,duration,metrics] +# @BRIEF Update an environment's last + running-average ID-sync duration (seconds). +# @PRE environment row exists in the DB; duration_seconds >= 0. +# @POST Environment.last/avg_sync_duration_seconds and sync_run_count are updated and committed. +# @SIDE_EFFECT Commits the current session. +# @INVARIANT Running average is cumulative: avg_new = (avg_old * n + d) / (n + 1). +# @RATIONALE Surfacing the real average sync time lets the migration UI show a trustworthy +# "≈ N sec" reference before opting into a pre-migration rescan. +def record_sync_duration(db: Session, environment_id: str, duration_seconds: float) -> None: + env = db.query(Environment).filter_by(id=environment_id).first() + if env is None: + return + n = env.sync_run_count or 0 + prev_avg = env.sync_duration_avg_seconds or 0.0 + env.last_sync_duration_seconds = round(duration_seconds, 2) + env.sync_duration_avg_seconds = round((prev_avg * n + duration_seconds) / (n + 1), 2) + env.sync_run_count = n + 1 + db.commit() +# #endregion Core.MappingService.RecordSyncDuration # #endregion Core.MappingService.IdMappingServiceModule diff --git a/backend/src/core/migration_engine.py b/backend/src/core/migration_engine.py index 8873b0867..2c7542c74 100644 --- a/backend/src/core/migration_engine.py +++ b/backend/src/core/migration_engine.py @@ -28,6 +28,7 @@ from src.core.mapping_service import IdMappingService from src.models.mapping import ResourceType from .logger import belief_scope, logger +from .literal_replace import replace_literals # #region Core.MigrationEngine [TYPE Class] @@ -71,6 +72,8 @@ class MigrationEngine: strip_databases: bool = True, target_env_id: str | None = None, fix_cross_filters: bool = False, + literal_find: str | None = None, + literal_replace: str | None = None, ) -> bool: """ Transform a Superset export ZIP by replacing database UUIDs and optionally fixing cross-filters. @@ -121,6 +124,16 @@ class MigrationEngine: ) for ds_file in dataset_files: self._transform_yaml(ds_file, db_mapping) + # 2.05 Apply literal ("sed") replacement to dataset YAML text + if literal_find: + logger.reason( + f"Applying literal replacement to {len(dataset_files)} dataset YAMLs", + payload={"find": literal_find, "replace": literal_replace or ""}, + ) + for ds_file in dataset_files: + self._apply_literal_replace( + ds_file, literal_find, literal_replace or "" + ) # 2.1 Transform YAMLs (Databases — replace UUID with target UUID) # When a database UUID in the archive matches a target UUID that exists # in the target Superset, Superset's import_database() will find it, @@ -316,6 +329,26 @@ class MigrationEngine: return database_uuids # #endregion Core.MigrationEngine.CollectDatasetDatabaseUuids + # #region Core.MigrationEngine.ApplyLiteralReplace [C:2] [TYPE Function] [SEMANTICS migration,dataset,literal-replace,sed] + # @BRIEF Apply a literal find→replace to a dataset YAML file's raw text. + # @PRE file_path points to a readable/writable UTF-8 dataset YAML. + # @POST The file is rewritten in place only when the replacement changed its content. + # @SIDE_EFFECT Conditionally rewrites the dataset YAML on disk. + # @INVARIANT Operates on raw text (never re-serialises YAML), so formatting and + # comments outside the replaced substring are preserved — true "sed" semantics. + def _apply_literal_replace(self, file_path: Path, find: str, replace: str) -> None: + with open(file_path, encoding="utf-8") as stream: + original = stream.read() + updated = replace_literals(original, find, replace) + if updated != original: + with open(file_path, "w", encoding="utf-8") as stream: + stream.write(updated) + logger.reason( + "Literal replacement applied to dataset YAML", + payload={"file": file_path.name, "find": find}, + ) + # #endregion Core.MigrationEngine.ApplyLiteralReplace + # #region Core.MigrationEngine.DatabaseYamlUuid [C:1] [TYPE Function] [SEMANTICS migration,archive,database] # @BRIEF Return the UUID declared by an exported database resource. # @POST Returns None for empty or malformed database YAML without making it importable. diff --git a/backend/src/models/dashboard.py b/backend/src/models/dashboard.py index 55ee54b09..b9fbff2dc 100644 --- a/backend/src/models/dashboard.py +++ b/backend/src/models/dashboard.py @@ -27,8 +27,11 @@ class DashboardSelection(BaseModel): target_env_id: str replace_db_config: bool = False fix_cross_filters: bool = True + rescan_ids_before_migration: bool = False composite_key_mutation_server: Literal['target', 'source'] = 'target' sync_dataset_composite_keys: bool = True + literal_find: str | None = Field(None, max_length=10000) + literal_replace: str | None = Field(None, max_length=10000) # #endregion Models.Dashboard.DashboardSelection diff --git a/backend/src/models/mapping.py b/backend/src/models/mapping.py index b22dc906c..185d2baee 100644 --- a/backend/src/models/mapping.py +++ b/backend/src/models/mapping.py @@ -23,7 +23,7 @@ import enum import uuid -from sqlalchemy import Boolean, Column, DateTime, Enum as SQLEnum, ForeignKey, String +from sqlalchemy import Boolean, Column, DateTime, Enum as SQLEnum, Float, ForeignKey, Integer, String from sqlalchemy.orm import declarative_base from sqlalchemy.sql import func @@ -62,6 +62,10 @@ class Environment(Base): name = Column(String, nullable=False) url = Column(String, nullable=False) credentials_id = Column(String, nullable=False) + # ID-mapping sync metrics (chart/dataset/dashboard ResourceMapping refresh). + last_sync_duration_seconds = Column(Float, nullable=True) + sync_duration_avg_seconds = Column(Float, nullable=True) + sync_run_count = Column(Integer, nullable=False, server_default="0") # #endregion Models.Mapping.Environment # #region Models.Mapping.DatabaseMapping [C:3] [TYPE Class] diff --git a/backend/src/models/profile.py b/backend/src/models/profile.py index a3e4542ea..7287dca61 100644 --- a/backend/src/models/profile.py +++ b/backend/src/models/profile.py @@ -44,7 +44,7 @@ class UserDashboardPreference(Base): superset_username_normalized = Column(String, nullable=True, index=True) show_only_my_dashboards = Column(Boolean, nullable=False, default=False) - show_only_slug_dashboards = Column(Boolean, nullable=False, default=True) + show_only_slug_dashboards = Column(Boolean, nullable=False, default=False) git_username = Column(String, nullable=True) git_email = Column(String, nullable=True) diff --git a/backend/src/plugins/migration.py b/backend/src/plugins/migration.py index cf639950f..48d08eddc 100755 --- a/backend/src/plugins/migration.py +++ b/backend/src/plugins/migration.py @@ -25,11 +25,12 @@ from pathlib import Path import re +import time from typing import Any from ..core.database import SessionLocal from ..core.logger import belief_scope, logger as app_logger -from ..core.mapping_service import IdMappingService +from ..core.mapping_service import IdMappingService, record_sync_duration from ..core.migration.dataset_key_sync import ( read_live_dataset_contracts, sync_dataset_composite_keys, @@ -373,7 +374,7 @@ class MigrationPlugin(PluginBase): "composite_key_mutation_server": { "type": "string", "title": "Composite Key Mutation Server", - "description": "Which server to mutate on non-password import failure fallback: target syncs the transformed archive onto the target; source aligns source datasets with the target live keys, re-exports and re-transforms.", + "description": "Which server to mutate on non-password import failure fallback: target updates dataset composite keys on the target via API (no archive rebuild) and retries; source aligns source datasets with the target live keys, re-exports and re-transforms.", "enum": ["target", "source"], "default": "target", }, @@ -422,6 +423,10 @@ class MigrationPlugin(PluginBase): if composite_key_mutation_server not in ("target", "source"): composite_key_mutation_server = "target" + literal_find = params.get("literal_find") or None + literal_replace = params.get("literal_replace") or None + rescan_ids_before_migration = params.get("rescan_ids_before_migration", False) + task_id = params.get("_task_id") from ..dependencies import get_task_manager tm = get_task_manager() @@ -521,6 +526,25 @@ class MigrationPlugin(PluginBase): mapping_service=IdMappingService(engine_db) ) + # Optional pre-migration ID rescan: refresh target chart/dataset/dashboard + # ResourceMapping so cross-filter patching resolves fresh target IDs. + if fix_cross_filters and rescan_ids_before_migration: + app_logger.reason( + "Pre-migration ID rescan requested", + extra={"target_env_id": target_env_id}, + ) + rescan_start = time.monotonic() + try: + await IdMappingService(engine_db).sync_environment(target_env_id, to_c) + record_sync_duration(engine_db, target_env_id, time.monotonic() - rescan_start) + app_logger.reflect("Pre-migration ID rescan complete") + except Exception as rescan_exc: + app_logger.explore( + "Pre-migration ID rescan failed; continuing with existing mappings", + extra={"target_env_id": target_env_id}, + error=str(rescan_exc), + ) + # Migration Loop — per-dashboard isolation (never re-raise into fatal task failure) for dash in dashboards_to_migrate: dash_id, dash_slug, title = dash["id"], dash.get("slug"), dash["dashboard_title"] @@ -546,7 +570,9 @@ class MigrationPlugin(PluginBase): db_mapping, strip_databases=False, target_env_id=tgt_env.id if tgt_env else None, - fix_cross_filters=fix_cross_filters + fix_cross_filters=fix_cross_filters, + literal_find=literal_find, + literal_replace=literal_replace, ) if not success and replace_db_config: @@ -573,7 +599,9 @@ class MigrationPlugin(PluginBase): db_mapping, strip_databases=False, target_env_id=tgt_env.id if tgt_env else None, - fix_cross_filters=fix_cross_filters + fix_cross_filters=fix_cross_filters, + literal_find=literal_find, + literal_replace=literal_replace, ) if not success: diff --git a/backend/src/plugins/translate/__tests__/test_batch_insert.py b/backend/src/plugins/translate/__tests__/test_batch_insert.py index 293308f1a..5d17403a7 100644 --- a/backend/src/plugins/translate/__tests__/test_batch_insert.py +++ b/backend/src/plugins/translate/__tests__/test_batch_insert.py @@ -2,14 +2,16 @@ # @BRIEF Tests for _batch_insert.py — column building, row building, context keys. # @RELATION BINDS_TO -> [_batch_insert_module] import pytest -from unittest.mock import MagicMock +from unittest.mock import AsyncMock, MagicMock import uuid +from src.core.db_executor import DbExecutionResult, DbExecutor from src.models.translate import TranslationJob, TranslationLanguage, TranslationRecord from src.plugins.translate._batch_insert import ( _build_context_keys, _build_insert_rows, _build_target_columns, + _execute_insert_sql, _fetch_batch_records, _generate_insert_sql, ) @@ -374,4 +376,43 @@ class TestGenerateInsertSQL: assert sql is not None assert count == 1 # #endregion Test.Tests.TestGenerateInsertSQL + + +# #region Test.Tests.TestExecuteInsertSql [C:3] [TYPE Class] [SEMANTICS test,translate,insert,failure] +# @BRIEF Verify _execute_insert_sql reports failures instead of swallowing them. +# @TEST_INVARIANT insert-failure-is-signalled -> VERIFIED_BY: [ +# test_direct_db_failure_returns_false, +# test_direct_db_success_returns_true, +# test_superset_timeout_returns_false, +# test_submission_exception_returns_false, +# ] +class TestExecuteInsertSql: + """Verify _execute_insert_sql returns a success bool instead of hiding failures.""" + + async def test_direct_db_failure_returns_false(self) -> None: + executor = DbExecutor(MagicMock()) + executor.execute_sql = AsyncMock(return_value=DbExecutionResult( + success=False, error="No route to host", + )) + ok = await _execute_insert_sql(executor, "INSERT ...", "batch-1", 13, connection_id="conn-1") + assert ok is False + + async def test_direct_db_success_returns_true(self) -> None: + executor = DbExecutor(MagicMock()) + executor.execute_sql = AsyncMock(return_value=DbExecutionResult(success=True, rows_affected=13)) + ok = await _execute_insert_sql(executor, "INSERT ...", "batch-1", 13, connection_id="conn-1") + assert ok is True + + async def test_superset_timeout_returns_false(self) -> None: + executor = MagicMock() + executor.execute_and_poll = AsyncMock(return_value={"status": "timeout"}) + ok = await _execute_insert_sql(executor, "INSERT ...", "batch-1", 13) + assert ok is False + + async def test_submission_exception_returns_false(self) -> None: + executor = DbExecutor(MagicMock()) + executor.execute_sql = AsyncMock(side_effect=RuntimeError("boom")) + ok = await _execute_insert_sql(executor, "INSERT ...", "batch-1", 13, connection_id="conn-1") + assert ok is False +# #endregion Test.Tests.TestExecuteInsertSql # #endregion Test.Tests.TestBatchInsert diff --git a/backend/src/plugins/translate/__tests__/test_executor.py b/backend/src/plugins/translate/__tests__/test_executor.py index d884570a4..35efd9566 100644 --- a/backend/src/plugins/translate/__tests__/test_executor.py +++ b/backend/src/plugins/translate/__tests__/test_executor.py @@ -709,7 +709,7 @@ class TestAutoSizeBatches: # @TEST_INVARIANT Pure-skip (s=0, f=0, sk>0) → COMPLETED (not FAILED). This was the root cause of "status: FAILED" on every scheduled translation run in the old prod logs. # @TEST_EDGE missing_success_zero_failure_pure_skip -> Must be COMPLETED (the prod bug case with 8 skipped rows) # @TEST_EDGE pure_failure -> Must be FAILED with clear error message -# @TEST_EDGE any_success -> Always COMPLETED even if failures/skips present +# @TEST_EDGE any_failure -> Must be FAILED even with some successes (signals LLM unavailability) # @TEST_EDGE empty -> COMPLETED # #endregion Test.Executor.FinalizeRun @@ -756,11 +756,36 @@ class TestFinalizeRun: assert result.status == "FAILED" assert "All 4 record(s) failed" in (result.error_message or "") - def test_mixed_fails_with_some_success_is_completed(self) -> None: - """Partial success should not mark whole run FAILED.""" + def test_mixed_fails_with_some_success_is_failed(self) -> None: + """Partial failure (e.g. LLM unavailable with some cache hits) must signal FAILED.""" run = self._make_run() result = TranslationExecutor._finalize_run(run, s=2, f=3, sk=1) - assert result.status == "COMPLETED" + assert result.status == "FAILED" + assert "3 record(s) failed" in (result.error_message or "") + + def test_compute_final_status_partial_failure(self) -> None: + """The pure decision function marks any failure (with or without success) as FAILED.""" + assert TranslationExecutor._compute_final_status(2, 3, 1) == ( + "FAILED", + "3 record(s) failed during translation (successful=2) — see individual record errors for details", + ) + assert TranslationExecutor._compute_final_status(0, 4, 0) == ( + "FAILED", + "All 4 record(s) failed — see individual record errors for details", + ) + assert TranslationExecutor._compute_final_status(0, 0, 8) == ("COMPLETED", None) + assert TranslationExecutor._compute_final_status(5, 0, 0) == ("COMPLETED", None) + + def test_apply_insert_failure_marks_run_failed(self) -> None: + """Insert failures surface as run FAILED + insert_status=failed.""" + run = self._make_run() + result = TranslationExecutor._apply_insert_failure( + run, [("batch-1", "No route to host"), ("batch-2", "connection refused")], + ) + assert result.status == "FAILED" + assert result.insert_status == "failed" + assert "No route to host" in (result.error_message or "") + assert "batch-1" in (result.error_message or "") def test_full_process_batches_path_with_only_skips(self, mock_job: MagicMock, mock_run: MagicMock) -> None: """Orthogonal integration test: exercises _process_batches -> _finalize_run with pure-skip batch results. diff --git a/backend/src/plugins/translate/_batch_insert.py b/backend/src/plugins/translate/_batch_insert.py index ff6d65b5d..364b0a9d4 100644 --- a/backend/src/plugins/translate/_batch_insert.py +++ b/backend/src/plugins/translate/_batch_insert.py @@ -36,7 +36,11 @@ async def insert_batch_to_target( db: Session, config_manager: ConfigManager, job: TranslationJob, batch_id: str, run_id: str, ) -> None: - """Insert successful batch records into the target table via Superset SQL Lab.""" + """Insert successful batch records into the target table via Superset SQL Lab or direct DB. + + Raises RuntimeError if the insert backend cannot be resolved or any chunk fails — + an unreachable target DB is an error, never a silent success. + """ with belief_scope("BatchInsertService.insert_batch_to_target"): records = _fetch_batch_records(db, batch_id) if not records: @@ -61,7 +65,11 @@ async def insert_batch_to_target( dialect, executor, connection_id = await _resolve_insert_backend(config_manager, job, batch_id) if dialect is None: - return + # Backend resolution failed (e.g. Superset SQL Lab unreachable) — surface as an error, + # not a silent skip, so the run is marked FAILED rather than reported as inserted. + raise RuntimeError( + f"Batch {batch_id[:12]} insert failed: could not resolve insert backend/dialect" + ) # Deduplicate by key columns for MERGE strategy to prevent PostgreSQL # "ON CONFLICT DO UPDATE command cannot affect row a second time" error. @@ -88,16 +96,38 @@ async def insert_batch_to_target( column_types=getattr(job, "target_column_types", None), ) total_inserted = 0 + failed_chunks = 0 prepared = len(rows_for_sql) for sql, chunk_count in statements: if sql is None: continue - await _execute_insert_sql(executor, sql, batch_id, chunk_count, connection_id=connection_id) - total_inserted += chunk_count - logger.reason(f"Batch {batch_id[:12]} inserted {total_inserted} rows", - {"batch_id": batch_id, "rows": total_inserted, "chunks": len(statements)}) + ok = await _execute_insert_sql(executor, sql, batch_id, chunk_count, connection_id=connection_id) + if ok: + total_inserted += chunk_count + else: + failed_chunks += 1 # Live run-level insert counters for WS progress (best-effort; ClickHouse may report 0 affected). _bump_run_insert_counters(db, run_id, prepared=prepared, affected=total_inserted) + + if failed_chunks: + # An unreachable target DB or failed INSERT must surface as an error, never as success. + logger.explore( + f"Batch {batch_id[:12]} insert failed", + { + "batch_id": batch_id, + "rows_inserted": total_inserted, + "failed_chunks": failed_chunks, + "total_chunks": len(statements), + "prepared": prepared, + }, + error=f"{failed_chunks} chunk(s) failed to insert into target", + ) + raise RuntimeError( + f"Batch {batch_id[:12]} insert failed: " + f"{failed_chunks}/{len(statements)} chunk(s) could not be inserted into the target" + ) + logger.reason(f"Batch {batch_id[:12]} inserted {total_inserted} rows", + {"batch_id": batch_id, "rows": total_inserted, "chunks": len(statements)}) # #endregion Plugin.BatchInsert.InsertBatchToTarget @@ -327,15 +357,17 @@ def _generate_insert_sql( # @ingroup Translate # @BRIEF Execute INSERT SQL via either Superset SQL Lab or direct DB (DbExecutor). # @PRE executor is a valid initialized executor. sql is dialect-appropriate. -# @POST SQL executed. Status logged (non-fatal — batch processing continues). +# @POST Returns True on a confirmed successful insert, False on any failure. # @SIDE_EFFECT Writes to target DB or calls Superset API. # @RATIONALE Branch on executor type: DbExecutor uses execute_sql(connection_id), # SupersetSqlLabExecutor uses execute_and_poll(). connection_id discriminates the path. +# Failures are reported (not swallowed) so the caller can surface them as a run-level error. async def _execute_insert_sql( executor: SupersetSqlLabExecutor | DbExecutor, sql: str, batch_id: str, row_count: int, connection_id: str | None = None, -) -> None: - """Execute the INSERT SQL via Superset SQL Lab or direct DB.""" +) -> bool: + """Execute the INSERT SQL via Superset SQL Lab or direct DB. Returns True on success.""" + status = "unknown" try: if isinstance(executor, DbExecutor): result = await executor.execute_sql(connection_id, sql) # type: ignore[arg-type] @@ -344,14 +376,22 @@ async def _execute_insert_sql( logger.explore("Direct DB insert failed for batch", {"batch_id": batch_id, "error": result.error}, error=result.error or "Unknown DB error") + return False else: # Superset SQL Lab path result = await executor.execute_and_poll(sql=sql, max_polls=30, poll_interval_seconds=2.0) status = result.get("status", "unknown") + if status in ("failed", "timeout", "error"): + err = result.get("error") or result.get("error_message") or status + logger.explore("Superset SQL insert failed for batch", + {"batch_id": batch_id, "status": status, "error": err}, + error=str(err)) + return False except Exception as e: - logger.explore("SQL submission failed for batch", {"batch_id": batch_id, "error": str(e)}) - return + logger.explore("SQL submission failed for batch", {"batch_id": batch_id, "error": str(e)}, error=str(e)) + return False logger.reason(f"Chunk inserted for batch {batch_id[:12]}", {"batch_id": batch_id, "rows": row_count, "status": status}) + return True # #endregion Plugin.BatchInsert.ExecuteInsertSql # #endregion Plugin.BatchInsert.BatchInsertService diff --git a/backend/src/plugins/translate/executor.py b/backend/src/plugins/translate/executor.py index 6a9fd25f5..a96651034 100644 --- a/backend/src/plugins/translate/executor.py +++ b/backend/src/plugins/translate/executor.py @@ -226,15 +226,18 @@ class TranslationExecutor: # region Plugin.Translate.StartInsertWorkers [C:3] [TYPE Function] [SEMANTICS translate,insert,queue] # @ingroup Translate # @BRIEF Start insert workers that consume batch_ids from a queue (decoupled from LLM). - # @POST Returns (queue, worker_tasks). Put None per worker to shut down. + # @POST Returns (queue, worker_tasks, insert_failures). Put None per worker to shut down. + # insert_failures collects (batch_id, error) tuples so the run can be marked FAILED + # when the target insert fails — an insert failure is an error, not a silent no-op. # @SIDE_EFFECT Background tasks open SessionLocal per insert. def _start_insert_workers( self, job: TranslationJob, run_id: str, n_workers: int, - ) -> tuple[asyncio.Queue, list[asyncio.Task]]: + ) -> tuple[asyncio.Queue, list[asyncio.Task], list[tuple[str, str]]]: from ...core.database import SessionLocal from ._batch_proc import BatchProcessingService q: asyncio.Queue = asyncio.Queue() + insert_failures: list[tuple[str, str]] = [] async def _worker(worker_id: int) -> None: while True: @@ -249,10 +252,11 @@ class TranslationExecutor: db.commit() except Exception as e: logger.explore( - "Insert worker failed (non-fatal)", + "Insert worker failed", {"batch_id": batch_id, "worker_id": worker_id}, error=str(e), ) + insert_failures.append((str(batch_id), str(e))) try: db.rollback() except Exception: @@ -267,9 +271,28 @@ class TranslationExecutor: "Insert workers started", {"run_id": run_id, "workers": len(tasks)}, ) - return q, tasks + return q, tasks, insert_failures # endregion Plugin.Translate.StartInsertWorkers + # region Plugin.Translate.ApplyInsertFailure [C:2] [TYPE Function] [SEMANTICS translate,insert,failure,status] + # @ingroup Translate + # @BRIEF Mark a run FAILED with insert_status=failed after target insert failures. + # @PRE insert_failures is a non-empty list of (batch_id, error) tuples. + # @POST run.status = FAILED, run.insert_status = failed, run.error_message populated. + # @SIDE_EFFECT Mutates the passed run object in place (caller commits). + @staticmethod + def _apply_insert_failure(run: TranslationRun, insert_failures: list[tuple[str, str]]) -> TranslationRun: + shown = insert_failures[:5] + detail = "; ".join(f"batch {bid}: {err}" for bid, err in shown) + if len(insert_failures) > 5: + detail += f" (+{len(insert_failures) - 5} more)" + run.insert_status = "failed" + run.status = "FAILED" + run.error_message = f"Target insert failed: {detail}" + run.completed_at = datetime.now(UTC) + return run + # endregion Plugin.Translate.ApplyInsertFailure + # region Plugin.Translate.ProcessBatchesSerial [C:3] [TYPE Function] [SEMANTICS translate,batch,serial] # @ingroup Translate # @BRIEF Process all batches sequentially; inserts go through async queue (non-blocking). @@ -284,7 +307,7 @@ class TranslationExecutor: language_stats_map: dict[str, TranslationRunLanguageStats] | None = None, ) -> TranslationRun: insert_n = self._resolve_insert_concurrency(job) - insert_q, insert_tasks = self._start_insert_workers(job, run.id, insert_n) + insert_q, insert_tasks, insert_failures = self._start_insert_workers(job, run.id, insert_n) successful_records = failed_records = skipped_records = cache_hits = 0 translated_records = same_language_skipped_records = 0 @@ -333,13 +356,17 @@ class TranslationExecutor: ) if self.on_batch_progress: self.on_batch_progress(run.id, batch_idx + 1, len(batches), successful_records, run.total_records or 0) - - return self._finalize_run(run, successful_records, failed_records, skipped_records) finally: for _ in insert_tasks: await insert_q.put(None) await insert_q.join() await asyncio.gather(*insert_tasks, return_exceptions=True) + + if insert_failures: + self._apply_insert_failure(run, insert_failures) + self.db.commit() + return run + return self._finalize_run(run, successful_records, failed_records, skipped_records) # endregion Plugin.Translate.ProcessBatchesSerial # region Plugin.Translate.ProcessBatchesParallel [C:4] [TYPE Function] [SEMANTICS translate,batch,parallel,concurrency] @@ -373,7 +400,7 @@ class TranslationExecutor: return self._finalize_run(run, 0, 0, 0) insert_n = self._resolve_insert_concurrency(job) - insert_q, insert_tasks = self._start_insert_workers(job, run.id, insert_n) + insert_q, insert_tasks, insert_failures = self._start_insert_workers(job, run.id, insert_n) sem = asyncio.Semaphore(concurrency) lock = asyncio.Lock() @@ -501,6 +528,11 @@ class TranslationExecutor: ) return run + if insert_failures: + self._apply_insert_failure(run, insert_failures) + self.db.commit() + return run + logger.reflect( "Parallel batch processing complete", { @@ -524,9 +556,12 @@ class TranslationExecutor: # @POST run.status is either "COMPLETED" or "FAILED". completed_at is always set. # @SIDE_EFFECT Mutates the passed run object (status + error_message + completed_at). # @INVARIANT Pure-skip (s=0, f=0, sk>0) and empty runs are treated as successful no-op (COMPLETED). - # Only (s==0 AND f>0) is FAILED. This prevents scheduled runs that found no new keys from being marked FAILED. + # Any failed record (f>0) is FAILED — an LLM unavailability or provider error must be + # surfaced as an error even when other rows succeeded from cache. This prevents a run + # with an unreachable LLM from being reported as COMPLETED because some rows hit cache. # @DATA_CONTRACT (s: int, f: int, sk: int) -> run.status in {"COMPLETED", "FAILED"} # @RATIONALE The previous elif s == 0 unconditionally marked any zero-success run as FAILED, even when all work was legitimately skipped (new_key_only scheduled runs with 8 skipped rows). This exactly matched the prod symptom "status: FAILED, successful:0, failed:0, skipped:8". + # Separately, f>0 with s>0 (e.g. LLM down but some cache hits) was silently COMPLETED — that hides LLM unavailability from operators. # @REJECTED Treating sk>0 + s==0 as failure — it would turn healthy "nothing to do" scheduled runs into red alerts and pollute history. # @TEST_INVARIANT Pure-skip scheduled translation must end COMPLETED (see Test.Executor.FinalizeRun). # @RELATION BINDS_TO -> [Test.Executor.FinalizeRun] @@ -547,14 +582,20 @@ class TranslationExecutor: # @PRE Counts are non-negative. # @POST Returns (status, error_message_or_None) # @DATA_CONTRACT (successful: int, failed: int, skipped: int) -> (str, Optional[str]) - # @INVARIANT Mirrors the logic in _finalize_run and is the single source of truth for status decision. + # @INVARIANT Pure failure (no successes and some failures) is FAILED; any failure alongside + # successes is also FAILED (signals LLM unavailability). Only zero failures completes. # @RELATION CALLED_BY -> [TranslationExecutor._finalize_run] # @RELATION CALLED_BY -> [TranslationRunRetryManager.retry_failed_batches] (via alignment) @staticmethod def _compute_final_status(successful: int, failed: int, skipped: int) -> tuple[str, str | None]: - """Pure function: only pure failure (no successes and some failures) is FAILED.""" - if successful == 0 and failed > 0: - return "FAILED", f"All {failed} record(s) failed — see individual record errors for details" + """Pure function: any failed record signals an error (e.g. LLM unavailable), even with partial success.""" + if failed > 0: + if successful == 0: + return "FAILED", f"All {failed} record(s) failed — see individual record errors for details" + return "FAILED", ( + f"{failed} record(s) failed during translation " + f"(successful={successful}) — see individual record errors for details" + ) return "COMPLETED", None # #endregion TranslationExecutor._compute_final_status diff --git a/backend/src/plugins/translate/orchestrator_retry.py b/backend/src/plugins/translate/orchestrator_retry.py index 37e218ac9..1968f8244 100644 --- a/backend/src/plugins/translate/orchestrator_retry.py +++ b/backend/src/plugins/translate/orchestrator_retry.py @@ -36,7 +36,7 @@ from .orchestrator_cancel import cancel_run as _cancel_run, retry_insert as _ret # @PRE db, config, event_log provided. # @POST Run status updated after retry. # @SIDE_EFFECT DB state changes, events. -# @INVARIANT Status decision uses same logic as main executor (pure failure only -> FAILED). +# @INVARIANT Status decision uses same logic as main executor (any failed record -> FAILED). class TranslationRunRetryManager: """Manage retry and cancellation of translation runs.""" @@ -132,15 +132,23 @@ class TranslationRunRetryManager: # #region TranslationRunRetryManager._finalize_retry_status [C:2] [TYPE Block] [SEMANTICS translate,retry,status] # @BRIEF Final status decision after retrying failed batches. Aligned with the canonical rule - # from TranslationExecutor._compute_final_status. - # @INVARIANT Same predicate as main finalize: pure failure (no successes + remaining failures) is FAILED. + # from TranslationExecutor._compute_final_status (any failed record -> FAILED). + # @INVARIANT Same predicate as main finalize: any failed record (with or without successes) is FAILED. # @RELATION BINDS_TO -> [TranslationExecutor._compute_final_status] - # Align with main path to prevent the same "pure-skip reported as FAILED" class of bug. - run.status, run.error_message = ( - ("FAILED", f"All {run.failed_records} record(s) failed after retry") - if (run.successful_records or 0) == 0 and (run.failed_records or 0) > 0 - else ("COMPLETED", None) - ) + # Align with main path to prevent the same "pure-skip reported as FAILED" class of bug, + # while still surfacing an unreachable LLM as FAILED even when some rows succeeded from cache. + successful = run.successful_records or 0 + failed = run.failed_records or 0 + if failed > 0: + run.status = "FAILED" + run.error_message = ( + f"All {failed} record(s) failed after retry" + if successful == 0 + else f"{failed} record(s) failed after retry (successful={successful})" + ) + else: + run.status = "COMPLETED" + run.error_message = None run.completed_at = datetime.now(UTC) self.db.flush() # #endregion TranslationRunRetryManager._finalize_retry_status diff --git a/backend/src/schemas/profile.py b/backend/src/schemas/profile.py index 1574bdcff..638641556 100644 --- a/backend/src/schemas/profile.py +++ b/backend/src/schemas/profile.py @@ -51,7 +51,7 @@ class ProfilePreference(BaseModel): superset_username: str | None = None superset_username_normalized: str | None = None show_only_my_dashboards: bool = False - show_only_slug_dashboards: bool = True + show_only_slug_dashboards: bool = False git_username: str | None = None git_email: str | None = None @@ -90,7 +90,7 @@ class ProfilePreferenceUpdateRequest(BaseModel): ) show_only_slug_dashboards: bool | None = Field( default=None, - description='When true, "/dashboards" hides dashboards without slug by default.', + description='When true, "/dashboards" hides dashboards without a URL slug (opt-in; off by default).', ) git_username: str | None = Field( default=None, diff --git a/backend/src/services/mapping_analysis.py b/backend/src/services/mapping_analysis.py index a1ba50d60..a4ec1937c 100644 --- a/backend/src/services/mapping_analysis.py +++ b/backend/src/services/mapping_analysis.py @@ -15,6 +15,7 @@ from typing import Any from src.core.logger import belief_scope, logger from src.core.superset_client import SupersetClient from src.core.utils.matching import suggest_mappings +from src.core.utils.network import NetworkError from src.models.mapping import DatabaseMapping @@ -22,10 +23,21 @@ from src.models.mapping import DatabaseMapping # @ingroup Mapping # @BRIEF Read one environment database catalog and always close its client. # @POST Returns (catalog, None) or ([], error); never raises a catalog failure. +# @INVARIANT The returned error is a short human-readable string (not the raw exception repr), +# so the coverage UI can show "env: Superset unavailable (502)" instead of a stack blob. async def _fetch_catalog(environment: Any) -> tuple[list[dict[str, Any]], str | None]: client = SupersetClient(environment) try: return await client.get_databases_summary(), None + except NetworkError as exc: + status_code = exc.context.get("status_code") if getattr(exc, "context", None) else None + message = f"Superset unavailable (Status {status_code})" if status_code else "Superset unreachable" + logger.explore( + "Mapping coverage catalog unavailable", + payload={"environment_id": environment.id, "status_code": status_code}, + error=str(exc), + ) + return [], message except Exception as exc: logger.explore( "Mapping coverage catalog unavailable", diff --git a/backend/src/services/profile_preference_service.py b/backend/src/services/profile_preference_service.py index a1674aae1..403b5da84 100644 --- a/backend/src/services/profile_preference_service.py +++ b/backend/src/services/profile_preference_service.py @@ -125,7 +125,7 @@ class ProfilePreferenceService: "superset_username": None, "superset_username_normalized": None, "show_only_my_dashboards": False, - "show_only_slug_dashboards": True, + "show_only_slug_dashboards": False, } return { @@ -139,7 +139,7 @@ class ProfilePreferenceService: "show_only_slug_dashboards": bool( preference.show_only_slug_dashboards if preference.show_only_slug_dashboards is not None - else True + else False ), } # #endregion Services.ProfilePreferenceService.GetDashboardFilterBinding diff --git a/backend/src/services/profile_utils.py b/backend/src/services/profile_utils.py index 6d1e2dd9d..5afea1bec 100644 --- a/backend/src/services/profile_utils.py +++ b/backend/src/services/profile_utils.py @@ -193,7 +193,7 @@ def build_default_preference(user_id: str) -> Any: superset_username=None, superset_username_normalized=None, show_only_my_dashboards=False, - show_only_slug_dashboards=True, + show_only_slug_dashboards=False, git_username=None, git_email=None, has_git_personal_access_token=False, diff --git a/backend/tests/api/test_mappings.py b/backend/tests/api/test_mappings.py index ff1651da5..a9030b622 100644 --- a/backend/tests/api/test_mappings.py +++ b/backend/tests/api/test_mappings.py @@ -219,6 +219,24 @@ class TestMappingAnalysis: assert healthy["status"] != "error" assert failed["status"] == "error" assert failed["errors"] == {"env-2": "Superset unavailable"} + + async def test_network_error_reports_short_status_message(self): + """NetworkError (e.g. 502 during Superset auth) → concise per-env message, not a stack blob.""" + from src.core.utils.network import NetworkError + from src.services.mapping_analysis import _fetch_catalog + + env = SimpleNamespace(id="ss-preprod", name="Preprod") + client = MagicMock() + client.get_databases_summary = AsyncMock( + side_effect=NetworkError("Environment unavailable during authentication (Status 502)", status_code=502) + ) + client.aclose = AsyncMock() + + with patch("src.services.mapping_analysis.SupersetClient", return_value=client): + catalog, error = await _fetch_catalog(env) + + assert catalog == [] + assert error == "Superset unavailable (Status 502)" # #endregion Test.Api.MappingAnalysis.PartialFailure # #region Test.Api.MappingAnalysis.ApiKeyScope [C:3] [TYPE Function] [SEMANTICS test,mappings,analysis,auth] diff --git a/backend/tests/core/test_migration_engine.py b/backend/tests/core/test_migration_engine.py index 388c0da8a..8f6a3ce73 100644 --- a/backend/tests/core/test_migration_engine.py +++ b/backend/tests/core/test_migration_engine.py @@ -331,6 +331,63 @@ def test_transform_zip_end_to_end(): # #endregion Test.MigrationEngine.TestTransformZipEndToEnd +# #region Test.MigrationEngine.TestTransformZipAppliesLiteralReplace [C:2] [TYPE Function] +# @BRIEF Literal find→replace mutates dataset YAML text (mart rename) without touching unrelated files. +def test_transform_zip_applies_literal_replace(): + """Verifies literal_find/literal_replace rewrites dataset YAML raw text in the archive.""" + 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 / "datasets" / "mart.yaml").write_text( + "table_name: dm_view.account_debt\nsql: SELECT * FROM dm_view.account_debt\n" + ) + with zipfile.ZipFile(source, "w") as zf: + for path in archive.rglob("*.yaml"): + zf.write(path, path.relative_to(archive)) + + ok = engine.transform_zip( + str(source), + str(target), + {}, + literal_find="dm_view.account_debt", + literal_replace="dm_view.account_debt_final", + ) + assert ok is True + with zipfile.ZipFile(target) as zf: + text = zf.read("datasets/mart.yaml").decode("utf-8") + assert text == ( + "table_name: dm_view.account_debt_final\n" + "sql: SELECT * FROM dm_view.account_debt_final\n" + ) + + +def test_transform_zip_literal_replace_noop_when_no_find(): + """Empty literal_find leaves dataset YAML unchanged.""" + 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 / "datasets" / "mart.yaml").write_text("table_name: dm_view.account_debt\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), {}) + with zipfile.ZipFile(target) as zf: + text = zf.read("datasets/mart.yaml").decode("utf-8") + assert "dm_view.account_debt" in text + + +# #endregion Test.MigrationEngine.TestTransformZipAppliesLiteralReplace + + # #region Test.MigrationEngine.TestTransformZipKeepsOnlyReferencedMappedDatabase [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. diff --git a/backend/tests/plugins/translate/test_executor.py b/backend/tests/plugins/translate/test_executor.py index 0d67d2d32..f9938ad83 100644 --- a/backend/tests/plugins/translate/test_executor.py +++ b/backend/tests/plugins/translate/test_executor.py @@ -274,7 +274,9 @@ class TestFinalizeRun: def test_mixed_failed(self, db_session): run = TranslationRun(id="r1", job_id=JOB_ID, status="RUNNING") result = TranslationExecutor._finalize_run(run, 8, 2, 0) - assert result.status == "COMPLETED" + # Any failed record (e.g. LLM unavailable) signals FAILED even with partial success. + assert result.status == "FAILED" + assert "2 record(s) failed" in result.error_message class TestDelegationWrappers: diff --git a/backend/tests/plugins/translate/test_orchestrator_retry.py b/backend/tests/plugins/translate/test_orchestrator_retry.py index f605d5f86..f0d7e0341 100644 --- a/backend/tests/plugins/translate/test_orchestrator_retry.py +++ b/backend/tests/plugins/translate/test_orchestrator_retry.py @@ -67,8 +67,8 @@ class TestTranslationRunRetryManager: return rec @pytest.mark.asyncio - async def test_retry_failed_batches_success(self): - """Retry failed batches updates run stats.""" + async def test_retry_failed_batches_partial_failure(self): + """Retry with remaining failures signals FAILED (LLM unavailable), even with successes.""" mgr, db, config, event_log = self._make_retry_mgr() mgr.event_log = event_log run = self._make_run() @@ -95,8 +95,8 @@ class TestTranslationRunRetryManager: result = await mgr.retry_failed_batches("run-1") - # failed=5+1=6, successful=90+3=93 (both non-zero -> COMPLETED) - assert result.status == "COMPLETED" + # failed=5+1=6, successful=90+3=93 -> any failure is FAILED (signals LLM unavailability) + assert result.status == "FAILED" assert result.successful_records >= 90 assert result.failed_records >= 5 assert result.completed_at is not None diff --git a/backend/tests/schemas/test_profile.py b/backend/tests/schemas/test_profile.py index c37980070..a8560be8d 100644 --- a/backend/tests/schemas/test_profile.py +++ b/backend/tests/schemas/test_profile.py @@ -83,7 +83,7 @@ class TestProfilePreference: assert p.superset_username is None assert p.superset_username_normalized is None assert p.show_only_my_dashboards is False - assert p.show_only_slug_dashboards is True + assert p.show_only_slug_dashboards is False assert p.start_page == "dashboards" assert p.auto_open_task_drawer is True assert p.dashboards_table_density == "comfortable" diff --git a/backend/tests/services/test_profile_preference_service.py b/backend/tests/services/test_profile_preference_service.py index 2c12a43bd..0978483c2 100644 --- a/backend/tests/services/test_profile_preference_service.py +++ b/backend/tests/services/test_profile_preference_service.py @@ -139,7 +139,7 @@ class TestGetDashboardFilterBinding: assert result["superset_username"] is None assert result["superset_username_normalized"] is None assert result["show_only_my_dashboards"] is False - assert result["show_only_slug_dashboards"] is True + assert result["show_only_slug_dashboards"] is False def test_with_preference(self, service): pref = make_preference_row( diff --git a/backend/tests/services/test_profile_utils.py b/backend/tests/services/test_profile_utils.py index 062e7a845..744de5e39 100644 --- a/backend/tests/services/test_profile_utils.py +++ b/backend/tests/services/test_profile_utils.py @@ -230,7 +230,7 @@ def test_build_default_preference(): assert pref.superset_username is None assert pref.superset_username_normalized is None assert pref.show_only_my_dashboards is False - assert pref.show_only_slug_dashboards is True + assert pref.show_only_slug_dashboards is False assert pref.git_username is None assert pref.git_email is None assert pref.has_git_personal_access_token is False diff --git a/backend/tests/test_app_ws_events.py b/backend/tests/test_app_ws_events.py index 8ffb354a6..5044dabb9 100644 --- a/backend/tests/test_app_ws_events.py +++ b/backend/tests/test_app_ws_events.py @@ -346,3 +346,83 @@ class TestTranslateRunWebSocket: assert ws.send_json.call_count == 2 # #endregion Test.AppModule.TestGenericExceptionOuter # #endregion Test.AppModule.WsEvents + + +# #region Test.AppModule.LogsStatusLoop [C:3] [TYPE Class] [SEMANTICS test,app,ws,logs,task_status,leak] +# @BRIEF Regression: the /ws/logs/{task_id} loop must not leak pending queue.get() coroutines, +# otherwise the terminal task_status (with the structured result) is swallowed. +class _FakeLogEntry: + def __init__(self, message: str): + self.message = message + self.level = "INFO" + self.source = "plugin" + from datetime import datetime + self.timestamp = datetime.now() + + def model_dump(self) -> dict: + return { + "message": self.message, + "level": self.level, + "source": self.source, + "timestamp": self.timestamp, + } + + +class TestLogsStatusLoop: + @pytest.mark.asyncio + async def test_terminal_status_forwarded_after_log(self): + """A log entry followed by a terminal status must both reach the client.""" + from src.app import websocket_endpoint + + ws = MagicMock() + ws.query_params = {"token": "valid"} + ws.accept = AsyncMock() + ws.send_json = AsyncMock() + ws.close = AsyncMock() + + log_queue: asyncio.Queue = asyncio.Queue() + status_queue: asyncio.Queue = asyncio.Queue() + + tm = MagicMock() + tm.subscribe_logs = AsyncMock(return_value=log_queue) + tm.subscribe_status = AsyncMock(return_value=status_queue) + tm.unsubscribe_logs = MagicMock() + tm.unsubscribe_status = MagicMock() + tm.get_task = MagicMock(return_value=None) + tm.get_task_logs = MagicMock(return_value=[]) + + with ( + patch("src.app._authenticate_websocket", return_value=True), + patch("src.app._authorize_websocket", return_value=True), + patch("src.app.get_task_manager", return_value=tm), + ): + runner = asyncio.create_task(websocket_endpoint(ws, "t1")) + + # Let the endpoint reach its main listen loop. + await asyncio.sleep(0.05) + + # 1) A log entry arrives → processed first (without the fix this leaks a status waiter). + await log_queue.put(_FakeLogEntry("some log")) + for _ in range(200): + if ws.send_json.call_count >= 1: + break + await asyncio.sleep(0.01) + + # 2) Terminal status arrives → must be forwarded, not swallowed by a leaked waiter. + await status_queue.put({ + "type": "task_status", + "task_id": "t1", + "task": {"status": "SUCCESS", "result": {"status": "SUCCESS"}}, + }) + await asyncio.wait_for(runner, timeout=5) + + sent = [call.args[0] for call in ws.send_json.call_args_list] + assert any( + isinstance(m, dict) + and m.get("type") == "task_status" + and m.get("task", {}).get("status") == "SUCCESS" + for m in sent + ), f"terminal task_status was not forwarded; sent={sent}" + + +# #endregion Test.AppModule.LogsStatusLoop diff --git a/docs/adr-sed-dataset-mutations.md b/docs/adr-sed-dataset-mutations.md new file mode 100644 index 000000000..ea0375661 --- /dev/null +++ b/docs/adr-sed-dataset-mutations.md @@ -0,0 +1,36 @@ +# ADR — Sed-мутации датасетов (viewer + migration) + +## Цель +Дать пользователю возможность править датасеты literal-заменой (а-ля `sed`): +- в модуле просмотра датасетов: ко всем / выбранным / одному датасету; +- в модуле миграции: как именованное сохраняемое правило, применяемое к датасетам + внутри экспортного ZIP перед импортом. + +## Решение +- **Единая семантика правила**: `{ name, find, replace }`, literal-замена подстроки. + `find` пустой → операция no-op (безопасный fallback). +- **Миграция**: черновик уже добавил `literal_find`/`literal_replace` в + `MigrationEngine.transform_zip` и `DashboardSelection`, но импортировал + несуществующий `src/core/literal_replace.py`. Создаём этот модуль и применяем + замену к raw-тексту каждого `datasets/**/*.yaml` после DB-UUID-трансформа. + Плагин пробрасывает поля из `params` в оба вызова `transform_zip`. +- **Просмотр (live mutation)**: `POST /api/datasets/mutate` + `POST /api/datasets/mutate/preview`. + GET полного датасета → literal-замена → PUT обратно (`override_columns=false`). + `dataset_ids` пустой = все датасеты env. +- **Цели замены (`targets`)**: `sql` (SQL датасета) / `metrics` (expression, verbose_name) / + `yaml` (полная рекурсивная замена всех строковых полей — опасно, эксклюзивно, + с явным предупреждением в UI и 422 при комбинации с другими). +- **Обязательный предпросмотр**: в просмотре apply заблокирован до валидного preview + (fingerprint-гейт); preview возвращает per-dataset `field: before → after` без записи. + В миграции предпросмотр не требуется (по явному решению пользователя). +- **Именованные правила**: общий frontend-store `sedRules` (localStorage, + ключ `dsh.datasetSedRules.v1`) + переиспользуемый `SedRuleEditor.svelte`. + +## Отклонено +- Regex-`sed` (полный синтаксис s///) — отклонили в пользу literal-замены: + детерминированно, предсказуемо, покрывает заявленный кейс + `dm_view.account_debt → dm_view.account_debt_final`. Regex легко добавить позже. +- Слепой рекурсивный replace по всем строковым полям live-датасета — отклонён: + задел бы описания/комментарии. Ограничились идентификаторными полями. +- Хранение правил в БД — отклонено для MVP: localStorage достаточно, правила + приватны пользователю и не требуют миграций схемы. diff --git a/frontend/src/lib/components/SedRuleEditor.svelte b/frontend/src/lib/components/SedRuleEditor.svelte new file mode 100644 index 000000000..facce6b24 --- /dev/null +++ b/frontend/src/lib/components/SedRuleEditor.svelte @@ -0,0 +1,105 @@ + + + + + + + + +
+
+ + +
+
+ + +
+ +
+
+ + +
+ +
+ + {#if sedRules.rules.length > 0} +
+
+ + +
+ +
+ {/if} + +
+ diff --git a/frontend/src/lib/components/migration/MappingCoverageOverview.svelte b/frontend/src/lib/components/migration/MappingCoverageOverview.svelte index 1b08931b4..0001115f3 100644 --- a/frontend/src/lib/components/migration/MappingCoverageOverview.svelte +++ b/frontend/src/lib/components/migration/MappingCoverageOverview.svelte @@ -30,6 +30,11 @@ targetEnvId?: string; onselect?: (_sourceEnvId: string, _targetEnvId: string) => void; } = $props(); + + /** Resolve a human environment name for a per-pair error key; fall back to the raw id. */ + function envName(envId: string): string { + return analysis?.environments?.find((env) => env.id === envId)?.name ?? envId; + }
@@ -91,9 +96,11 @@ {$t.migration?.[`mapping_status_${pair.status}`] || pair.status} {#if Object.keys(pair.errors).length > 0} -

- {Object.values(pair.errors).join("; ")} -

+
+ {#each Object.entries(pair.errors) as [envId, errMsg] (envId)} + {envName(envId)}: {errMsg} + {/each} +
{/if} diff --git a/frontend/src/lib/components/migration/__tests__/MappingCoverageOverview.test.ts b/frontend/src/lib/components/migration/__tests__/MappingCoverageOverview.test.ts index 253672fb1..28aaebb97 100644 --- a/frontend/src/lib/components/migration/__tests__/MappingCoverageOverview.test.ts +++ b/frontend/src/lib/components/migration/__tests__/MappingCoverageOverview.test.ts @@ -53,5 +53,21 @@ describe("MappingCoverageOverview", () => { expect(screen.getByRole("alert").textContent).toContain("Coverage unavailable"); }); + + it("renders per-pair environment error with resolved name", () => { + const withError = { + ...analysis, + pairs: [ + { + ...analysis.pairs[0], + status: "error" as const, + errors: { prod: "Superset unavailable (Status 502)" }, + }, + ], + }; + render(MappingCoverageOverview, { analysis: withError }); + + expect(screen.getByText("Production: Superset unavailable (Status 502)")).toBeTruthy(); + }); }); // #endregion Test.Migration.MappingCoverageOverview diff --git a/frontend/src/lib/components/ui/MappingTable.svelte b/frontend/src/lib/components/ui/MappingTable.svelte index 9feef3593..2755db4a8 100644 --- a/frontend/src/lib/components/ui/MappingTable.svelte +++ b/frontend/src/lib/components/ui/MappingTable.svelte @@ -152,7 +152,13 @@ {/if}
- +
+ + + + + + - - -
@@ -188,10 +194,10 @@ ? 'bg-surface-card' : 'bg-surface-card'} > - - {sDb.database_name} + + {sDb.database_name} + + {#if isSaved} {/if} + {#if isDraft && suggestion} @@ -177,6 +196,7 @@
+
@@ -247,5 +267,122 @@ {/if} + + + {#if model.showSedModal} + + {/if} diff --git a/frontend/src/routes/datasets/DatasetList.svelte b/frontend/src/routes/datasets/DatasetList.svelte index 0130a04f6..cce741c68 100644 --- a/frontend/src/routes/datasets/DatasetList.svelte +++ b/frontend/src/routes/datasets/DatasetList.svelte @@ -174,6 +174,13 @@ + diff --git a/frontend/src/routes/migration/+page.svelte b/frontend/src/routes/migration/+page.svelte index 847fd9947..f88042a16 100644 --- a/frontend/src/routes/migration/+page.svelte +++ b/frontend/src/routes/migration/+page.svelte @@ -47,6 +47,7 @@ import TaskLogViewer from "$lib/components/tasks/TaskLogViewer.svelte"; import PasswordPrompt from "$lib/components/migration/PasswordPrompt.svelte"; import DryRunDiffList from "$lib/components/migration/DryRunDiffList.svelte"; + import SedRuleEditor from "$lib/components/SedRuleEditor.svelte"; import { MigrationModel } from "../../lib/models/MigrationModel.svelte.ts"; import { t } from "$lib/i18n/index.svelte.js"; @@ -338,121 +339,162 @@

{$t.dashboard?.settings || "Options"}

-
- - {#if model.replaceDb} -
-
-

{$t.migration?.database_mappings || "Database Mappings"}

- -
- {#if model.fetchingDbs} -
- - {$t.migration?.loading_dbs || "Loading databases and suggestions..."} -
- {:else if model.sourceDatabases.length > 0} - model.saveMapping(mapping.sourceUuid, mapping.targetUuid)} - onAcceptAll={() => model.saveAllSuggestions()} - /> - {:else if model.sourceEnvId && model.targetEnvId} -

{$t.migration?.mapping_hint || "Select environments and click \"Fetch Databases\" to start mapping."}

- {/if} -
- {/if} -