Merge pull request #304 from poissoncorp/unskip-tests

Unskip/fix the 7.2 test suite: Corax, 7.x logging, topology cache, exception-dispatcher

Author: poissoncorp

ravendb/changes/database_changes.py

@@ -87,7 +87,12 @@ def _get_server_certificate(self) -> Optional[str]:
     def _connect_websocket_secured(self, url: str) -> None:
         # Get server certificate via HTTPS and prepare SSL context
         server_certificate = base64.b64decode(self._get_server_certificate())
-        ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
+        # PROTOCOL_TLS_CLIENT replaces the deprecated PROTOCOL_TLSv1_2 but defaults to CA verification and
+        # hostname checking, which the old PROTOCOL_TLSv1_2 context did not do by default. Clear both to keep
+        # the previous behavior; a trust store, when configured, re-enables CA verification below.
+        ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+        ssl_context.check_hostname = False
+        ssl_context.verify_mode = ssl.CERT_NONE
         ssl_context.load_cert_chain(self._request_executor.certificate_path)
         if self._request_executor.trust_store_path:
             ssl_context.verify_mode = ssl.CERT_REQUIRED

ravendb/documents/conventions.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 import inspect
+import os
+import sys
 import threading
 from abc import abstractmethod, ABC
 from datetime import timedelta, datetime
@@ -51,6 +53,10 @@ def __init__(self):
 
         # Flags
         self.disable_topology_updates = False
+        # On-disk topology cache (mirrors the .NET client): enabled by default. Topology is persisted to, and
+        # - when the initial urls are unreachable on startup - seeded from topology_cache_location.
+        self.disable_topology_cache = False
+        self.topology_cache_location: Optional[str] = DocumentConventions._default_topology_cache_location()
         self._optimistic_concurrency_mode = None
         # Track which setter the user touched so we can reject mixing them.
         self._use_optimistic_concurrency_was_set = False
@@ -107,6 +113,18 @@ def freeze(self):
     def is_frozen(self):
         return self._frozen
 
+    @staticmethod
+    def _default_topology_cache_location() -> str:
+        # Per-user, OS-appropriate cache directory: stable across restarts and never clutters the directory
+        # the application happens to run from (unlike cwd). Mirrors the intent of .NET's AppContext.BaseDirectory.
+        if sys.platform == "win32":
+            base = os.environ.get("LOCALAPPDATA") or os.path.join(os.path.expanduser("~"), "AppData", "Local")
+        elif sys.platform == "darwin":
+            base = os.path.join(os.path.expanduser("~"), "Library", "Caches")
+        else:
+            base = os.environ.get("XDG_CACHE_HOME") or os.path.join(os.path.expanduser("~"), ".cache")
+        return os.path.join(base, "ravendb", "topology")
+
     def get_python_class_name(self, entity_type: type):
         return self._find_python_class_name(entity_type)
 

ravendb/documents/operations/schema_validation/__init__.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import json
-from datetime import datetime
+from datetime import datetime, timezone
 from typing import Optional, Dict, Any, List
 
 import requests
@@ -23,7 +23,7 @@ def __init__(
     ):
         self.schema = schema
         self.disabled = disabled
