fix(API): Updating assets via the API should preserve ownership configuration

superset/commands/base.py

@@ -19,7 +19,7 @@
 
 from flask_appbuilder.security.sqla.models import User
 
-from superset.commands.utils import populate_owners
+from superset.commands.utils import populate_owner_list, update_owner_list
 
 
 class BaseCommand(ABC):
@@ -55,7 +55,7 @@
         :raises OwnersNotFoundValidationError: if at least one owner can't be resolved
         :returns: Final list of owners
         """
-        return populate_owners(owner_ids, default_to_user=True)
+        return populate_owner_list(owner_ids, default_to_user=True)
 
 
 class UpdateMixin:  # pylint: disable=too-few-public-methods
@@ -69,4 +69,18 @@
         :raises OwnersNotFoundValidationError: if at least one owner can't be resolved
         :returns: Final list of owners
         """
-        return populate_owners(owner_ids, default_to_user=False)
+        return populate_owner_list(owner_ids, default_to_user=False)
+
+    @staticmethod
+    def update_owners(
+        current_owners: Optional[list[User]],
+        new_owners: Optional[list[int]],
+    ) -> list[User]:
+        """
+        Handle list of owners for update events.
+
+        :param current_owners: list of current owners
+        :param new_owners: list of new owners specified in the update payload
+        :returns: Final list of owners
+        """
+        return update_owner_list(current_owners, new_owners)

superset/commands/chart/update.py

@@ -90,7 +90,10 @@ def validate(self) -> None:
         if not is_query_context_update(self._properties):
             try:
                 security_manager.raise_for_ownership(self._model)
-                owners = self.populate_owners(owner_ids)
+                owners = self.update_owners(
+                    self._model.owners,
+                    owner_ids,
+                )
                 self._properties["owners"] = owners
             except SupersetSecurityException as ex:
                 raise ChartForbiddenError() from ex

superset/commands/dashboard/update.py

@@ -66,7 +66,7 @@ def run(self) -> Model:
 
     def validate(self) -> None:
         exceptions: list[ValidationError] = []
-        owners_ids: Optional[list[int]] = self._properties.get("owners")
+        owner_ids: Optional[list[int]] = self._properties.get("owners")
         roles_ids: Optional[list[int]] = self._properties.get("roles")
         slug: Optional[str] = self._properties.get("slug")
 
@@ -85,10 +85,11 @@ def validate(self) -> None:
             exceptions.append(DashboardSlugExistsValidationError())
 
         # Validate/Populate owner
-        if owners_ids is None:
-            owners_ids = [owner.id for owner in self._model.owners]
         try:
-            owners = self.populate_owners(owners_ids)
+            owners = self.update_owners(
+                self._model.owners,
+                owner_ids,
+            )
             self._properties["owners"] = owners
         except ValidationError as ex:
             exceptions.append(ex)

superset/commands/dataset/update.py

@@ -100,7 +100,10 @@ def validate(self) -> None:
             exceptions.append(DatabaseChangeValidationError())
         # Validate/Populate owner
         try:
-            owners = self.populate_owners(owner_ids)
+            owners = self.update_owners(
+                self._model.owners,
+                owner_ids,
+            )
             self._properties["owners"] = owners
         except ValidationError as ex:
             exceptions.append(ex)

superset/commands/report/update.py

@@ -118,10 +118,11 @@ def validate(self) -> None:
             raise ReportScheduleForbiddenError() from ex
 
         # Validate/Populate owner
-        if owner_ids is None:
-            owner_ids = [owner.id for owner in self._model.owners]
         try:
-            owners = self.populate_owners(owner_ids)
+            owners = self.update_owners(
+                self._model.owners,
+                owner_ids,
+            )
             self._properties["owners"] = owners
         except ValidationError as ex:
             exceptions.append(ex)

superset/commands/utils.py

@@ -35,7 +35,7 @@
     from superset.connectors.sqla.models import BaseDatasource
 
 
