feat: Virtual dataset duplication (#20309)

* Inital duplicate functionality

* Fix formatting

* Create dedicated duplicate API

* Make use of new API

* Make use of new api permissions

* Add integration tests for duplicating datasets

* Add licenses

* Fix linting errors

* Change confirm button to 'Duplicate'

* Fix HTTP status code and response

* Add missing import

* Use user id instead of user object

* Remove stray debug print

* Fix sqlite tests

* Specify type of extra

* Add frontend tests

* Add match statement to test

tests/integration_tests/datasets/api_tests.py

@@ -99,6 +99,13 @@ class TestDatasetApi(SupersetTestCase):
             .all()
         )
 
+    def get_fixture_virtual_datasets(self) -> List[SqlaTable]:
+        return (
+            db.session.query(SqlaTable)
+            .filter(SqlaTable.table_name.in_(self.fixture_virtual_table_names))
+            .all()
+        )
+
     @pytest.fixture()
     def create_virtual_datasets(self):
         with self.create_app().app_context():
@@ -443,7 +450,12 @@ class TestDatasetApi(SupersetTestCase):
         rv = self.get_assert_metric(uri, "info")
         data = json.loads(rv.data.decode("utf-8"))
         assert rv.status_code == 200
-        assert set(data["permissions"]) == {"can_read", "can_write", "can_export"}
+        assert set(data["permissions"]) == {
+            "can_read",
+            "can_write",
+            "can_export",
+            "can_duplicate",
+        }
 
     def test_create_dataset_item(self):
         """
@@ -2134,3 +2146,78 @@ class TestDatasetApi(SupersetTestCase):
 
         db.session.delete(table_w_certification)
         db.session.commit()
+
+    @pytest.mark.usefixtures("create_virtual_datasets")
+    def test_duplicate_virtual_dataset(self):
+        """
+        Dataset API: Test duplicate virtual dataset
+        """
+        if backend() == "sqlite":
+            return
+
+        dataset = self.get_fixture_virtual_datasets()[0]
+
+        self.login(username="admin")
+        uri = f"api/v1/dataset/duplicate"
+        table_data = {"base_model_id": dataset.id, "table_name": "Dupe1"}
+        rv = self.post_assert_metric(uri, table_data, "duplicate")
+        assert rv.status_code == 201
+        rv_data = json.loads(rv.data)
+        new_dataset: SqlaTable = (
+            db.session.query(SqlaTable).filter_by(id=rv_data["id"]).one_or_none()
+        )
+        assert new_dataset is not None
+        assert new_dataset.id != dataset.id
+        assert new_dataset.table_name == "Dupe1"
+        assert len(new_dataset.columns) == 2
+        assert new_dataset.columns[0].column_name == "id"
+        assert new_dataset.columns[1].column_name == "name"
+
+    @pytest.mark.usefixtures("create_datasets")
+    def test_duplicate_physical_dataset(self):
+        """
+        Dataset API: Test duplicate physical dataset
+        """
+        if backend() == "sqlite":
+            return
+
+        dataset = self.get_fixture_datasets()[0]
+
+        self.login(username="admin")
+        uri = f"api/v1/dataset/duplicate"
+        table_data = {"base_model_id": dataset.id, "table_name": "Dupe2"}
+        rv = self.post_assert_metric(uri, table_data, "duplicate")
+        assert rv.status_code == 422
+
+    @pytest.mark.usefixtures("create_virtual_datasets")
+    def test_duplicate_existing_dataset(self):
+        """
+        Dataset API: Test duplicate dataset with existing name
+        """
+        if backend() == "sqlite":
+            return
+
+        dataset = self.get_fixture_virtual_datasets()[0]
+
+        self.login(username="admin")
+        uri = f"api/v1/dataset/duplicate"
+        table_data = {
+            "base_model_id": dataset.id,
+            "table_name": "sql_virtual_dataset_2",
+        }
+        rv = self.post_assert_metric(uri, table_data, "duplicate")
+        assert rv.status_code == 422
+
+    def test_duplicate_invalid_dataset(self):
+        """
+        Dataset API: Test duplicate invalid dataset
+        """
+
+        self.login(username="admin")
+        uri = f"api/v1/dataset/duplicate"
+        table_data = {
+            "base_model_id": -1,
+            "table_name": "Dupe3",
+        }
+        rv = self.post_assert_metric(uri, table_data, "duplicate")
+        assert rv.status_code == 422