[wip] more query tests

Author: timgraham

tests/encryption_/models.py

@@ -30,6 +30,15 @@
 from django_mongodb_backend.models import EmbeddedModel
 
 
+class Author(models.Model):
+    name = models.CharField(max_length=255)
+
+
+class Book(models.Model):
+    title = models.CharField(max_length=255)
+    author = models.ForeignKey(Author, models.CASCADE)
+
+
 class EncryptedTestModel(models.Model):
     class Meta:
         abstract = True

tests/encryption_/test_fields.py

@@ -1,6 +1,7 @@
 import datetime
 import uuid
 from decimal import Decimal
+from operator import attrgetter
 
 from bson import ObjectId
 from django.db import DatabaseError
@@ -20,6 +21,7 @@
     BigIntegerModel,
     Billing,
     BinaryModel,
+    Book,
     BooleanModel,
     CharModel,
     DateModel,
@@ -217,6 +219,23 @@ def test_time(self):
 
 
 class QueryTests(EncryptionTestCase):
+    def test_aggregate(self):
+        msg = (
+            "Aggregation stage $internalFacetTeeConsumer is not allowed or "
+            "supported with automatic encryption."
+        )
+        with self.assertRaisesMessage(DatabaseError, msg):
+            list(IntegerModel.objects.aggregate(Avg("value")))
+
+    def test_alias(self):
+        msg = (
+            "Cannot group on field '_id.value' which is encrypted with the "
+            "random algorithm or whose encryption properties are not known "
+            "until runtime"
+        )
+        with self.assertRaisesMessage(DatabaseError, msg):
+            list(IntegerModel.objects.alias(avg=Avg("value")))
+
     def test_annotate(self):
         msg = (
             "Cannot group on field '_id.value' which is encrypted with the "
@@ -226,6 +245,27 @@ def test_annotate(self):
         with self.assertRaisesMessage(DatabaseError, msg):
             list(IntegerModel.objects.annotate(avg=Avg("value")))
 
+    def test_bulk_create(self):
+        CharModel.objects.bulk_create([CharModel(value="abc"), CharModel(value="xyz")])
+        self.assertQuerySetEqual(
+            CharModel.objects.order_by("pk"), ["abc", "xyz"], attrgetter("value")
+        )
+
+    def test_bulk_update(self):
+        objs = [
+            CharModel.objects.create(value="abc"),
+            CharModel.objects.create(value="xyz"),
+        ]
+        objs[0].value = "def"
+        objs[1].value = "mno"
+        msg = "Multi-document updates are not allowed with Queryable Encryption"
+        with self.assertRaisesMessage(DatabaseError, msg):
+            CharModel.objects.bulk_update(objs, ["value"])
+
+    def test_contains(self):
+        obj = CharModel.objects.create(value="abc")
+        self.assertIs(CharModel.objects.contains(obj), True)
+
     def test_count(self):
         msg = (
             "Aggregation stage $internalFacetTeeConsumer is not allowed or "
@@ -248,32 +288,66 @@ def test_datetimes(self):
         with self.assertRaisesMessage(DatabaseError, msg):
             list(DateTimeModel.objects.datetimes("value", "year"))
 
+    def test_distinct(self):
+        msg = (
+            "Cannot group on field '_id.value' which is encrypted with the "
+            "random algorithm or whose encryption properties are not known "
+            "until runtime"
+        )
+        with self.assertRaisesMessage(DatabaseError, msg):
+            list(CharModel.objects.distinct("value"))
+
+    def test_exclude(self):
+        obj1 = CharModel.objects.create(value="abc")
+        obj2 = CharModel.objects.create(value="xyz")
+        self.assertSequenceEqual(CharModel.objects.exclude(value=obj1.value), [obj2])
+
     def test_exists(self):
         self.assertIs(CharModel.objects.exists(), False)
 
+    def test_get_or_create(self):
+        obj1, created1 = CharModel.objects.get_or_create(value="abc")
+        self.assertIs(created1, True)
+        obj2, created2 = CharModel.objects.get_or_create(value="abc")
+        self.assertIs(created2, False)
+        self.assertEqual(obj1, obj2)
+
+    def test_join(self):
+        list(Book.objects.filter(author__name="xxx"))
+
     def test_order_by(self):
         msg = "Cannot add an encrypted field as a prefix of another encrypted field"
         with self.assertRaisesMessage(DatabaseError, msg):
             list(CharModel.objects.order_by("value"))
 
+    def test_select_related(self):
+        list(Book.objects.select_related("author"))
+
+    def test_update(self):
+        msg = "Multi-document updates are not allowed with Queryable Encryption"
+        with self.assertRaisesMessage(DatabaseError, msg):
+            self.assertEqual(CharModel.objects.update(value="xyz"), 1)
+
+    def test_update_or_create(self):
+        # TODO
+        # msg = "Multi-document updates are not allowed with Queryable Encryption"
+        # with self.assertRaisesMessage(DatabaseError, msg):
+        # obj, created = CharModel.objects.update_or_create(value="xyz"), 1)
+        pass
+
+    def test_union(self):
+        msg = "Aggregation stage $unionWith is not allowed or supported with automatic encryption."
+        qs1 = IntegerModel.objects.filter(value__gt=1)
+        qs2 = IntegerModel.objects.filter(value__gte=8)
+        with self.assertRaisesMessage(DatabaseError, msg):
+            list(qs1.union(qs2))
+
     def test_values(self):
         list(CharModel.objects.values("value"))
 
     def test_values_list(self):
         list(CharModel.objects.values_list("value"))
 
-    def test_distinct(self):
-        msg = (
-            "Cannot group on field '_id.value' which is encrypted with the "
-            "random algorithm or whose encryption properties are not known "
-            "until runtime"
-        )
-        with self.assertRaisesMessage(DatabaseError, msg):
-            list(CharModel.objects.distinct("value"))
-
-    def test_only(self):
-        list(CharModel.objects.only("value"))
-
 
 class FieldMixinTests(EncryptionTestCase):
     def test_db_index(self):