TST: page_length parameter

Summary

Tests for caosdb-pylib!110 (merged)

Related issue: https://gitlab.indiscale.com/caosdb/customers/geomar/management/-/issues/159

tests/test_query.py

@@ -1334,3 +1334,80 @@ def test_find_query_default_role():
     db.administration.set_server_property("FIND_QUERY_DEFAULT_ROLE", "asdf")
     with raises(TransactionError):
         assert db.execute_query("FIND Test*", unique=True).id == rec.id
+
+
+def test_query_paging():
+    test_query = "FIND RECORDTYPE TestRT*"
+
+    # insert 20 entities
+    for i in range(20):
+        db.RecordType(f"TestRT{i}").insert()
+
+    assert len(db.execute_query(test_query)) == 20
+
+    # paging off
+    assert isinstance(db.execute_query(test_query, page_length=0), db.Container)
+    assert isinstance(db.execute_query(test_query, page_length=-1), db.Container)
+    assert isinstance(db.execute_query(test_query, page_length=False), db.Container)
+    assert isinstance(db.execute_query(test_query, page_length=None), db.Container)
+    assert isinstance(
+        db.execute_query(
+            test_query +
+            " WITH name=TestRT0",
+            page_length=10,
+            unique=True),
+        db.RecordType)
+    assert isinstance(db.execute_query("COUNT RECORDTYPE", page_length=10), int)
+
+    # zero pages
+    assert len([page for page in db.execute_query(
+        test_query + " WITH unmet_condition", page_length=109)]) == 0
+    # two pages
+    assert len([page for page in db.execute_query(test_query, page_length=10)]) == 2
+    assert len([page for page in db.execute_query(test_query, page_length=11)]) == 2
+    assert len([page for page in db.execute_query(test_query, page_length=19)]) == 2
+    # one page
+    assert len([page for page in db.execute_query(test_query, page_length=20)]) == 1
+    assert len([page for page in db.execute_query(test_query, page_length=100)]) == 1
+
+    # concat
+    assert len([entity for page in db.execute_query(test_query, page_length=10)
+                for entity in page]) == 20
+
+    # concat zero pages
+    assert len([entity for page in db.execute_query(
+        test_query + " WITH unmet_condition", page_length=10)
+        for entity in page]) == 0
+
+    # this kind of update is allowed
+    for entity in [entity for page in db.execute_query(
+            test_query, page_length=10) for entity in page]:
+        entity.description = "updated"
+        entity.update()
+
+    # as well as this
+    container = db.Container().extend(
+        [entity for page in db.execute_query(test_query, page_length=10) for entity in page])
+    for entity in container:
+        entity.description = "updated 2"
+    container.update()
+
+    # but not this
+    with raises(db.exceptions.PagingConsistencyError):
+        for page in db.execute_query(test_query, page_length=10):
+            for entity in page:
+                entity.description = "update not allowed"
+                entity.update()
+
+    # not ideal: the first page has been written:
+    pages = db.execute_query(test_query, page_length=10)
+    for entity in next(pages):
+        assert entity.description == "update not allowed"
+    for entity in next(pages):
+        assert entity.description == "updated 2"
+
+    # the above pattern is ok if you don't write, tho
+    for page in db.execute_query(test_query, page_length=10):
+        for entity in page:
+            assert entity.description is not None
+            # don't: entity.update()