TestKit backend: fix warning assertions

Author: robsdedude

neo4j/_async/driver.py

@@ -675,6 +675,5 @@ def __init__(self, pool, default_workspace_config):
 
         def session(self, **config) -> AsyncSession:
             session_config = SessionConfig(self._default_workspace_config,
-                                           config)
-            SessionConfig.consume(config)  # Consume the config
+                                           SessionConfig.consume(config))
             return AsyncSession(self._pool, session_config)

neo4j/_meta.py

@@ -17,6 +17,7 @@
 
 
 import asyncio
+import tracemalloc
 import typing as t
 from functools import wraps
 from warnings import warn
@@ -130,8 +131,6 @@ def inner(*args, **kwargs):
 
 
 def unclosed_resource_warn(obj):
-    import tracemalloc
-    from warnings import warn
     msg = f"Unclosed {obj!r}."
     trace = tracemalloc.get_object_traceback(obj)
     if trace:

neo4j/_sync/driver.py

@@ -674,6 +674,5 @@ def __init__(self, pool, default_workspace_config):
 
         def session(self, **config) -> Session:
             session_config = SessionConfig(self._default_workspace_config,
-                                           config)
-            SessionConfig.consume(config)  # Consume the config
+                                           SessionConfig.consume(config))
             return Session(self._pool, session_config)

testkitbackend/_async/requests.py

@@ -21,8 +21,6 @@
 import warnings
 from os import path
 
-import pytest
-
 import neo4j
 import neo4j.api
 from neo4j._async_compat.util import AsyncUtil
@@ -32,6 +30,7 @@
     test_subtest_skips,
     totestkit,
 )
+from .._warning_check import warning_check
 from ..exceptions import MarkdAsDriverException
 
 
@@ -178,10 +177,10 @@ async def GetServerInfo(backend, data):
 async def CheckMultiDBSupport(backend, data):
     driver_id = data["driverId"]
     driver = backend.drivers[driver_id]
-    with pytest.warns(
-        DeprecationWarning,
-        match="Feature support query, based on Bolt protocol version and "
-              "Neo4j server version will change in the future."
+    with warning_check(
+        neo4j.ExperimentalWarning,
+        "Feature support query, based on Bolt protocol version and Neo4j "
+        "server version will change in the future."
     ):
         available = await driver.supports_multi_db()
     await backend.send_response("MultiDBSupport", {
@@ -266,7 +265,14 @@ async def NewBookmarkManager(backend, data):
             backend, bookmark_manager_id
         )
 
-    bookmark_manager = neo4j.AsyncGraphDatabase.bookmark_manager(**bmm_kwargs)
+    with warning_check(
+        neo4j.ExperimentalWarning,
+        "The bookmark manager feature is experimental. It might be changed or "
+        "removed any time even without prior notice."
+    ):
+        bookmark_manager = neo4j.AsyncGraphDatabase.bookmark_manager(
+            **bmm_kwargs
+        )
     backend.bookmark_managers[bookmark_manager_id] = bookmark_manager
     await backend.send_response("BookmarkManager", {"id": bookmark_manager_id})
 
@@ -382,7 +388,15 @@ async def NewSession(backend, data):
     ):
         if data_name in data:
             config[conf_name] = data[data_name]
-    session = driver.session(**config)
+    if "bookmark_manager" in config:
+        with warning_check(
+            neo4j.ExperimentalWarning,
+            "The 'bookmark_manager' config key is experimental. It might be "
+            "changed or removed any time even without prior notice."
+        ):
+            session = driver.session(**config)
+    else:
+        session = driver.session(**config)
     key = backend.next_key()
     backend.sessions[key] = SessionTracker(session)
     await backend.send_response("Session", {"id": key})

testkitbackend/_sync/requests.py

@@ -21,8 +21,6 @@
 import warnings
 from os import path
 
-import pytest
-
 import neo4j
 import neo4j.api
 from neo4j._async_compat.util import Util
@@ -32,6 +30,7 @@
     test_subtest_skips,
     totestkit,
 )
+from .._warning_check import warning_check
 from ..exceptions import MarkdAsDriverException
 
 
@@ -178,10 +177,10 @@ def GetServerInfo(backend, data):
 def CheckMultiDBSupport(backend, data):
     driver_id = data["driverId"]
     driver = backend.drivers[driver_id]
