aminghadersohi commented on code in PR #40130:
URL: https://github.com/apache/superset/pull/40130#discussion_r3342304108
##########
tests/unit_tests/datasets/commands/importers/v1/import_test.py:
##########
@@ -1014,6 +1015,331 @@ def test_import_dataset_access_check(
import_dataset(config)
+def test_import_soft_deleted_dataset_overwrite_restores_in_place(
+ mocker: MockerFixture,
+ session: Session,
+) -> None:
+ """
+ Overwrite-importing a soft-deleted dataset must restore the row in
+ place rather than hard-delete-and-replace. A hard delete would
+ cascade through the chart back-reference and table_columns /
+ sql_metrics rows; in-place restore preserves them.
+ """
+ mocker.patch.object(security_manager, "can_access", return_value=True)
+
+ engine = db.session.get_bind()
+ SqlaTable.metadata.create_all(engine) # pylint: disable=no-member
+
+ database = Database(database_name="my_database",
sqlalchemy_uri="sqlite://")
+ db.session.add(database)
+ db.session.flush()
+
+ config = copy.deepcopy(dataset_fixture)
+ config["database_id"] = database.id
+
+ initial = import_dataset(config)
+ original_id = initial.id
+
+ existing = db.session.query(SqlaTable).filter_by(uuid=config["uuid"]).one()
+ existing.deleted_at = datetime(2026, 1, 1, 12, 0, 0)
+ db.session.flush()
+
+ admin = User(
+ first_name="Alice",
+ last_name="Doe",
+ email="[email protected]",
+ username="admin",
+ roles=[Role(name="Admin")],
+ )
+
+ with override_user(admin):
+ restored = import_dataset(config, overwrite=True)
+
+ assert restored.id == original_id
+ assert restored.deleted_at is None
+
+
+def test_import_soft_deleted_dataset_non_overwrite_restores_for_owner(
+ mocker: MockerFixture,
+ session: Session,
+) -> None:
+ """
+ Non-overwrite re-import of a soft-deleted UUID is implicitly a
+ restore-and-update: the user is bringing the dataset back by
+ uploading it again. The same ownership rule as the overwrite path
+ applies, so an owner (or admin) succeeds without setting
+ overwrite=True.
+ """
+ mocker.patch.object(security_manager, "can_access", return_value=True)
+
+ engine = db.session.get_bind()
+ SqlaTable.metadata.create_all(engine) # pylint: disable=no-member
+
+ database = Database(database_name="my_database",
sqlalchemy_uri="sqlite://")
+ db.session.add(database)
+ db.session.flush()
+
+ config = copy.deepcopy(dataset_fixture)
+ config["database_id"] = database.id
+
+ initial = import_dataset(config)
+ original_id = initial.id
+
+ existing = db.session.query(SqlaTable).filter_by(uuid=config["uuid"]).one()
+ existing.deleted_at = datetime(2026, 1, 1, 12, 0, 0)
+ db.session.flush()
+
+ admin = User(
+ first_name="Alice",
+ last_name="Doe",
+ email="[email protected]",
+ username="admin",
+ roles=[Role(name="Admin")],
+ )
+
+ with override_user(admin):
+ restored = import_dataset(config, overwrite=False)
+
+ assert restored.id == original_id
+ assert restored.deleted_at is None
+
+
+def test_import_soft_deleted_dataset_non_overwrite_raises_for_non_owner(
+ mocker: MockerFixture,
+ session: Session,
+) -> None:
+ """
+ Non-overwrite re-import that would resurrect a soft-deleted dataset
+ must respect ownership: a non-owner without admin role cannot
+ restore-via-import. Mirrors the explicit /restore endpoint's check.
+ """
+ mocker.patch.object(security_manager, "can_access", return_value=True)
+
+ engine = db.session.get_bind()
+ SqlaTable.metadata.create_all(engine) # pylint: disable=no-member
+
+ database = Database(database_name="my_database",
sqlalchemy_uri="sqlite://")
+ db.session.add(database)
+ db.session.flush()
+
+ config = copy.deepcopy(dataset_fixture)
+ config["database_id"] = database.id
+
+ import_dataset(config)
+ existing = db.session.query(SqlaTable).filter_by(uuid=config["uuid"]).one()
+ existing.deleted_at = datetime(2026, 1, 1, 12, 0, 0)
+ db.session.flush()
+
+ non_owner = User(
+ first_name="Bob",
+ last_name="Roe",
+ email="[email protected]",
+ username="bob",
+ roles=[Role(name="Gamma")],
+ )
+
+ with override_user(non_owner):
+ with pytest.raises(ImportFailedError) as excinfo:
+ import_dataset(config, overwrite=False)
+ assert "permissions to restore" in str(excinfo.value)
+
Review Comment:
**MEDIUM — Missing post-exception state assertion**
After `pytest.raises(ImportFailedError)` exits, there is no assertion that
`existing.deleted_at is not None`. A bug that clears `deleted_at` **before**
raising the exception would pass this test undetected (Rule 26 — test that
removes the change still passes).
Add:
```python
# Verify the permission check prevented the mutation
assert existing.deleted_at is not None, (
"deleted_at was cleared before the exception — mutation happened before
raise"
)
```
##########
tests/unit_tests/datasets/commands/importers/v1/import_test.py:
##########
@@ -1014,6 +1015,331 @@ def test_import_dataset_access_check(
import_dataset(config)
+def test_import_soft_deleted_dataset_overwrite_restores_in_place(
+ mocker: MockerFixture,
+ session: Session,
+) -> None:
+ """
+ Overwrite-importing a soft-deleted dataset must restore the row in
+ place rather than hard-delete-and-replace. A hard delete would
+ cascade through the chart back-reference and table_columns /
+ sql_metrics rows; in-place restore preserves them.
+ """
+ mocker.patch.object(security_manager, "can_access", return_value=True)
+
+ engine = db.session.get_bind()
+ SqlaTable.metadata.create_all(engine) # pylint: disable=no-member
+
+ database = Database(database_name="my_database",
sqlalchemy_uri="sqlite://")
+ db.session.add(database)
+ db.session.flush()
+
+ config = copy.deepcopy(dataset_fixture)
+ config["database_id"] = database.id
+
+ initial = import_dataset(config)
+ original_id = initial.id
+
+ existing = db.session.query(SqlaTable).filter_by(uuid=config["uuid"]).one()
+ existing.deleted_at = datetime(2026, 1, 1, 12, 0, 0)
+ db.session.flush()
+
+ admin = User(
+ first_name="Alice",
+ last_name="Doe",
+ email="[email protected]",
+ username="admin",
+ roles=[Role(name="Admin")],
+ )
+
+ with override_user(admin):
+ restored = import_dataset(config, overwrite=True)
+
+ assert restored.id == original_id
+ assert restored.deleted_at is None
+
+
+def test_import_soft_deleted_dataset_non_overwrite_restores_for_owner(
+ mocker: MockerFixture,
+ session: Session,
+) -> None:
+ """
+ Non-overwrite re-import of a soft-deleted UUID is implicitly a
+ restore-and-update: the user is bringing the dataset back by
+ uploading it again. The same ownership rule as the overwrite path
+ applies, so an owner (or admin) succeeds without setting
+ overwrite=True.
+ """
+ mocker.patch.object(security_manager, "can_access", return_value=True)
+
+ engine = db.session.get_bind()
+ SqlaTable.metadata.create_all(engine) # pylint: disable=no-member
+
+ database = Database(database_name="my_database",
sqlalchemy_uri="sqlite://")
+ db.session.add(database)
+ db.session.flush()
+
+ config = copy.deepcopy(dataset_fixture)
+ config["database_id"] = database.id
+
+ initial = import_dataset(config)
+ original_id = initial.id
+
+ existing = db.session.query(SqlaTable).filter_by(uuid=config["uuid"]).one()
+ existing.deleted_at = datetime(2026, 1, 1, 12, 0, 0)
+ db.session.flush()
+
+ admin = User(
+ first_name="Alice",
+ last_name="Doe",
+ email="[email protected]",
+ username="admin",
+ roles=[Role(name="Admin")],
+ )
+
+ with override_user(admin):
+ restored = import_dataset(config, overwrite=False)
+
+ assert restored.id == original_id
+ assert restored.deleted_at is None
+
+
+def test_import_soft_deleted_dataset_non_overwrite_raises_for_non_owner(
+ mocker: MockerFixture,
+ session: Session,
+) -> None:
+ """
+ Non-overwrite re-import that would resurrect a soft-deleted dataset
+ must respect ownership: a non-owner without admin role cannot
+ restore-via-import. Mirrors the explicit /restore endpoint's check.
+ """
+ mocker.patch.object(security_manager, "can_access", return_value=True)
+
+ engine = db.session.get_bind()
+ SqlaTable.metadata.create_all(engine) # pylint: disable=no-member
+
+ database = Database(database_name="my_database",
sqlalchemy_uri="sqlite://")
+ db.session.add(database)
+ db.session.flush()
+
+ config = copy.deepcopy(dataset_fixture)
+ config["database_id"] = database.id
+
+ import_dataset(config)
+ existing = db.session.query(SqlaTable).filter_by(uuid=config["uuid"]).one()
+ existing.deleted_at = datetime(2026, 1, 1, 12, 0, 0)
+ db.session.flush()
+
+ non_owner = User(
+ first_name="Bob",
+ last_name="Roe",
+ email="[email protected]",
+ username="bob",
+ roles=[Role(name="Gamma")],
+ )
+
+ with override_user(non_owner):
+ with pytest.raises(ImportFailedError) as excinfo:
+ import_dataset(config, overwrite=False)
+ assert "permissions to restore" in str(excinfo.value)
+
+
+def test_import_soft_deleted_dataset_raises_when_caller_lacks_can_write(
+ mocker: MockerFixture,
+ session: Session,
+) -> None:
+ """
+ Case B: re-import of a soft-deleted UUID by a caller without
+ can_write must raise, not silently return the soft-deleted row.
+
+ Real-world scenario: a user has can_write Dashboard but not
+ can_write Dataset, and they import a dashboard zip that references
+ a soft-deleted dataset. Silently returning the row would let the
+ dashboard importer wire the dashboard's charts to a deleted dataset
+ and produce broken chart loads.
+ """
+ mocker.patch.object(security_manager, "can_access", return_value=False)
+
+ engine = db.session.get_bind()
+ SqlaTable.metadata.create_all(engine) # pylint: disable=no-member
+
+ database = Database(database_name="my_database",
sqlalchemy_uri="sqlite://")
+ db.session.add(database)
+ db.session.flush()
+
+ config = copy.deepcopy(dataset_fixture)
+ config["database_id"] = database.id
+
+ # Seed a soft-deleted dataset with the matching UUID directly, so the
+ # test doesn't need to flip permissions mid-test.
+ existing = SqlaTable(
+ table_name="soft_deleted_dataset",
+ database_id=database.id,
+ uuid=config["uuid"],
+ deleted_at=datetime(2026, 1, 1, 12, 0, 0),
+ )
+ db.session.add(existing)
+ db.session.flush()
+
+ with pytest.raises(ImportFailedError) as excinfo:
+ import_dataset(config, overwrite=False)
+ assert "can_write" in str(excinfo.value)
+
Review Comment:
**MEDIUM — Missing post-exception state assertion (Case B path)**
Same gap as the non-owner test: only the error message is asserted, not that
`existing.deleted_at` is still set after the failed call. A regression that
clears `deleted_at` before raising would pass this test.
Add:
```python
assert existing.deleted_at is not None, (
"Case B: deleted_at was cleared before raising — mutation happened
before raise"
)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