-def populate_owners(
+def populate_owner_list(
     owner_ids: list[int] | None,
     default_to_user: bool,
 ) -> list[User]:
@@ -62,6 +62,25 @@ def populate_owners(
     return owners
 
 
+def update_owner_list(
+    current_owners: list[User] | None,
+    new_owners: list[int] | None,
+) -> list[User]:
+    """
+    Helper function for update commands, to properly handle the owners list.
+    Preserve the previous configuration unless included in the update payload.
+
+    :param current_owners: list of current owners
+    :param new_owners: list of new owners specified in the update payload
+    :returns: Final list of owners
+    """
+    current_owners = current_owners or []
+    owners_ids = (
+        [owner.id for owner in current_owners] if new_owners is None else new_owners
+    )
+    return populate_owner_list(owners_ids, default_to_user=False)
+
+
 def populate_roles(role_ids: list[int] | None = None) -> list[Role]:
     """
     Helper function for commands, will fetch all roles from roles id's

superset/views/datasource/views.py

@@ -31,7 +31,7 @@
     DatasetForbiddenError,
     DatasetNotFoundError,
 )
-from superset.commands.utils import populate_owners
+from superset.commands.utils import populate_owner_list
 from superset.connectors.sqla.models import SqlaTable
 from superset.connectors.sqla.utils import get_physical_table_metadata
 from superset.daos.datasource import DatasourceDAO
@@ -94,7 +94,7 @@ def save(self) -> FlaskResponse:
             except SupersetSecurityException as ex:
                 raise DatasetForbiddenError() from ex
 
-        datasource_dict["owners"] = populate_owners(
+        datasource_dict["owners"] = populate_owner_list(
             datasource_dict["owners"], default_to_user=False
         )
 

tests/integration_tests/charts/api_tests.py

@@ -735,10 +735,59 @@ def test_update_chart_new_owner_admin(self):
         db.session.delete(model)
         db.session.commit()
 
+    @pytest.mark.usefixtures("add_dashboard_to_chart")
+    def test_update_chart_preserve_ownership(self):
+        """
+        Chart API: Test update chart preserves owner list (if un-changed)
+        """
+        chart_data = {
+            "slice_name": "title1_changed",
+        }
+        admin = self.get_user("admin")
+        self.login(username="admin")
+        uri = f"api/v1/chart/{self.chart.id}"
+        rv = self.put_assert_metric(uri, chart_data, "put")
+        self.assertEqual(rv.status_code, 200)
+        self.assertEqual([admin], self.chart.owners)
+
+    @pytest.mark.usefixtures("add_dashboard_to_chart")
+    def test_update_chart_clear_owner_list(self):
+        """
+        Chart API: Test update chart admin can clear owner list
+        """
+        chart_data = {"slice_name": "title1_changed", "owners": []}
+        admin = self.get_user("admin")
+        self.login(username="admin")
+        uri = f"api/v1/chart/{self.chart.id}"
+        rv = self.put_assert_metric(uri, chart_data, "put")
+        self.assertEqual(rv.status_code, 200)
+        self.assertEqual([], self.chart.owners)
+
+    def test_update_chart_populate_owner(self):
+        """
+        Chart API: Test update admin can update chart with
+        no owners to a different owner
+        """
+        gamma = self.get_user("gamma")
+        admin = self.get_user("admin")
+        chart_id = self.insert_chart("title", [], 1).id
+        model = db.session.query(Slice).get(chart_id)
+        self.assertEqual(model.owners, [])
+        chart_data = {"owners": [gamma.id]}
+        self.login(username="admin")
+        uri = f"api/v1/chart/{chart_id}"
+        rv = self.put_assert_metric(uri, chart_data, "put")
+        self.assertEqual(rv.status_code, 200)
+        model_updated = db.session.query(Slice).get(chart_id)
+        self.assertNotIn(admin, model_updated.owners)
+        self.assertIn(gamma, model_updated.owners)
+        db.session.delete(model_updated)
+        db.session.commit()
+
     @pytest.mark.usefixtures("add_dashboard_to_chart")
     def test_update_chart_new_dashboards(self):
         """
-        Chart API: Test update set new owner to current user
+        Chart API: Test update chart associating it with new dashboard
         """
         chart_data = {
             "slice_name": "title1_changed",
@@ -754,7 +803,7 @@ def test_update_chart_new_dashboards(self):
     @pytest.mark.usefixtures("add_dashboard_to_chart")
     def test_not_update_chart_none_dashboards(self):
         """
-        Chart API: Test update set new owner to current user
+        Chart API: Test update chart without changing dashboards configuration
         """
         chart_data = {"slice_name": "title1_changed_again"}
         self.login(username="admin")

tests/integration_tests/dashboards/api_tests.py

@@ -1550,6 +1550,43 @@ def test_update_dashboard_new_owner_admin(self):
         db.session.delete(model)
         db.session.commit()
 
+    def test_update_dashboard_clear_owner_list(self):
+        """
+        Dashboard API: Test update admin can clear up owners list
+        """
+        admin = self.get_user("admin")
+        dashboard_id = self.insert_dashboard("title1", "slug1", [admin.id]).id
+        self.login(username="admin")
+        uri = f"api/v1/dashboard/{dashboard_id}"
+        dashboard_data = {"owners": []}
+        rv = self.client.put(uri, json=dashboard_data)
+        self.assertEqual(rv.status_code, 200)
+        model = db.session.query(Dashboard).get(dashboard_id)
+        self.assertEqual([], model.owners)
+        db.session.delete(model)
+        db.session.commit()
+
+    def test_update_dashboard_populate_owner(self):
+        """
+        Dashboard API: Test update admin can update dashboard with
+        no owners to a different owner
+        """
+        self.login(username="admin")
+        gamma = self.get_user("gamma")
+        dashboard = self.insert_dashboard(
+            "title1",
+            "slug1",
+            [],
+        )
+        uri = f"api/v1/dashboard/{dashboard.id}"
+        dashboard_data = {"owners": [gamma.id]}
+        rv = self.client.put(uri, json=dashboard_data)
+        self.assertEqual(rv.status_code, 200)
+        model = db.session.query(Dashboard).get(dashboard.id)
+        self.assertEqual([gamma], model.owners)
+        db.session.delete(model)
+        db.session.commit()
+
     def test_update_dashboard_slug_formatting(self):
         """
         Dashboard API: Test update slug formatting

tests/integration_tests/datasets/api_tests.py

@@ -833,19 +833,74 @@ def test_create_dataset_sqlalchemy_error(self, mock_dao_create):
         assert rv.status_code == 422
         assert data == {"message": "Dataset could not be created."}
 
+    def test_update_dataset_preserve_ownership(self):
+        """
+        Dataset API: Test update dataset preserves owner list (if un-changed)
+        """
+
+        dataset = self.insert_default_dataset()
+        current_owners = dataset.owners
+        self.login(username="admin")
+        dataset_data = {"description": "new description"}
+        uri = f"api/v1/dataset/{dataset.id}"
+        rv = self.put_assert_metric(uri, dataset_data, "put")
+        assert rv.status_code == 200
+        model = db.session.query(SqlaTable).get(dataset.id)
+        assert model.owners == current_owners
+
+        db.session.delete(dataset)
+        db.session.commit()
+
+    def test_update_dataset_clear_owner_list(self):
+        """
+        Dataset API: Test update dataset admin can clear ownership config
+        """
+
+        dataset = self.insert_default_dataset()
+        self.login(username="admin")
+        dataset_data = {"owners": []}
+        uri = f"api/v1/dataset/{dataset.id}"
+        rv = self.put_assert_metric(uri, dataset_data, "put")
+        assert rv.status_code == 200
+        model = db.session.query(SqlaTable).get(dataset.id)
+        assert model.owners == []
+
+        db.session.delete(dataset)
+        db.session.commit()
+
+    def test_update_dataset_populate_owner(self):
+        """
+        Dataset API: Test update admin can update dataset with
+        no owners to a different owner
+        """
+        self.login(username="admin")
+        gamma = self.get_user("gamma")
+        dataset = self.insert_dataset("ab_permission", [], get_main_database())
+        dataset_data = {"owners": [gamma.id]}
+        uri = f"api/v1/dataset/{dataset.id}"
+        rv = self.put_assert_metric(uri, dataset_data, "put")
+        assert rv.status_code == 200
+        model = db.session.query(SqlaTable).get(dataset.id)
+        assert model.owners == [gamma]
+
+        db.session.delete(dataset)
+        db.session.commit()
+
     def test_update_dataset_item(self):
         """
         Dataset API: Test update dataset item
         """
 
         dataset = self.insert_default_dataset()
+        current_owners = dataset.owners
         self.login(username="admin")
         dataset_data = {"description": "changed_description"}
         uri = f"api/v1/dataset/{dataset.id}"
         rv = self.put_assert_metric(uri, dataset_data, "put")
         assert rv.status_code == 200
         model = db.session.query(SqlaTable).get(dataset.id)
         assert model.description == dataset_data["description"]
+        assert model.owners == current_owners
 
         db.session.delete(dataset)
         db.session.commit()