diff --git a/lib/galaxy/model/unittest_utils/gxy_model_fixtures.py b/lib/galaxy/model/unittest_utils/gxy_model_fixtures.py
index d3ece2b7ecdb..764f907f3ffa 100644
--- a/lib/galaxy/model/unittest_utils/gxy_model_fixtures.py
+++ b/lib/galaxy/model/unittest_utils/gxy_model_fixtures.py
@@ -604,7 +604,7 @@ def tool_tag_association(session):
 
 @pytest.fixture
 def user(session):
-    instance = model.User(email=get_unique_value(), password="password")
+    instance = model.User(email=get_unique_value(), password="password", username=get_unique_value())
     yield from dbcleanup_wrapper(session, instance)
 
 
diff --git a/test/unit/app/jobs/test_job_context.py b/test/unit/app/jobs/test_job_context.py
index 29b030982595..5f454c9aea1c 100644
--- a/test/unit/app/jobs/test_job_context.py
+++ b/test/unit/app/jobs/test_job_context.py
@@ -46,7 +46,7 @@ def test_job_context_discover_outputs_flushes_once(mocker):
     sa_session = app.model.context
     # mocker is a pytest-mock fixture
 
-    u = model.User(email="collection@example.com", password="password")
+    u = model.User(email="collection@example.com", password="password", username="collection")
     h = model.History(name="Test History", user=u)
 
     tool = Tool(app)
diff --git a/test/unit/app/jobs/test_rule_helper.py b/test/unit/app/jobs/test_rule_helper.py
index 1d1197a0dc78..2dfefcd93f00 100644
--- a/test/unit/app/jobs/test_rule_helper.py
+++ b/test/unit/app/jobs/test_rule_helper.py
@@ -64,9 +64,9 @@ def __setup_fixtures(app):
     # user1 has 3 jobs queued and 2 jobs running on cluster1 and one queued and
     # on running job on local. user2 has a queued and running job on the cluster.
     # user3 has no jobs.
-    user1 = model.User(email=USER_EMAIL_1, password="pass1")
-    user2 = model.User(email=USER_EMAIL_2, password="pass2")
-    user3 = model.User(email=USER_EMAIL_2, password="pass2")
+    user1 = model.User(email=USER_EMAIL_1, password="pass1", username="u1")
+    user2 = model.User(email=USER_EMAIL_2, password="pass2", username="u2")
+    user3 = model.User(email=USER_EMAIL_2, password="pass2", username="u3")
 
     app.add(user1, user2, user3)
 
diff --git a/test/unit/app/managers/test_JobConnectionsManager.py b/test/unit/app/managers/test_JobConnectionsManager.py
index 5ec06f388fd7..2c12ea52aced 100644
--- a/test/unit/app/managers/test_JobConnectionsManager.py
+++ b/test/unit/app/managers/test_JobConnectionsManager.py
@@ -3,6 +3,8 @@
 
 from galaxy.managers.job_connections import JobConnectionsManager
 from galaxy.model import (
+    DatasetCollection,
+    History,
     HistoryDatasetAssociation,
     HistoryDatasetCollectionAssociation,
     Job,
@@ -25,11 +27,19 @@ def job_connections_manager(sa_session) -> JobConnectionsManager:
 
 # =============================================================================
 def setup_connected_dataset(sa_session: galaxy_scoped_session):
+    history = History()
+    sa_session.add(history)
     center_hda = HistoryDatasetAssociation(sa_session=sa_session, create_dataset=True)
     input_hda = HistoryDatasetAssociation(sa_session=sa_session, create_dataset=True)
+    input_dc = DatasetCollection(collection_type="list")
     input_hdca = HistoryDatasetCollectionAssociation()
+    input_hdca.collection = input_dc
     output_hda = HistoryDatasetAssociation(sa_session=sa_session, create_dataset=True)
+    output_dc = DatasetCollection(collection_type="list")
     output_hdca = HistoryDatasetCollectionAssociation()
+    output_hdca.collection = output_dc
+    history.stage_addition([center_hda, input_hda, input_hdca, output_hda, output_hdca])
+    history.add_pending_items()
     input_job = Job()
     output_job = Job()
     input_job.add_output_dataset("output_hda", center_hda)
@@ -55,12 +65,22 @@ def setup_connected_dataset(sa_session: galaxy_scoped_session):
 
 
 def setup_connected_dataset_collection(sa_session: galaxy_scoped_session):
+    history = History()
+    sa_session.add(history)
+    center_dc = DatasetCollection(collection_type="list")
     center_hdca = HistoryDatasetCollectionAssociation()
+    center_hdca.collection = center_dc
     input_hda1 = HistoryDatasetAssociation(sa_session=sa_session, create_dataset=True)
     input_hda2 = HistoryDatasetAssociation(sa_session=sa_session, create_dataset=True)
+    input_dc = DatasetCollection(collection_type="list")
     input_hdca = HistoryDatasetCollectionAssociation()
+    input_hdca.collection = input_dc
     output_hda = HistoryDatasetAssociation(sa_session=sa_session, create_dataset=True)
+    output_dc = DatasetCollection(collection_type="list")
     output_hdca = HistoryDatasetCollectionAssociation()
+    output_hdca.collection = output_dc
+    history.stage_addition([center_hdca, input_hda1, input_hda2, input_hdca, output_hda, output_hdca])
+    history.add_pending_items()
     input_job = Job()
     output_job = Job()
     input_job.add_output_dataset_collection("output_hdca", center_hdca)
diff --git a/test/unit/app/tools/test_error_reporting.py b/test/unit/app/tools/test_error_reporting.py
index 7db9cb4bf60e..5651c1c73bf9 100644
--- a/test/unit/app/tools/test_error_reporting.py
+++ b/test/unit/app/tools/test_error_reporting.py
@@ -57,7 +57,7 @@ def test_hda_security(self, tmp_path):
         permissions = {access_action: [private_role], manage_action: [private_role]}
         security_agent.set_all_dataset_permissions(hda.dataset, permissions)
 
-        other_user = model.User(email="otheruser@galaxyproject.org", password="mockpass2")
+        other_user = model.User(email="otheruser@galaxyproject.org", password="mockpass2", username="otheruser")
         self._commit_objects([other_user])
         security_agent = self.app.security_agent
         email_path = self.email_path
@@ -125,7 +125,7 @@ def test_no_redact_user_details_in_bugreport(self, tmp_path):
         )
 
     def _setup_model_objects(self):
