Add model fixtures

test/unit/data/model/conftest.py

@@ -1,4 +1,7 @@
+import contextlib
 import os
+import random
+import string
 import tempfile
 import uuid
 
@@ -6,6 +9,8 @@ import pytest
 from sqlalchemy import create_engine
 from sqlalchemy.orm import Session
 
+from galaxy import model as m
+
 
 @pytest.fixture
 def database_name():
@@ -43,3 +48,259 @@ def session(init_model, engine):
 def tmp_directory():
     with tempfile.TemporaryDirectory() as tmp_dir:
         yield tmp_dir
+
+
+# model fixture factories
+
+
+@pytest.fixture
+def make_cleanup_event_history_association(session):
+    def f(**kwd):
+        model = m.CleanupEventHistoryAssociation(**kwd)
+        write_to_db(session, model)
+        return model
+
+    return f
+
+
+@pytest.fixture
+def make_data_manager_history_association(session):
+    def f(**kwd):
+        model = m.DataManagerHistoryAssociation(**kwd)
+        write_to_db(session, model)
+        return model
+
+    return f
+
+
+@pytest.fixture
+def make_default_history_permissions(session, make_history, make_role):
+    def f(**kwd):
+        if "history" not in kwd:
+            kwd["history"] = make_history()
+        if "action" not in kwd:
+            kwd["action"] = random_str()
+        if "role" not in kwd:
+            kwd["role"] = make_role()
+        model = m.DefaultHistoryPermissions(**kwd)
+        write_to_db(session, model)
+        return model
+
+    return f
+
+
+@pytest.fixture
+def make_event(session):
+    def f(**kwd):
+        model = m.Event(**kwd)
+        write_to_db(session, model)
+        return model
+
+    return f
+
+
+@pytest.fixture
+def make_galaxy_session(session):
+    def f(**kwd):
+        model = m.GalaxySession(**kwd)
+        write_to_db(session, model)
+        return model
+
+    return f
+
+
+@pytest.fixture
+def make_galaxy_session_to_history_association(session, make_history, make_galaxy_session):
+    def f(**kwd):
+        if "galaxy_session" not in kwd:
+            kwd["galaxy_session"] = make_galaxy_session()
+        if "history" not in kwd:
+            kwd["history"] = make_history()
+        model = m.GalaxySessionToHistoryAssociation(**kwd)
+        write_to_db(session, model)
+        return model
+
+    return f
+
+
+@pytest.fixture
+def make_history(session, make_user):
+    def f(**kwd):
+        if "user" not in kwd:
+            kwd["user"] = make_user()
+        model = m.History(**kwd)
+        write_to_db(session, model)
+        return model
+
+    return f
+
+
+@pytest.fixture
+def make_history_annotation_association(session):
+    def f(**kwd):
+        model = m.HistoryAnnotationAssociation(**kwd)
+        write_to_db(session, model)
+        return model
+
+    return f
+
+
+@pytest.fixture
+def make_history_dataset_association(session):
+    def f(**kwd):
+        model = m.HistoryDatasetAssociation(**kwd)
+        write_to_db(session, model)
+        return model
+
+    return f
+
+
+@pytest.fixture
+def make_history_dataset_collection_association(session):
+    def f(**kwd):
+        model = m.HistoryDatasetCollectionAssociation(**kwd)
+        write_to_db(session, model)
+        return model
+
+    return f
+
+
+@pytest.fixture
+def make_history_rating_association(session, make_user, make_history):
+    def f(**kwd):
+        if "user" not in kwd:
+            kwd["user"] = make_user()
+        if "item" not in kwd:
+            kwd["item"] = make_history()
+        model = m.HistoryRatingAssociation(**kwd)
+        write_to_db(session, model)
+        return model
+
+    return f
+
+
+@pytest.fixture
+def make_history_tag_association(session):
+    def f(**kwd):
+        model = m.HistoryTagAssociation(**kwd)
+        write_to_db(session, model)
+        return model
+
+    return f
+
+
+@pytest.fixture
+def make_history_user_share_association(session):
+    def f(**kwd):
+        model = m.HistoryUserShareAssociation(**kwd)
+        write_to_db(session, model)
+        return model
+
+    return f
+
+
+@pytest.fixture
+def make_job(session):
+    def f(**kwd):
+        model = m.Job(**kwd)
+        write_to_db(session, model)
+        return model
+
+    return f
+
+
+@pytest.fixture
+def make_job_export_history_archive(session):
+    def f(**kwd):
+        model = m.JobExportHistoryArchive(**kwd)
+        write_to_db(session, model)
+        return model
+
+    return f
+
+
+@pytest.fixture
+def make_job_import_history_archive(session):
+    def f(**kwd):
+        model = m.JobImportHistoryArchive(**kwd)
+        write_to_db(session, model)
+        return model
+
+    return f
+
+
+@pytest.fixture
+def make_role(session):
+    def f(**kwd):
+        model = m.Role(**kwd)
+        write_to_db(session, model)
+        return model
+
+    return f
+
+
+@pytest.fixture
+def make_workflow(session):
+    def f(**kwd):
+        model = m.Workflow(**kwd)
+        write_to_db(session, model)
+        return model
+
+    return f
+
+
+@pytest.fixture
+def make_workflow_invocation(session, make_workflow):
+    def f(**kwd):
+        if "workflow" not in kwd:
+            kwd["workflow"] = make_workflow()
+        model = m.WorkflowInvocation(**kwd)
+        write_to_db(session, model)
+        return model
+
+    return f
+
+
+@pytest.fixture
+def make_user(session):
+    def f(**kwd):
+        if "username" not in kwd:
+            kwd["username"] = random_str()
+        if "email" not in kwd:
+            kwd["email"] = random_email()
+        if "password" not in kwd:
+            kwd["password"] = random_str()
+        model = m.User(**kwd)
+        write_to_db(session, model)
+        return model
+
+    return f
+
+
+# utility functions
+
+
+@contextlib.contextmanager
+def transaction(session):
+    if not session.in_transaction():
+        with session.begin():
+            yield
+    else:
+        yield
+
+
+def random_str() -> str:
+    alphabet = string.ascii_lowercase + string.digits
+    size = random.randint(5, 10)
+    return "".join(random.choices(alphabet, k=size))
+
+
+def random_email() -> str:
+    text = random_str()
+    return f"{text}@galaxy.testing"
+
+
+def write_to_db(session, model) -> None:
+    with transaction(session):
+        session.add(model)
+        session.commit()