-    with pytest.warns(
-        DeprecationWarning,
-        match="Feature support query, based on Bolt protocol version and "
-              "Neo4j server version will change in the future."
+    with warning_check(
+        neo4j.ExperimentalWarning,
+        "Feature support query, based on Bolt protocol version and Neo4j "
+        "server version will change in the future."
     ):
         available = driver.supports_multi_db()
     backend.send_response("MultiDBSupport", {
@@ -266,7 +265,14 @@ def NewBookmarkManager(backend, data):
             backend, bookmark_manager_id
         )
 
-    bookmark_manager = neo4j.GraphDatabase.bookmark_manager(**bmm_kwargs)
+    with warning_check(
+        neo4j.ExperimentalWarning,
+        "The bookmark manager feature is experimental. It might be changed or "
+        "removed any time even without prior notice."
+    ):
+        bookmark_manager = neo4j.GraphDatabase.bookmark_manager(
+            **bmm_kwargs
+        )
     backend.bookmark_managers[bookmark_manager_id] = bookmark_manager
     backend.send_response("BookmarkManager", {"id": bookmark_manager_id})
 
@@ -382,7 +388,15 @@ def NewSession(backend, data):
     ):
         if data_name in data:
             config[conf_name] = data[data_name]
-    session = driver.session(**config)
+    if "bookmark_manager" in config:
+        with warning_check(
+            neo4j.ExperimentalWarning,
+            "The 'bookmark_manager' config key is experimental. It might be "
+            "changed or removed any time even without prior notice."
+        ):
+            session = driver.session(**config)
+    else:
+        session = driver.session(**config)
     key = backend.next_key()
     backend.sessions[key] = SessionTracker(session)
     backend.send_response("Session", {"id": key})

testkitbackend/_warning_check.py

@@ -0,0 +1,64 @@
+# Copyright (c) "Neo4j"
+# Neo4j Sweden AB [https://neo4j.com]
+#
+# This file is part of Neo4j.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+import re
+import warnings
+from contextlib import contextmanager
+
+
+@contextmanager
+def warning_check(category, message):
+    with warnings.catch_warnings(record=True) as warn_log:
+        warnings.filterwarnings("always", category=category, message=message)
+        yield
+    if len(warn_log) != 1:
+        raise AssertionError("Expected 1 warning, found %d: %s"
+                             % (len(warn_log), warn_log))
+
+
+@contextmanager
+def warnings_check(category_message_pairs):
+    with warnings.catch_warnings(record=True) as warn_log:
+        for category, message in category_message_pairs:
+            warnings.filterwarnings("always", category=category,
+                                    message=message)
+        yield
+    if len(warn_log) != len(category_message_pairs):
+        raise AssertionError(
+            "Expected %d warnings, found %d: %s"
+            % (len(category_message_pairs), len(warn_log), warn_log)
+        )
+    category_message_pairs = [
+        (category, re.compile(message, re.I))
+        for category, message in category_message_pairs
+    ]
+    for category, matcher in category_message_pairs:
+        match = None
+        for i, warning in enumerate(warn_log):
+            if (
+                warning.category == category
+                and matcher.match(warning.message.args[0])
+            ):
+                match = i
+                break
+        if match is None:
+            raise AssertionError(
+                "Expected warning not found: %r %r"
+                % (category, matcher.pattern)
+            )
+        warn_log.pop(match)

testkitbackend/totestkit.py

@@ -18,8 +18,6 @@
 
 import math
 
-import pytest
-
 from neo4j.graph import (
     Node,
     Path,
@@ -36,6 +34,8 @@
     Time,
 )
 
+from ._warning_check import warning_check
+
 
 def record(rec):
     fields = []
@@ -75,9 +75,8 @@ def to(name, val):
     if isinstance(v, (bytes, bytearray)):
         return to("CypherBytes", " ".join("{:02x}".format(byte) for byte in v))
     if isinstance(v, Node):
-        with pytest.warns(
-            DeprecationWarning,
-            match="`id` is deprecated, use `element_id` instead"
+        with warning_check(
+            DeprecationWarning, "`id` is deprecated, use `element_id` instead"
         ):
             id_ = v.id
         node = {
@@ -88,19 +87,16 @@ def to(name, val):
         }
         return {"name": "Node", "data": node}
     if isinstance(v, Relationship):
-        with pytest.warns(
-            DeprecationWarning,
-            match="`id` is deprecated, use `element_id` instead"
+        with warning_check(
+            DeprecationWarning, "`id` is deprecated, use `element_id` instead"
         ):
             id_ = v.id
-        with pytest.warns(
-            DeprecationWarning,
-            match="`id` is deprecated, use `element_id` instead"
+        with warning_check(
+            DeprecationWarning, "`id` is deprecated, use `element_id` instead"
         ):
             start_id = v.start_node.id
-        with pytest.warns(
-            DeprecationWarning,
-            match="`id` is deprecated, use `element_id` instead"
+        with warning_check(
+            DeprecationWarning, "`id` is deprecated, use `element_id` instead"
         ):
             end_id = v.end_node.id
         rel = {