PLAT-138 Convert pytest-compatible nosetest modules

tests/__init__.py

@@ -10,60 +10,3 @@
     user=os.getenv("DJ_USER"),
     password=os.getenv("DJ_PASS"),
 )
-
-
-@pytest.fixture
-def connection_root():
-    """Root user database connection."""
-    dj.config["safemode"] = False
-    connection = dj.Connection(
-        host=os.getenv("DJ_HOST"),
-        user=os.getenv("DJ_USER"),
-        password=os.getenv("DJ_PASS"),
-    )
-    yield connection
-    dj.config["safemode"] = True
-    connection.close()
-
-
-@pytest.fixture
-def connection_test(connection_root):
-    """Test user database connection."""
-    database = f"{PREFIX}%%"
-    credentials = dict(
-        host=os.getenv("DJ_HOST"), user="datajoint", password="datajoint"
-    )
-    permission = "ALL PRIVILEGES"
-
-    # Create MySQL users
-    if version.parse(
-        connection_root.query("select @@version;").fetchone()[0]
-    ) >= version.parse("8.0.0"):
-        # create user if necessary on mysql8
-        connection_root.query(
-            f"""
-            CREATE USER IF NOT EXISTS '{credentials["user"]}'@'%%'
-            IDENTIFIED BY '{credentials["password"]}';
-            """
-        )
-        connection_root.query(
-            f"""
-            GRANT {permission} ON `{database}`.*
-            TO '{credentials["user"]}'@'%%';
-            """
-        )
-    else:
-        # grant permissions. For MySQL 5.7 this also automatically creates user
-        # if not exists
-        connection_root.query(
-            f"""
-            GRANT {permission} ON `{database}`.*
-            TO '{credentials["user"]}'@'%%'
-            IDENTIFIED BY '{credentials["password"]}';
-            """
-        )
-
-    connection = dj.Connection(**credentials)
-    yield connection
-    connection_root.query(f"""DROP USER `{credentials["user"]}`""")
-    connection.close()

tests/conftest.py

@@ -0,0 +1,154 @@
+import datajoint as dj
+from packaging import version
+import os
+import pytest
+from . import PREFIX, schema, schema_simple, schema_advanced
+
+namespace = locals()
+
+
+@pytest.fixture(scope="session")
+def connection_root():
+    """Root user database connection."""
+    dj.config["safemode"] = False
+    connection = dj.Connection(
+        host=os.getenv("DJ_HOST"),
+        user=os.getenv("DJ_USER"),
+        password=os.getenv("DJ_PASS"),
+    )
+    yield connection
+    dj.config["safemode"] = True
+    connection.close()
+
+
+@pytest.fixture(scope="session")
+def connection_test(connection_root):
+    """Test user database connection."""
+    database = f"{PREFIX}%%"
+    credentials = dict(
+        host=os.getenv("DJ_HOST"), user="datajoint", password="datajoint"
+    )
+    permission = "ALL PRIVILEGES"
+
+    # Create MySQL users
+    if version.parse(
+        connection_root.query("select @@version;").fetchone()[0]
+    ) >= version.parse("8.0.0"):
+        # create user if necessary on mysql8
+        connection_root.query(
+            f"""
+            CREATE USER IF NOT EXISTS '{credentials["user"]}'@'%%'
+            IDENTIFIED BY '{credentials["password"]}';
+            """
+        )
+        connection_root.query(
+            f"""
+            GRANT {permission} ON `{database}`.*
+            TO '{credentials["user"]}'@'%%';
+            """
+        )
+    else:
+        # grant permissions. For MySQL 5.7 this also automatically creates user
+        # if not exists
+        connection_root.query(
+            f"""
+            GRANT {permission} ON `{database}`.*
+            TO '{credentials["user"]}'@'%%'
+            IDENTIFIED BY '{credentials["password"]}';
+            """
+        )
+
+    connection = dj.Connection(**credentials)
+    yield connection
+    connection_root.query(f"""DROP USER `{credentials["user"]}`""")
+    connection.close()
+
+
+@pytest.fixture(scope="module")
+def schema_any(connection_test):
+    schema_any = dj.Schema(
+        PREFIX + "_test1", schema.__dict__, connection=connection_test
+    )
+    schema_any(schema.TTest)
+    schema_any(schema.TTest2)
+    schema_any(schema.TTest3)
+    schema_any(schema.NullableNumbers)
+    schema_any(schema.TTestExtra)
+    schema_any(schema.TTestNoExtra)
+    schema_any(schema.Auto)
+    schema_any(schema.User)
+    schema_any(schema.Subject)
+    schema_any(schema.Language)
+    schema_any(schema.Experiment)
+    schema_any(schema.Trial)
+    schema_any(schema.Ephys)
+    schema_any(schema.Image)
+    schema_any(schema.UberTrash)
+    schema_any(schema.UnterTrash)
+    schema_any(schema.SimpleSource)
+    schema_any(schema.SigIntTable)
+    schema_any(schema.SigTermTable)
+    schema_any(schema.DjExceptionName)
+    schema_any(schema.ErrorClass)
+    schema_any(schema.DecimalPrimaryKey)
+    schema_any(schema.IndexRich)
+    schema_any(schema.ThingA)
+    schema_any(schema.ThingB)
+    schema_any(schema.ThingC)
+    schema_any(schema.Parent)
+    schema_any(schema.Child)
+    schema_any(schema.ComplexParent)
+    schema_any(schema.ComplexChild)
+    schema_any(schema.SubjectA)
+    schema_any(schema.SessionA)
+    schema_any(schema.SessionStatusA)
+    schema_any(schema.SessionDateA)
+    schema_any(schema.Stimulus)
+    schema_any(schema.Longblob)
+    yield schema_any
+    schema_any.drop()
+
+
+@pytest.fixture(scope="module")
+def schema_simp(connection_test):
+    schema = dj.Schema(
+        PREFIX + "_relational", schema_simple.__dict__, connection=connection_test
+    )
+    schema(schema_simple.IJ)
+    schema(schema_simple.JI)
+    schema(schema_simple.A)
+    schema(schema_simple.B)
+    schema(schema_simple.L)
+    schema(schema_simple.D)
+    schema(schema_simple.E)
+    schema(schema_simple.F)
+    schema(schema_simple.F)
+    schema(schema_simple.DataA)
+    schema(schema_simple.DataB)
+    schema(schema_simple.Website)
+    schema(schema_simple.Profile)
+    schema(schema_simple.Website)
+    schema(schema_simple.TTestUpdate)
+    schema(schema_simple.ArgmaxTest)
+    schema(schema_simple.ReservedWord)
+    schema(schema_simple.OutfitLaunch)
+    yield schema
+    schema.drop()
+
+
+@pytest.fixture(scope="module")
+def schema_adv(connection_test):
+    schema = dj.Schema(
+        PREFIX + "_advanced", schema_advanced.__dict__, connection=connection_test
+    )
+    schema(schema_advanced.Person)
+    schema(schema_advanced.Parent)
+    schema(schema_advanced.Subject)
+    schema(schema_advanced.Prep)
+    schema(schema_advanced.Slice)
+    schema(schema_advanced.Cell)
+    schema(schema_advanced.InputCell)
+    schema(schema_advanced.LocalSynapse)
+    schema(schema_advanced.GlobalSynapse)
+    yield schema
+    schema.drop()