Remove unused functions in dataset managers

lib/galaxy/managers/datasets.py

@@ -64,23 +64,6 @@ def __init__(self, app: MinimalManagerApp):
         self.quota_agent = app.quota_agent
         self.security_agent = app.model.security_agent
 
-    def create(self, manage_roles=None, access_roles=None, flush=True, **kwargs):
-        """
-        Create and return a new Dataset object.
-        """
-        # default to NEW state on new datasets
-        kwargs.update(dict(state=(kwargs.get("state", model.Dataset.states.NEW))))
-        dataset = model.Dataset(**kwargs)
-        self.session().add(dataset)
-
-        self.permissions.set(dataset, manage_roles, access_roles, flush=False)
-
-        if flush:
-            session = self.session()
-            with transaction(session):
-                session.commit()
-        return dataset
-
     def copy(self, item, **kwargs):
         raise exceptions.NotImplemented("Datasets cannot be copied")
 

lib/galaxy/managers/histories.py

@@ -237,12 +237,6 @@ def p_tag_filter(term_text: str, quoted: bool):
             stmt = stmt.offset(payload.offset)
         return trans.sa_session.scalars(stmt), total_matches  # type:ignore[return-value]
 
-    def copy(self, history, user, **kwargs):
-        """
-        Copy and return the given `history`.
-        """
-        return history.copy(target_user=user, **kwargs)
-
     # .... sharable
     # overriding to handle anonymous users' current histories in both cases
     def by_user(

test/unit/app/managers/test_DatasetManager.py

@@ -31,6 +31,11 @@ def set_up_managers(self):
         super().set_up_managers()
         self.dataset_manager = DatasetManager(self.app)
 
+    def create_dataset_with_permissions(self, manage_roles=None, access_roles=None) -> model.Dataset:
+        dataset = self.dataset_manager.create(flush=False)
+        self.dataset_manager.permissions.set(dataset, manage_roles, access_roles)
+        return dataset
+
     def test_create(self):
         self.log("should be able to create a new Dataset")
         dataset1 = self.dataset_manager.create()
@@ -125,7 +130,7 @@ def test_create_public_dataset(self):
         )
         owner = self.user_manager.create(**user2_data)
         owner_private_role = self.user_manager.private_role(owner)
-        dataset = self.dataset_manager.create(manage_roles=[owner_private_role])
+        dataset = self.create_dataset_with_permissions(manage_roles=[owner_private_role])
 
         permissions = self.dataset_manager.permissions.get(dataset)
         assert isinstance(permissions, tuple)
@@ -157,7 +162,9 @@ def test_create_private_dataset(self):
         self.log("should be able to create a new Dataset and give it private permissions")
         owner = self.user_manager.create(**user2_data)
         owner_private_role = self.user_manager.private_role(owner)
-        dataset = self.dataset_manager.create(manage_roles=[owner_private_role], access_roles=[owner_private_role])
+        dataset = self.create_dataset_with_permissions(
+            manage_roles=[owner_private_role], access_roles=[owner_private_role]
+        )
 
         permissions = self.dataset_manager.permissions.get(dataset)
         assert isinstance(permissions, tuple)
@@ -268,7 +275,7 @@ def test_serialize_permissions(self):
 
     def test_serializers(self):
         # self.user_manager.create( **user2_data )
-        dataset = self.dataset_manager.create()
+        dataset = self.dataset_manager.create(state=model.Dataset.states.NEW)
         all_keys = list(self.dataset_serializer.serializable_keyset)
         serialized = self.dataset_serializer.serialize(dataset, all_keys)
 

test/unit/app/managers/test_HistoryManager.py

@@ -66,7 +66,7 @@ def test_base(self):
             == self.trans.sa_session.execute(select(model.History).filter(model.History.user == user2)).scalar_one()
         )
 
-        history2 = self.history_manager.copy(history1, user=user3)
+        history2 = history1.copy(target_user=user3)
 
         self.log("should be able to query")
         histories = self.trans.sa_session.scalars(select(model.History)).all()
@@ -108,7 +108,7 @@ def test_copy(self):
         self.app.tag_handler.set_tags_from_list(user=user2, item=hda, new_tags_list=hda_tags)
         self.hda_manager.annotate(hda, hda_annotation, user=user2)
 
-        history2 = self.history_manager.copy(history1, user=user3)
+        history2 = history1.copy(target_user=user3)
         assert isinstance(history2, model.History)
         assert history2.user == user3
         assert history2 == self.trans.sa_session.get(model.History, history2.id)