-        self.last_modified_time = last_modified_time or datetime.utcnow()
+        self.last_modified_time = last_modified_time or datetime.now(timezone.utc).replace(tzinfo=None)
 
     def to_json(self) -> Dict[str, Any]:
         return {

ravendb/documents/operations/time_series.py

@@ -101,8 +101,8 @@ def from_json(cls, json_dict: Dict[str, Any]) -> TimeSeriesCollectionConfigurati
     def to_json(self) -> Dict[str, Any]:
         return {
             "Disabled": self.disabled,
-            "Policies": [policy.to_json() for policy in self.policies],
-            "RawPolicy": self.raw_policy.to_json(),
+            "Policies": [policy.to_json() for policy in (self.policies or [])],
+            "RawPolicy": self.raw_policy.to_json() if self.raw_policy is not None else None,
         }
 
 

ravendb/documents/subscriptions/worker.py

@@ -744,11 +744,12 @@ def __run_async() -> None:
 
     def _assert_last_connection_failure(self) -> None:
         if self._last_connection_failure is None:
-            self._last_connection_failure = datetime.datetime.utcnow()
+            self._last_connection_failure = datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None)
             return
 
         if (
-            datetime.datetime.utcnow().timestamp() - self._last_connection_failure.timestamp()
+            datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None).timestamp()
+            - self._last_connection_failure.timestamp()
             > self._options.max_erroneous_period.total_seconds()
         ):
             raise SubscriptionInvalidStateException(

ravendb/http/request_executor.py

@@ -39,6 +39,7 @@
 from ravendb.http.raven_command import RavenCommand, RavenCommandResponseType
 from ravendb.http.server_node import ServerNode
 from ravendb.http.topology import Topology, NodeStatus, NodeSelector, CurrentIndexAndNode, UpdateTopologyParameters
+from ravendb.http import topology_local_cache
 from ravendb.serverwide.commands import GetDatabaseTopologyCommand, GetClusterTopologyCommand
 
 from http import HTTPStatus
@@ -416,6 +417,14 @@ def __supply_async():
 
                 self._topology_etag = self._node_selector.topology.etag
 
+                if not self.conventions.disable_topology_cache and self.conventions.topology_cache_location:
+                    topology_local_cache.try_save(
+                        self.conventions.topology_cache_location,
+                        topology_local_cache.server_hash(parameters.node.url, self._database_name),
+                        self._node_selector.topology,
+                        topology_local_cache.DATABASE_TOPOLOGY_EXTENSION,
+                    )
+
                 self._on_topology_updated_invoke(topology)
             except Exception as e:
                 if not self._disposed:
@@ -456,6 +465,16 @@ def __run(errors: list):
 
                     errors.append((url, e))
 
+            # all initial urls unreachable - fall back to the on-disk topology cache when enabled
+            for url in initial_urls:
+                cached_topology = self._try_load_topology_from_cache(url)
+                if cached_topology is not None:
+                    self._node_selector = NodeSelector(cached_topology, self._thread_pool_executor)
+                    self._topology_etag = cached_topology.etag
+                    self.__initialize_update_topology_timer()
+                    self.__topology_taken_from_node = ServerNode(url, self._database_name)
+                    return
+
             topology = Topology(
                 self._topology_etag,
                 (
@@ -476,6 +495,13 @@ def __run(errors: list):
 
         return self._thread_pool_executor.submit(__run, errors)
 
+    def _try_load_topology_from_cache(self, url):
+        return topology_local_cache.try_load(
+            None if self.conventions.disable_topology_cache else self.conventions.topology_cache_location,
+            topology_local_cache.server_hash(url, self._database_name),
+            topology_local_cache.DATABASE_TOPOLOGY_EXTENSION,
+        )
+
     @staticmethod
     def validate_urls(initial_urls: List[str]) -> List[str]:
         # todo: implement validation
@@ -635,7 +661,7 @@ def execute(
                     return  # we either handled this already in the unsuccessful response or we are throwing
                 self._on_succeed_request_invoke(self._database_name, url, response, request, attempt_num)
                 response_dispose = command.process_response(self._cache, response, url)
-                self._last_returned_response = datetime.datetime.utcnow()
+                self._last_returned_response = datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None)
             finally:
                 if response_dispose == ResponseDisposeHandling.AUTOMATIC:
                     response.close()
@@ -1445,6 +1471,14 @@ def __supply_async():
                 new_topology = Topology(results.etag, nodes)
                 self._topology_etag = results.etag
 
+                if not self.conventions.disable_topology_cache and self.conventions.topology_cache_location:
+                    topology_local_cache.try_save(
+                        self.conventions.topology_cache_location,
+                        topology_local_cache.server_hash(parameters.node.url),
+                        new_topology,
+                        topology_local_cache.CLUSTER_TOPOLOGY_EXTENSION,
+                    )
+
                 if self._node_selector is None:
                     self._node_selector = NodeSelector(new_topology, self._thread_pool_executor)
 
@@ -1469,5 +1503,12 @@ def __supply_async():
 
         return self._thread_pool_executor.submit(__supply_async)
 
+    def _try_load_topology_from_cache(self, url):
+        return topology_local_cache.try_load(
+            None if self.conventions.disable_topology_cache else self.conventions.topology_cache_location,
+            topology_local_cache.server_hash(url),
+            topology_local_cache.CLUSTER_TOPOLOGY_EXTENSION,
+        )
+
     def _throw_exceptions(self, details: str):
         raise RuntimeError(f"Failed to retrieve cluster topology from all known nodes {os.linesep}{details}")

ravendb/http/server_node.py

@@ -47,6 +47,24 @@ def __hash__(self) -> int:
     def last_server_version(self) -> str:
         return self.__last_server_version
 
+    def to_json(self) -> dict:
+        return {
+            "Url": self.url,
+            "Database": self.database,
+            "ClusterTag": self.cluster_tag,
+            "ServerRole": self.server_role.value if self.server_role is not None else None,
+        }
+
+    @classmethod
+    def from_json(cls, json_dict: dict) -> "ServerNode":
+        role = json_dict.get("ServerRole")
+        return cls(
+            json_dict.get("Url"),
+            json_dict.get("Database"),
+            json_dict.get("ClusterTag"),
+            cls.Role(role) if role else None,
+        )
+
     @classmethod
     def create_from(cls, topology: "ClusterTopology"):
         nodes = []

ravendb/http/topology.py

@@ -25,6 +25,16 @@ def __init__(self, etag: int, nodes: List[ServerNode]):
         self.etag = etag
         self.nodes = nodes
 
+    def to_json(self) -> Dict:
+        return {"Etag": self.etag, "Nodes": [node.to_json() for node in (self.nodes or [])]}
+
+    @classmethod
+    def from_json(cls, json_dict: Dict) -> "Topology":
+        return cls(
+            json_dict.get("Etag"),
+            [ServerNode.from_json(node_json) for node_json in (json_dict.get("Nodes") or [])],
+        )
+
 
 class ClusterTopology:
     def __init__(self):

ravendb/http/topology_local_cache.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+from typing import Optional
+
+from ravendb.http.topology import Topology
+
+# On-disk topology cache (mirrors the .NET client's TopologyLocalCache). The feature is
+# opt-in: it is active only when DocumentConventions.topology_cache_location is set.
+
+DATABASE_TOPOLOGY_EXTENSION = ".raven-topology"
+CLUSTER_TOPOLOGY_EXTENSION = ".raven-cluster-topology"
+
+
+def server_hash(url: str, database: Optional[str] = None) -> str:
+    key = "{0}{1}".format(url, database if database else "")
+    return hashlib.md5(key.encode("utf-8")).hexdigest()
+
+
+def _path(location: str, topology_hash: str, extension: str) -> str:
+    return os.path.join(location, topology_hash + extension)
+
+
+def try_save(location: Optional[str], topology_hash: str, topology: Topology, extension: str) -> None:
+    # Best-effort: caching failures must never break a topology update.
+    if not location or topology is None:
+        return
+    try:
+        os.makedirs(location, exist_ok=True)
+        with open(_path(location, topology_hash, extension), "w", encoding="utf-8") as stream:
+            json.dump(topology.to_json(), stream)
+    except Exception:
+        pass
+
+
+def try_load(location: Optional[str], topology_hash: str, extension: str) -> Optional[Topology]:
+    if not location:
+        return None
+    try:
+        path = _path(location, topology_hash, extension)
+        if not os.path.isfile(path):
+            return None
+        with open(path, "r", encoding="utf-8") as stream:
+            return Topology.from_json(json.load(stream))
+    except Exception:
+        return None