-        user = model.User(email=TEST_USER_EMAIL, password="mockpass")
+        user = model.User(email=TEST_USER_EMAIL, password="mockpass", username=TEST_USER_EMAIL.split("@")[0])
         job = model.Job()
         job.tool_id = "cat1"
         job.history = model.History()
diff --git a/test/unit/app/tools/test_history_imp_exp.py b/test/unit/app/tools/test_history_imp_exp.py
index 1f54029642da..097940bc1b84 100644
--- a/test/unit/app/tools/test_history_imp_exp.py
+++ b/test/unit/app/tools/test_history_imp_exp.py
@@ -36,7 +36,7 @@ def t_data_path(name):
 def _run_jihaw_cleanup(archive_dir, app=None):
     app = app or _mock_app()
     job = model.Job()
-    job.user = model.User(email="test@test.org", password="test")
+    job.user = model.User(email="test@test.org", password="test", username="test")
     job.tool_stderr = ""
     jiha = model.JobImportHistoryArchive(job=job, archive_dir=archive_dir)
     app.model.context.current.add_all([job, jiha])
@@ -661,7 +661,7 @@ def _setup_history_for_export(history_name):
     sa_session = app.model.context
 
     email = history_name.replace(" ", "-") + "-user@example.org"
-    u = model.User(email=email, password="password")
+    u = model.User(email=email, password="password", username=email.split("@")[0])
     h = model.History(name=history_name, user=u)
 
     return app, sa_session, h
diff --git a/test/unit/app/tools/test_toolbox.py b/test/unit/app/tools/test_toolbox.py
index f4faaa370a1a..54bf1aad6c9d 100644
--- a/test/unit/app/tools/test_toolbox.py
+++ b/test/unit/app/tools/test_toolbox.py
@@ -356,9 +356,7 @@ def __test_workflow(self):
         workflow = model.Workflow()
         workflow.stored_workflow = stored_workflow
         stored_workflow.latest_workflow = workflow
-        user = model.User()
-        user.email = "test@example.com"
-        user.password = "passw0rD1"
+        user = model.User(email="test@example.com", password="passw0rD1", username="test")
         stored_workflow.user = user
         self.app.model.context.add(workflow)
         self.app.model.context.add(stored_workflow)
diff --git a/test/unit/data/model/test_model_discovery.py b/test/unit/data/model/test_model_discovery.py
index 7237d37f0ef0..b4bcfdf01289 100644
--- a/test/unit/data/model/test_model_discovery.py
+++ b/test/unit/data/model/test_model_discovery.py
@@ -225,7 +225,7 @@ def _import_library_target(target, work_directory):
     with store.DirectoryModelExportStore(temp_directory, app=app, serialize_dataset_objects=True) as export_store:
         persist_target_to_export_store(target, export_store, app.object_store, work_directory)
 
