Migrate most tests from #1084 to pytest

Author: ethho

tests/conftest.py

@@ -330,6 +330,8 @@ def schema_simp(connection_test, prefix):
     schema = dj.Schema(
         prefix + "_relational", schema_simple.LOCALS_SIMPLE, connection=connection_test
     )
+    schema(schema_simple.SelectPK)
+    schema(schema_simple.KeyPK)
     schema(schema_simple.IJ)
     schema(schema_simple.JI)
     schema(schema_simple.A)

tests/schema_simple.py

@@ -13,6 +13,26 @@
 import inspect
 
 
+@schema
+class SelectPK(dj.Lookup):
+    definition = """  # tests sql keyword escaping
+    id: int
+    select  : int
+    """
+    contents = list(dict(id=i, select=i * j)
+                    for i in range(3) for j in range(4, 0, -1))
+
+
+@schema
+class KeyPK(dj.Lookup):
+    definition = """  # tests sql keyword escaping
+    id   : int
+    key  : int
+    """
+    contents = list(dict(id=i, key=i + j)
+                    for i in range(3) for j in range(4, 0, -1))
+
+
 class IJ(dj.Lookup):
     definition = """  # tests restrictions
     i  : int

tests/test_declare.py

@@ -339,6 +339,17 @@ class WithSuchALongPartNameThatItCrashesMySQL(dj.Part):
         schema_any(WhyWouldAnyoneCreateATableNameThisLong)
 
 
+def test_regex_mismatch(schema_any):
+
+    class IndexAttribute(dj.Manual):
+        definition = """
+        index: int
+        """
+
+    with pytest.raises(dj.DataJointError):
+        schema_any(IndexAttribute)
+
+
 def test_table_name_with_underscores(schema_any):
     """
     Test issue #1150 -- Reject table names containing underscores. Tables should be in strict

tests/test_relational_operand.py

@@ -5,6 +5,7 @@
 import datetime
 import numpy as np
 import datajoint as dj
+from datajoint.errors import DataJointError
 from .schema_simple import *
 from .schema import *
 
@@ -570,3 +571,94 @@ def test_union_multiple(schema_simp_pop):
     y = set(zip(*q2.fetch("i", "j")))
     assert x == y
     assert q1.fetch(as_dict=True) == q2.fetch(as_dict=True)
+
+
+class TestDjTop:
+    """TODO: migrate"""
+
+    def test_restrictions_by_top(self):
+        a = L() & dj.Top()
+        b = L() & dj.Top(order_by=["cond_in_l", "KEY"])
+        x = L() & dj.Top(5, "id_l desc", 4) & "cond_in_l=1"
+        y = L() & "cond_in_l=1" & dj.Top(5, "id_l desc", 4)
+        z = (
+            L()
+            & dj.Top(None, order_by="id_l desc")
+            & "cond_in_l=1"
+            & dj.Top(5, "id_l desc")
+            & ("id_l=20", "id_l=16", "id_l=17")
+            & dj.Top(2, "id_l asc", 1)
+        )
+        assert len(a) == 1
+        assert len(b) == 1
+        assert len(x) == 1
+        assert len(y) == 5
+        assert len(z) == 2
+        assert a.fetch(as_dict=True) == [
+            {"id_l": 0, "cond_in_l": 1},
+        ]
+        assert b.fetch(as_dict=True) == [
+            {"id_l": 3, "cond_in_l": 0},
+        ]
+        assert x.fetch(as_dict=True) == [{"id_l": 25, "cond_in_l": 1}]
+        assert y.fetch(as_dict=True) == [
+            {"id_l": 16, "cond_in_l": 1},
+            {"id_l": 15, "cond_in_l": 1},
+            {"id_l": 11, "cond_in_l": 1},
+            {"id_l": 10, "cond_in_l": 1},
+            {"id_l": 5, "cond_in_l": 1},
+        ]
+        assert z.fetch(as_dict=True) == [
+            {"id_l": 17, "cond_in_l": 1},
+            {"id_l": 20, "cond_in_l": 1},
+        ]
+
+    def test_top_restriction_with_keywords(self):
+        select = SelectPK() & dj.Top(limit=9, order_by=["select desc"])
+        key = KeyPK() & dj.Top(limit=9, order_by="key desc")
+        assert select.fetch(as_dict=True) == [
+            {"id": 2, "select": 8},
+            {"id": 2, "select": 6},
+            {"id": 1, "select": 4},
+            {"id": 2, "select": 4},
+            {"id": 1, "select": 3},
+            {"id": 1, "select": 2},
+            {"id": 2, "select": 2},
+            {"id": 1, "select": 1},
+            {"id": 0, "select": 0},
+        ]
+        assert key.fetch(as_dict=True) == [
+            {"id": 2, "key": 6},
+            {"id": 2, "key": 5},
+            {"id": 1, "key": 5},
+            {"id": 0, "key": 4},
+            {"id": 1, "key": 4},
+            {"id": 2, "key": 4},
+            {"id": 0, "key": 3},
+            {"id": 1, "key": 3},
+            {"id": 2, "key": 3},
+        ]
+
+    def test_top_errors(self):
+        with assert_raises(DataJointError) as err1:
+            L() & ("cond_in_l=1", dj.Top())
+        with assert_raises(DataJointError) as err2:
+            L() & dj.AndList(["cond_in_l=1", dj.Top()])
+        with assert_raises(TypeError) as err3:
+            L() & dj.Top(limit="1")
+        with assert_raises(TypeError) as err4:
+            L() & dj.Top(order_by=1)
+        with assert_raises(TypeError) as err5:
+            L() & dj.Top(offset="1")
+        assert (
+            "Invalid restriction type Top(limit=1, order_by=['KEY'], offset=0)"
+            == str(err1.exception)
+        )
+        assert (
+            "Invalid restriction type Top(limit=1, order_by=['KEY'], offset=0)"
+            == str(err2.exception)
+        )
+        assert "Top limit must be an integer" == str(err3.exception)
+        assert "Top order_by attributes must all be strings" == str(
+            err4.exception)
+        assert "The offset argument must be an integer" == str(err5.exception)

tests/test_schema.py

@@ -210,6 +210,8 @@ def test_list_tables(schema_simp):
             "#website",
             "profile",
             "profile__website",
+            "#select_p_k",
+            "#key_p_k",
         ]
     )
     actual = set(schema_simp.list_tables())