Add db_creds_test fixture

Author: ethho

tests/conftest.py

@@ -56,6 +56,15 @@ def enable_filepath_feature(monkeypatch):
     monkeypatch.delenv(FILEPATH_FEATURE_SWITCH, raising=True)
 
 
+@pytest.fixture(scope="session")
+def db_creds_test() -> Dict:
+    return dict(
+        host=os.getenv("DJ_TEST_HOST", "fakeservices.datajoint.io"),
+        user=os.getenv("DJ_TEST_USER", "datajoint"),
+        password=os.getenv("DJ_TEST_PASSWORD", "datajoint"),
+    )
+
+
 @pytest.fixture(scope="session")
 def db_creds_root() -> Dict:
     return dict(
@@ -142,12 +151,9 @@ def connection_root(connection_root_bare):
 
 
 @pytest.fixture(scope="session")
-def connection_test(connection_root):
+def connection_test(connection_root, db_creds_test):
     """Test user database connection."""
     database = f"{PREFIX}%%"
-    credentials = dict(
-        host=os.getenv("DJ_HOST"), user="datajoint", password="datajoint"
-    )
     permission = "ALL PRIVILEGES"
 
     # Create MySQL users
@@ -157,14 +163,14 @@ def connection_test(connection_root):
         # create user if necessary on mysql8
         connection_root.query(
             f"""
-            CREATE USER IF NOT EXISTS '{credentials["user"]}'@'%%'
-            IDENTIFIED BY '{credentials["password"]}';
+            CREATE USER IF NOT EXISTS '{db_creds_test["user"]}'@'%%'
+            IDENTIFIED BY '{db_creds_test["password"]}';
             """
         )
         connection_root.query(
             f"""
             GRANT {permission} ON `{database}`.*
-            TO '{credentials["user"]}'@'%%';
+            TO '{db_creds_test["user"]}'@'%%';
             """
         )
     else:
@@ -173,14 +179,14 @@ def connection_test(connection_root):
         connection_root.query(
             f"""
             GRANT {permission} ON `{database}`.*
-            TO '{credentials["user"]}'@'%%'
-            IDENTIFIED BY '{credentials["password"]}';
+            TO '{db_creds_test["user"]}'@'%%'
+            IDENTIFIED BY '{db_creds_test["password"]}';
             """
         )
 
-    connection = dj.Connection(**credentials)
+    connection = dj.Connection(**db_creds_test)
     yield connection
-    connection_root.query(f"""DROP USER `{credentials["user"]}`""")
+    connection_root.query(f"""DROP USER `{db_creds_test["user"]}`""")
     connection.close()