-    u = model.User(email="library@example.com", password="password")
+    u = model.User(email="library@example.com", password="password", username="library")
 
     import_options = store.ImportOptions(allow_dataset_object_edit=True, allow_library_creation=True)
     import_model_store = store.get_import_model_store_for_directory(
@@ -240,7 +240,7 @@ def _import_library_target(target, work_directory):
 def _import_directory_to_history(app, target, work_directory):
     sa_session = app.model.context
 
-    u = model.User(email="collection@example.com", password="password")
+    u = model.User(email="collection@example.com", password="password", username="collection")
     import_history = model.History(name="Test History for Import", user=u)
 
     sa_session = app.model.context
diff --git a/test/unit/data/model/test_model_store.py b/test/unit/data/model/test_model_store.py
index a30410dd0207..7ac5b05a724a 100644
--- a/test/unit/data/model/test_model_store.py
+++ b/test/unit/data/model/test_model_store.py
@@ -312,7 +312,7 @@ def test_import_library_require_permissions():
     app = _mock_app()
     sa_session = app.model.context
 
-    u = model.User(email="collection@example.com", password="password")
+    u = model.User(email="collection@example.com", password="password", username="collection")
 
     library = model.Library(name="my library 1", description="my library description", synopsis="my synopsis")
     root_folder = model.LibraryFolder(name="my library 1", description="folder description")
@@ -340,7 +340,7 @@ def test_import_export_library():
     app = _mock_app()
     sa_session = app.model.context
 
-    u = model.User(email="collection@example.com", password="password")
+    u = model.User(email="collection@example.com", password="password", username="collection")
 
     library = model.Library(name="my library 1", description="my library description", synopsis="my synopsis")
     root_folder = model.LibraryFolder(name="my library 1", description="folder description")
@@ -684,7 +684,7 @@ def test_import_export_edit_collection():
     app = _mock_app()
     sa_session = app.model.context
 
-    u = model.User(email="collection@example.com", password="password")
+    u = model.User(email="collection@example.com", password="password", username="collection")
     h = model.History(name="Test History", user=u)
 
     c1 = model.DatasetCollection(collection_type="list", populated=False)
@@ -761,7 +761,7 @@ def test_import_export_composite_datasets():
     app = _mock_app()
     sa_session = app.model.context
 
-    u = model.User(email="collection@example.com", password="password")
+    u = model.User(email="collection@example.com", password="password", username="collection")
     h = model.History(name="Test History", user=u)
 
     d1 = _create_datasets(sa_session, h, 1, extension="html")[0]
@@ -804,7 +804,7 @@ def test_edit_metadata_files():
     app = _mock_app(store_by="uuid")
     sa_session = app.model.context
 
-    u = model.User(email="collection@example.com", password="password")
+    u = model.User(email="collection@example.com", password="password", username="collection")
     h = model.History(name="Test History", user=u)
 
     d1 = _create_datasets(sa_session, h, 1, extension="bam")[0]
@@ -910,7 +910,7 @@ def _assert_simple_cat_job_imported(imported_history, state="ok"):
 def _setup_simple_cat_job(app, state="ok"):
     sa_session = app.model.context
 
-    u = model.User(email="collection@example.com", password="password")
+    u = model.User(email="collection@example.com", password="password", username="collection")
     h = model.History(name="Test History", user=u)
 
     d1, d2 = _create_datasets(sa_session, h, 2)
@@ -966,7 +966,7 @@ def _setup_invocation(app):
 def _setup_simple_collection_job(app, state="ok"):
     sa_session = app.model.context
 
-    u = model.User(email="collection@example.com", password="password")
+    u = model.User(email="collection@example.com", password="password", username="collection")
     h = model.History(name="Test History", user=u)
 
     d1, d2, d3, d4 = _create_datasets(sa_session, h, 4)
@@ -1170,7 +1170,7 @@ def setup_fixture_context_with_user(
 ) -> StoreFixtureContextWithUser:
     app = _mock_app(store_by=store_by)
     sa_session = app.model.context
-    user = model.User(email=user_email, password="password")
+    user = model.User(email=user_email, password="password", username=user_email.split("@")[0])
     return StoreFixtureContextWithUser(app=app, sa_session=sa_session, user=user)
 
 
diff --git a/test/unit/data/test_galaxy_mapping.py b/test/unit/data/test_galaxy_mapping.py
index b21318c46f01..38a38a556c95 100644
--- a/test/unit/data/test_galaxy_mapping.py
+++ b/test/unit/data/test_galaxy_mapping.py
@@ -40,6 +40,11 @@
 PRIVATE_OBJECT_STORE_ID = "my_private_data"
 
 
+@pytest.fixture()
+def u():
+    return model.User(f"{uuid.uuid4()}@#user.com", str(uuid.uuid4()), str(uuid.uuid4()))
+
+
 class BaseModelTestCase(TestCase):
     model: mapping.GalaxyModelMapping
 
@@ -78,7 +83,7 @@ def expunge(cls):
 class TestMappings(BaseModelTestCase):
     def test_ratings(self):
         user_email = "rater@example.com"
-        u = model.User(email=user_email, password="password")
+        u = model.User(email=user_email, password="password", username=user_email.split("@")[0])
         self.persist(u)
 
         def persist_and_check_rating(rating_class, item):
@@ -118,6 +123,8 @@ def persist_and_check_rating(rating_class, item):
 
         dataset_collection = model.DatasetCollection(collection_type="paired")
         history_dataset_collection = model.HistoryDatasetCollectionAssociation(collection=dataset_collection)
+        self.session().add_all([h, history_dataset_collection])
+        h.add_dataset_collection(history_dataset_collection)
         self.persist(history_dataset_collection)
         persist_and_check_rating(model.HistoryDatasetCollectionRatingAssociation, history_dataset_collection)
 
@@ -153,7 +160,7 @@ def assert_display_name_converts_to_unicode(item, name):
 
     def test_hda_to_library_dataset_dataset_association(self):
         model = self.model
-        u = self.model.User(email="mary@example.com", password="password")
+        u = self.model.User(email="mary@example.com", password="password", username="mary")
         h1 = model.History(name="History 1", user=u)
         hda = model.HistoryDatasetAssociation(
             name="hda_name", create_dataset=True, history=h1, sa_session=model.session
@@ -185,7 +192,7 @@ def test_hda_to_library_dataset_dataset_association(self):
 
     def test_hda_to_library_dataset_dataset_association_fails_if_private(self):
         model = self.model
-        u = model.User(email="mary2@example.com", password="password")
+        u = model.User(email="mary2@example.com", password="password", username="mary2")
         h1 = model.History(name="History 1", user=u)
         hda = model.HistoryDatasetAssociation(
             name="hda_name", create_dataset=True, history=h1, sa_session=model.session
@@ -204,7 +211,7 @@ def test_hda_to_library_dataset_dataset_association_fails_if_private(self):
     def test_tags(self):
         TAG_NAME = "Test Tag"
         my_tag = model.Tag(name=TAG_NAME)
-        u = model.User(email="tagger@example.com", password="password")
+        u = model.User(email="tagger@example.com", password="password", username="tagger")
         self.persist(my_tag, u)
 
         def tag_and_test(taggable_object, tag_association_class):
@@ -238,13 +245,13 @@ def tag_and_test(taggable_object, tag_association_class):
 
         dataset_collection = model.DatasetCollection(collection_type="paired")
         history_dataset_collection = model.HistoryDatasetCollectionAssociation(collection=dataset_collection)
+        h.add_dataset_collection(history_dataset_collection)
         tag_and_test(history_dataset_collection, model.HistoryDatasetCollectionTagAssociation)
 
         library_dataset_collection = model.LibraryDatasetCollectionAssociation(collection=dataset_collection)
         tag_and_test(library_dataset_collection, model.LibraryDatasetCollectionTagAssociation)
 
-    def test_collection_get_interface(self):
-        u = model.User(email="mary@example.com", password="password")
+    def test_collection_get_interface(self, u):
         h1 = model.History(name="History 1", user=u)
         d1 = model.HistoryDatasetAssociation(
             extension="txt", history=h1, create_dataset=True, sa_session=self.model.session
@@ -260,8 +267,7 @@ def test_collection_get_interface(self):
         for i in range(elements):
             assert c1[i] == dces[i]
 
-    def test_dataset_instance_order(self) -> None:
-        u = model.User(email="mary@example.com", password="password")
+    def test_dataset_instance_order(self, u) -> None:
         h1 = model.History(name="History 1", user=u)
         elements = []
         list_pair = model.DatasetCollection(collection_type="list:paired")
@@ -307,8 +313,7 @@ def test_dataset_instance_order(self) -> None:
         assert all(d.name == f"forward_{i}" for i, d in enumerate(forward_hdas))
         assert all(d.name == f"reverse_{i}" for i, d in enumerate(reverse_hdas))
 
-    def test_collections_in_histories(self):
-        u = model.User(email="mary@example.com", password="password")
+    def test_collections_in_histories(self, u):
         h1 = model.History(name="History 1", user=u)
         d1 = model.HistoryDatasetAssociation(
             extension="txt", history=h1, create_dataset=True, sa_session=self.model.session
@@ -322,6 +327,8 @@ def test_collections_in_histories(self):
 
         dce1 = model.DatasetCollectionElement(collection=c1, element=d1, element_identifier="left")
         dce2 = model.DatasetCollectionElement(collection=c1, element=d2, element_identifier="right")
+        h1.stage_addition([d1, d2, hc1])
+        h1.add_pending_items()
 
         self.persist(u, h1, d1, d2, c1, hc1, dce1, dce2)
 
@@ -336,8 +343,7 @@ def test_collections_in_histories(self):
         assert loaded_dataset_collection["left"] == dce1
         assert loaded_dataset_collection["right"] == dce2
 
-    def test_collections_in_library_folders(self):
-        u = model.User(email="mary2@example.com", password="password")
+    def test_collections_in_library_folders(self, u):
         lf = model.LibraryFolder(name="RootFolder")
         library = model.Library(name="Library1", root_folder=lf)
         ld1 = model.LibraryDataset()
@@ -357,7 +363,7 @@ def test_collections_in_library_folders(self):
         # assert loaded_dataset_collection.collection_type == "pair"
 
     def test_dataset_action_tuples(self):
-        u = model.User(email="foo", password="foo")
+        u = model.User(email="foo", password="foo", username="foo")
         h1 = model.History(user=u)
         hda1 = model.HistoryDatasetAssociation(history=h1, create_dataset=True, sa_session=self.model.session)
         hda2 = model.HistoryDatasetAssociation(history=h1, create_dataset=True, sa_session=self.model.session)
@@ -372,8 +378,7 @@ def test_dataset_action_tuples(self):
         self.model.session.flush()
         assert c1.dataset_action_tuples == [("action1", r1.id), ("action3", r1.id)]
 
-    def test_nested_collection_attributes(self):
-        u = model.User(email="mary2@example.com", password="password")
+    def test_nested_collection_attributes(self, u):
         h1 = model.History(name="History 1", user=u)
         d1 = model.HistoryDatasetAssociation(
             extension="bam", history=h1, create_dataset=True, sa_session=self.model.session
@@ -460,8 +465,7 @@ def test_nested_collection_attributes(self):
         ]
         assert c4.dataset_elements == [dce1, dce2]
 
-    def test_dataset_dbkeys_and_extensions_summary(self):
-        u = model.User(email="mary2@example.com", password="password")
+    def test_dataset_dbkeys_and_extensions_summary(self, u):
         h1 = model.History(name="History 1", user=u)
         d1 = model.HistoryDatasetAssociation(
             extension="bam", dbkey="hg19", history=h1, create_dataset=True, sa_session=self.model.session
@@ -474,12 +478,13 @@ def test_dataset_dbkeys_and_extensions_summary(self):
         dce2 = model.DatasetCollectionElement(collection=c1, element=d2, element_identifier="reverse", element_index=1)
         hdca = model.HistoryDatasetCollectionAssociation(collection=c1, history=h1)
         self.model.session.add_all([d1, d2, c1, dce1, dce2, hdca])
+        h1.stage_addition([d1, d2, hdca])
+        h1.add_pending_items()
         self.model.session.flush()
         assert hdca.dataset_dbkeys_and_extensions_summary[0] == {"hg19"}
         assert hdca.dataset_dbkeys_and_extensions_summary[1] == {"bam", "txt"}
 
-    def test_populated_optimized_ok(self):
-        u = model.User(email="mary2@example.com", password="password")
+    def test_populated_optimized_ok(self, u):
         h1 = model.History(name="History 1", user=u)
         d1 = model.HistoryDatasetAssociation(
             extension="txt", history=h1, create_dataset=True, sa_session=self.model.session
@@ -491,6 +496,8 @@ def test_populated_optimized_ok(self):
         dce1 = model.DatasetCollectionElement(collection=c1, element=d1, element_identifier="forward", element_index=0)
         dce2 = model.DatasetCollectionElement(collection=c1, element=d2, element_identifier="reverse", element_index=1)
         self.model.session.add_all([d1, d2, c1, dce1, dce2])
+        h1.stage_addition([d1, d2])
+        h1.add_pending_items()
         self.model.session.flush()
         assert c1.populated
         assert c1.populated_optimized
@@ -523,7 +530,7 @@ def test_populated_optimized_list_list_not_populated(self):
         assert not c2.populated_optimized
 
     def test_default_disk_usage(self):
-        u = model.User(email="disk_default@test.com", password="password")
+        u = model.User(email="disk_default@test.com", password="password", username="disk_default")
         self.persist(u)
         u.adjust_total_disk_usage(1, None)
         u_id = u.id
@@ -535,7 +542,7 @@ def test_basic(self):
         original_user_count = len(self.model.session.scalars(select(model.User)).all())
 
         # Make some changes and commit them
-        u = model.User(email="james@foo.bar.baz", password="password")
+        u = model.User(email="james@foo.bar.baz", password="password", username="james")
         h1 = model.History(name="History 1", user=u)
         h2 = model.History(name=("H" * 1024))
         self.persist(u, h1, h2)
@@ -591,8 +598,7 @@ def test_dataset_job_relationship(self):
         ).scalar_one()
         assert loaded_dataset.job_id == job.id
 
-    def test_jobs(self):
-        u = model.User(email="jobtest@foo.bar.baz", password="password")
+    def test_jobs(self, u):
         job = model.Job()
         job.user = u
         job.tool_id = "cat1"
@@ -602,8 +608,7 @@ def test_jobs(self):
         loaded_job = self.model.session.scalars(select(model.Job).filter(model.Job.user == u).limit(1)).first()
         assert loaded_job.tool_id == "cat1"
 
-    def test_job_metrics(self):
-        u = model.User(email="jobtest@foo.bar.baz", password="password")
+    def test_job_metrics(self, u):
         job = model.Job()
         job.user = u
         job.tool_id = "cat1"
@@ -623,8 +628,7 @@ def test_job_metrics(self):
         # Ensure big values truncated
         assert len(task.text_metrics[1].metric_value) <= 1023
 
-    def test_tasks(self):
-        u = model.User(email="jobtest@foo.bar.baz", password="password")
+    def test_tasks(self, u):
         job = model.Job()
         task = model.Task(job=job, working_directory="/tmp", prepare_files_cmd="split.sh")
         job.user = u
@@ -633,8 +637,7 @@ def test_tasks(self):
         loaded_task = self.model.session.scalars(select(model.Task).filter(model.Task.job == job).limit(1)).first()
         assert loaded_task.prepare_input_files_cmd == "split.sh"
 
-    def test_history_contents(self):
-        u = model.User(email="contents@foo.bar.baz", password="password")
+    def test_history_contents(self, u):
         # gs = model.GalaxySession()
         h1 = model.History(name="HistoryContentsHistory1", user=u)
 
@@ -667,8 +670,7 @@ def contents_iter_names(**kwds):
 
         assert contents_iter_names(ids=[d1.id, d3.id]) == ["1", "3"]
 
-    def test_history_audit(self):
-        u = model.User(email="contents@foo.bar.baz", password="password")
+    def test_history_audit(self, u):
         h1 = model.History(name="HistoryAuditHistory", user=u)
         h2 = model.History(name="HistoryAuditHistory", user=u)
 
@@ -722,8 +724,8 @@ def _non_empty_flush(self):
         session.add(lf)
         session.flush()
 
-    def test_current_session(self):
-        user = model.User(email="testworkflows@bx.psu.edu", password="password")
+    def test_current_session(self, u):
+        user = u
         galaxy_session = model.GalaxySession()
         galaxy_session.user = user
         self.persist(user, galaxy_session)
@@ -733,12 +735,12 @@ def test_current_session(self):
         self.persist(user, new_galaxy_session)
         assert user.current_galaxy_session == new_galaxy_session
 
-    def test_flush_refreshes(self):
+    def test_flush_refreshes(self, u):
         # Normally I don't believe in unit testing library code, but the behaviors around attribute
         # states and flushing in SQL Alchemy is very subtle and it is good to have a executable
         # reference for how it behaves in the context of Galaxy objects.
         model = self.model
-        user = model.User(email="testworkflows@bx.psu.edu", password="password")
+        user = u
         galaxy_session = model.GalaxySession()
         galaxy_session_other = model.GalaxySession()
         galaxy_session.user = user
@@ -810,10 +812,8 @@ def test_flush_refreshes(self):
         session.flush(model.GalaxySession())
         assert "id" not in inspect(galaxy_model_object_new).unloaded
 
-    def test_workflows(self):
-        user = model.User(email="testworkflows@bx.psu.edu", password="password")
-
-        child_workflow = _workflow_from_steps(user, [])
+    def test_workflows(self, u):
+        child_workflow = _workflow_from_steps(u, [])
         self.persist(child_workflow)
 
         workflow_step_1 = model.WorkflowStep()
@@ -830,19 +830,19 @@ def test_workflows(self):
         workflow_step_2.get_or_add_input("moo")
         workflow_step_1.add_connection("foo", "cow", workflow_step_2)
 
-        workflow = _workflow_from_steps(user, [workflow_step_1, workflow_step_2])
+        workflow = _workflow_from_steps(u, [workflow_step_1, workflow_step_2])
         self.persist(workflow)
         workflow_id = workflow.id
 
         annotation = model.WorkflowStepAnnotationAssociation()
         annotation.annotation = "Test Step Annotation"
-        annotation.user = user
+        annotation.user = u
         add_object_to_object_session(annotation, workflow_step_1)
         annotation.workflow_step = workflow_step_1
         self.persist(annotation)
 
         assert workflow_step_1.id is not None
-        workflow_invocation = _invocation_for_workflow(user, workflow)
+        workflow_invocation = _invocation_for_workflow(u, workflow)
 
         invocation_uuid = uuid.uuid1()
 
@@ -930,7 +930,7 @@ def check_private_role(private_role, email):
             assert private_role.description == "Private Role for " + email
 
         email = "rule_user_1@example.com"
-        u = model.User(email=email, password="password")
+        u = model.User(email=email, password="password", username=email.split("@")[0])
         self.persist(u)
 
         role = security_agent.get_private_user_role(u)
@@ -940,7 +940,7 @@ def check_private_role(private_role, email):
         check_private_role(role, email)
 
         email = "rule_user_2@example.com"
-        u = model.User(email=email, password="password")
+        u = model.User(email=email, password="password", username=email.split("@")[0])
         self.persist(u)
         role = security_agent.get_private_user_role(u)
         assert role is None
@@ -1034,8 +1034,7 @@ def test_can_manage_private_dataset(self):
         assert security_agent.can_manage_dataset(u_from.all_roles(), d1.dataset)
         assert not security_agent.can_manage_dataset(u_other.all_roles(), d1.dataset)
 
-    def test_history_hid_counter_is_expired_after_next_hid_call(self):
-        u = model.User(email="hid_abuser@example.com", password="password")
+    def test_history_hid_counter_is_expired_after_next_hid_call(self, u):
         h = model.History(name="History for hid testing", user=u)
         self.persist(u, h)
         state = inspect(h)
@@ -1049,8 +1048,7 @@ def test_history_hid_counter_is_expired_after_next_hid_call(self):
         assert "id" not in state.unloaded  # but other attributes have NOT been expired
         assert h.hid_counter == 2  # check this last: this causes thie hid_counter to be reloaded
 
-    def test_next_hid(self):
-        u = model.User(email="hid_abuser@example.com", password="password")
+    def test_next_hid(self, u):
         h = model.History(name="History for hid testing", user=u)
         self.persist(u, h)
         assert h.hid_counter == 1
@@ -1135,9 +1133,9 @@ def _three_users(self, suffix):
         email_to = f"user_{suffix}e2@example.com"
         email_other = f"user_{suffix}e3@example.com"
 
-        u_from = model.User(email=email_from, password="password")
-        u_to = model.User(email=email_to, password="password")
-        u_other = model.User(email=email_other, password="password")
+        u_from = model.User(email=email_from, password="password", username=email_from.split("@")[0])
+        u_to = model.User(email=email_to, password="password", username=email_to.split("@")[0])
+        u_other = model.User(email=email_other, password="password", username=email_other.split("@")[0])
         self.persist(u_from, u_to, u_other)
         return u_from, u_to, u_other
 
diff --git a/test/unit/data/test_model_copy.py b/test/unit/data/test_model_copy.py
index 29416b552b83..0b755364a580 100644
--- a/test/unit/data/test_model_copy.py
+++ b/test/unit/data/test_model_copy.py
@@ -87,12 +87,10 @@ def test_history_collection_copy(list_size=NUM_DATASETS):
                 list_elements.append(paired_collection_element)
                 model.context.add_all([forward_dce, reverse_dce, paired_collection_element])
             history_dataset_collection = model.HistoryDatasetCollectionAssociation(collection=list_collection)
-            history_dataset_collection.user = old_history.user
+            history_dataset_collection.user = old_history.username
             model.context.add(history_dataset_collection)
 
             session = model.context
-            with transaction(session):
-                session.commit()
 
             old_history.add_dataset_collection(history_dataset_collection)
             history_dataset_collection.add_item_annotation(
@@ -147,7 +145,7 @@ def _setup_mapping_and_user():
             thread_local_log=THREAD_LOCAL_LOG,
         )
 
-        u = User(email="historycopy@example.com", password="password")
+        u = User(email="historycopy@example.com", password="password", username="historycopy")
         h1 = History(name="HistoryCopyHistory1", user=u)
         model.context.add_all([u, h1])
         session = model.context
diff --git a/test/unit/data/test_quota.py b/test/unit/data/test_quota.py
index 5c34ea7cb056..143968ac809d 100644
--- a/test/unit/data/test_quota.py
+++ b/test/unit/data/test_quota.py
@@ -16,7 +16,7 @@ class TestPurgeUsage(BaseModelTestCase):
     def setUp(self):
         super().setUp()
         model = self.model
-        u = model.User(email="purge_usage@example.com", password="password")
+        u = model.User(email="purge_usage@example.com", password="password", username=str(uuid.uuid4()))
         u.disk_usage = 25
         self.persist(u)
 
@@ -67,7 +67,8 @@ def test_calculate_usage_per_source(self):
 class TestCalculateUsage(BaseModelTestCase):
     def setUp(self):
         model = self.model
-        u = model.User(email=f"calc_usage{uuid.uuid1()}@example.com", password="password")
+        email = f"calc_usage{uuid.uuid1()}@example.com"
+        u = model.User(email, password="password", username=email.split("@")[0])
         self.persist(u)
         h = model.History(name="History for Calculated Usage", user=u)
         self.persist(h)
@@ -359,7 +360,7 @@ def setUp(self):
         self.quota_agent = DatabaseQuotaAgent(model)
 
     def test_quota(self):
-        u = model.User(email="quota@example.com", password="password")
+        u = model.User(email="quota@example.com", password="password", username="quota")
         self.persist(u)
 
         self._assert_user_quota_is(u, None)
@@ -405,7 +406,7 @@ def test_quota(self):
 
     def test_labeled_quota(self):
         model = self.model
-        u = model.User(email="labeled_quota@example.com", password="password")
+        u = model.User(email="labeled_quota@example.com", password="password", username="labeled_quota")
         self.persist(u)
 
         label1 = "coollabel1"
@@ -456,7 +457,7 @@ def _assert_user_quota_is(self, user, amount, quota_source_label=None):
 class TestUsage(BaseModelTestCase):
     def test_usage(self):
         model = self.model
-        u = model.User(email="usage@example.com", password="password")
+        u = model.User(email="usage@example.com", password="password", username="usage")
         self.persist(u)
 
         u.adjust_total_disk_usage(123, None)
@@ -466,7 +467,7 @@ def test_usage(self):
 
     def test_labeled_usage(self):
         model = self.model
-        u = model.User(email="labeled.usage@example.com", password="password")
+        u = model.User(email="labeled.usage@example.com", password="password", username="labeled")
         self.persist(u)
         assert len(u.quota_source_usages) == 0
 
diff --git a/test/unit/test_galaxy_transactions.py b/test/unit/test_galaxy_transactions.py
index 8093df91c0ce..581c32f2c2d5 100644
--- a/test/unit/test_galaxy_transactions.py
+++ b/test/unit/test_galaxy_transactions.py
@@ -62,7 +62,7 @@ def test_logging_actions_on(transaction):
 
 
 def test_expunge_all(transaction):
-    user = model.User("foo", "bar1")
+    user = model.User("foo", "bar1", username="foo")
     transaction.sa_session.add(user)
 
     user.password = "bar2"
diff --git a/test/unit/workflows/test_run_parameters.py b/test/unit/workflows/test_run_parameters.py
index 76eae8955744..a45c34b0f44d 100644
--- a/test/unit/workflows/test_run_parameters.py
+++ b/test/unit/workflows/test_run_parameters.py
@@ -1,3 +1,5 @@
+import uuid
+
 from galaxy import model
 from galaxy.model.base import transaction
 from galaxy.workflow.run_request import (
@@ -89,7 +91,7 @@ def __new_input():
 
 
 def __workflow_fixure(trans):
-    user = model.User(email="testworkflow_params@bx.psu.edu", password="pass")
+    user = model.User(email="testworkflow_params@bx.psu.edu", password="pass", username=str(uuid.uuid4()))
     stored_workflow = model.StoredWorkflow()
     stored_workflow.user = user
     workflow = model.Workflow()
diff --git a/test/unit/workflows/workflow_support.py b/test/unit/workflows/workflow_support.py
index 05064e722fac..0e89a20ec91c 100644
--- a/test/unit/workflows/workflow_support.py
+++ b/test/unit/workflows/workflow_support.py
@@ -28,7 +28,7 @@ def save_workflow(self, workflow):
     @property
     def user(self):
         if self._user is None:
-            self._user = model.User(email="testworkflows@bx.psu.edu", password="password")
+            self._user = model.User(email="testworkflows@bx.psu.edu", password="password", username="testworkflows")
         return self._user