[WIP] Add converted pytests from backup

tests/__init__.py

@@ -1,14 +1,20 @@
 import datajoint as dj
 from packaging import version
-import pytest
-import os
+import pytest, os, urllib3, certifi, minio
 
 PREFIX = "djtest"
 
 CONN_INFO_ROOT = dict(
-    host=os.getenv("DJ_HOST"),
-    user=os.getenv("DJ_USER"),
-    password=os.getenv("DJ_PASS"),
+    host=(os.getenv("DJ_HOST")),
+    user=(os.getenv("DJ_USER")),
+    password=(os.getenv("DJ_PASS")),
+)
+
+S3_CONN_INFO = dict(
+    endpoint=(os.getenv("DJ_HOST")),
+    access_key="datajoint",
+    secret_key="datajoint",
+    bucket="datajoint.test",
 )
 
 
@@ -17,9 +23,9 @@ 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"),
+        host=(os.getenv("DJ_HOST")),
+        user=(os.getenv("DJ_USER")),
+        password=(os.getenv("DJ_PASS")),
     )
     yield connection
     dj.config["safemode"] = True
@@ -29,41 +35,58 @@ def connection_root():
 @pytest.fixture
 def connection_test(connection_root):
     """Test user database connection."""
-    database = f"{PREFIX}%%"
+    target = f"`{PREFIX}%%`.*"
     credentials = dict(
-        host=os.getenv("DJ_HOST"), user="datajoint", password="datajoint"
+        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"]}';
+            CREATE USER IF NOT EXISTS '{credentials['user']}'@'%%'
+            IDENTIFIED BY '{credentials['password']}';
             """
         )
         connection_root.query(
             f"""
-            GRANT {permission} ON `{database}`.*
-            TO '{credentials["user"]}'@'%%';
+            GRANT {permission} ON {target}
+            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"]}';
+            GRANT {permission} ON {target}
+            TO '{credentials['user']}'@'%%'
+            IDENTIFIED BY '{credentials['password']}';
             """
         )
-
-    connection = dj.Connection(**credentials)
+    connection = (dj.Connection)(**credentials)
     yield connection
-    connection_root.query(f"""DROP USER `{credentials["user"]}`""")
+    connection_root.query(f"DROP USER `{credentials['user']}`")
     connection.close()
+
+
+@pytest.fixture
+def bucket():
+    httpClient = urllib3.PoolManager(
+        timeout=30,
+        cert_reqs="CERT_REQUIRED",
+        ca_certs=(certifi.where()),
+        retries=urllib3.Retry(
+            total=3, backoff_factor=0.2, status_forcelist=[500, 502, 503, 504]
+        ),
+    )
+    minioClient = minio.Minio(
+        **{k: v for k, v in S3_CONN_INFO.items() if k != "bucket"},
+        **{"secure": True, "http_client": httpClient},
+    )
+    minioClient.make_bucket((S3_CONN_INFO["bucket"]), location="us-east-1")
+    yield
+    minioClient.remove_bucket(S3_CONN_INFO["bucket"])

tests/schemas/adapted.py

@@ -0,0 +1,109 @@
+import pytest, datajoint as dj
+from datajoint import errors
+import tempfile, networkx as nx, json
+from pathlib import Path
+from .. import PREFIX, S3_CONN_INFO, bucket, connection_test
+
+
+@pytest.fixture
+def schema(connection_test):
+    schema = dj.Schema((PREFIX + "_test_custom_datatype"), connection=connection_test)
+    yield schema
+    schema.drop()
+
+
+@pytest.fixture
+def graph():
+    class GraphAdapter(dj.AttributeAdapter):
+        attribute_type = "longblob"
+
+        @staticmethod
+        def get(obj):
+            return nx.Graph(obj)
+
+        @staticmethod
+        def put(obj):
+            assert isinstance(obj, nx.Graph)
+            return list(obj.edges)
+
+    graph = GraphAdapter()
+    yield graph
+
+
+@pytest.fixture
+def Connectivity(schema, graph):
+    errors._switch_adapted_types(True)
+
+    @schema
+    class Connectivity(dj.Manual):
+        definition = """
+        connid : int
+        ---
+        conn_graph = null : <graph>
+        """
+
+    yield Connectivity
+    Connectivity.drop()
+    errors._switch_adapted_types(False)
+
+
+@pytest.fixture
+def stores():
+    dj.config["stores"] = dict()
+    yield dj.config["stores"]
+    del dj.config["stores"]
+
+
+@pytest.fixture
+def store(schema, bucket, stores):
+    with tempfile.TemporaryDirectory() as stage_dir:
+        store_name = "repo-s3"
+        dj.config["stores"][store_name] = dict(
+            S3_CONN_INFO, protocol="s3", location="adapted/repo", stage=stage_dir
+        )
+        yield store_name
+        schema.external[store_name].delete(delete_external_files=True)
+        schema.connection.query(
+            f"DROP TABLE IF EXISTS `{schema.database}`.`~external_{store_name}`"
+        )
+        del dj.config["stores"][store_name]
+
+
+@pytest.fixture
+def Layout(schema, store, Connectivity):
+    errors._switch_adapted_types(True)
+    errors._switch_filepath_types(True)
+
+    class LayoutToFilepath(dj.AttributeAdapter):
+        __doc__ = """
+        An adapted data type that saves a graph layout into fixed filepath
+        """
+        attribute_type = f"filepath@{store}"
+
+        @staticmethod
+        def get(path):
+            with open(path, "r") as f:
+                return json.load(f)
+
+        @staticmethod
+        def put(layout):
+            path = Path(dj.config["stores"][store]["stage"], "layout.json")
+            with open(str(path), "w") as f:
+                json.dump(layout, f)
+            return path
+
+    layout_to_filepath = LayoutToFilepath()
+
+    @schema
+    class Layout(dj.Manual):
+        definition = """
+        # stores graph layout
+        -> Connectivity
+        ---
+        layout: <layout_to_filepath>
+        """
+
+    yield Layout
+    Layout.drop()
+    errors._switch_adapted_types(False)
+    errors._switch_filepath_types(False)