TestKit backend: fix warning assertions

Author: robsdedude

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/requests.py

@@ -14,15 +14,21 @@
 # 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 json
 from os import path
-import warnings
 
 import neo4j
 import testkitbackend.fromtestkit as fromtestkit
 import testkitbackend.totestkit as totestkit
 from testkitbackend.fromtestkit import to_meta_and_timeout
 
+from ._warning_check import (
+    warning_check,
+    warnings_check,
+)
+
 
 class FrontendError(Exception):
     pass
@@ -98,9 +104,21 @@ def NewDriver(backend, data):
     data.mark_item_as_read_if_equals("livenessCheckTimeoutMs", None)
 
     data.mark_item_as_read("domainNameResolverRegistered")
-    driver = neo4j.GraphDatabase.driver(
-        data["uri"], auth=auth, user_agent=data["userAgent"], **kwargs
-    )
+    expected_warnings = []
+    if "update_routing_table_timeout" in kwargs:
+        expected_warnings.append((
+            DeprecationWarning,
+            "The 'update_routing_table_timeout' config key is deprecated"
+        ))
+    if "session_connection_timeout" in kwargs:
+        expected_warnings.append((
+            DeprecationWarning,
+            "The 'session_connection_timeout' config key is deprecated"
+        ))
+    with warnings_check(expected_warnings):
+        driver = neo4j.GraphDatabase.driver(
+            data["uri"], auth=auth, user_agent=data["userAgent"], **kwargs
+        )
     key = backend.next_key()
     backend.drivers[key] = driver
     backend.send_response("Driver", {"id": key})
@@ -109,19 +127,26 @@ def NewDriver(backend, data):
 def VerifyConnectivity(backend, data):
     driver_id = data["driverId"]
     driver = backend.drivers[driver_id]
-    with warnings.catch_warnings():
-        warnings.simplefilter("ignore", category=neo4j.ExperimentalWarning)
+    with warning_check(
+        neo4j.ExperimentalWarning,
+            "The configuration may change in the future."
+    ):
         driver.verify_connectivity()
     backend.send_response("Driver", {"id": driver_id})
 
 
 def CheckMultiDBSupport(backend, data):
     driver_id = data["driverId"]
     driver = backend.drivers[driver_id]
-    backend.send_response(
-        "MultiDBSupport",
-        {"id": backend.next_key(), "available": driver.supports_multi_db()}
-    )
+    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", {
+        "id": backend.next_key(), "available": available
+    })
 
 
 def resolution_func(backend, custom_resolver=False, custom_dns_resolver=False):