Bookmark tests

Author: technige

test/test_session.py

@@ -20,7 +20,9 @@
 
 from mock import patch
 from unittest import skipUnless
+from uuid import uuid4
 
+from neo4j.v1 import READ_ACCESS, WRITE_ACCESS
 from neo4j.v1.exceptions import CypherError, ResultError
 from neo4j.v1.session import GraphDatabase, basic_auth, Record
 from neo4j.v1.types import Node, Relationship, Path
@@ -449,13 +451,33 @@ def tearDown(self):
 
     @skipUnless(SERVER_VERSION >= (3, 1), "Bookmarking is not supported by this version of Neo4j")
     def test_can_obtain_bookmark_after_commit(self):
-        from neo4j.util import watch
-        watch("neo4j.bolt")
         with self.driver.session() as session:
             with session.begin_transaction() as tx:
                 tx.run("RETURN 1")
             assert session.last_bookmark is not None
 
+    @skipUnless(SERVER_VERSION >= (3, 1), "Bookmarking is not supported by this version of Neo4j")
+    def test_can_pass_bookmark_into_next_transaction(self):
+        unique_id = uuid4().hex
+
+        with self.driver.session(WRITE_ACCESS) as session:
+            with session.begin_transaction() as tx:
+                tx.run("CREATE (a:Thing {uuid:$uuid})", uuid=unique_id)
+            bookmark = session.last_bookmark
+
+        assert bookmark is not None
+
+        with self.driver.session(READ_ACCESS) as session:
+            with session.begin_transaction(bookmark) as tx:
+                result = tx.run("MATCH (a:Thing {uuid:$uuid}) RETURN a", uuid=unique_id)
+                record_list = list(result)
+                assert len(record_list) == 1
+                record = record_list[0]
+                assert len(record) == 1
+                thing = record[0]
+                assert isinstance(thing, Node)
+                assert thing["uuid"] == unique_id
+
 
 class ResultConsumptionTestCase(ServerTestCase):