From 797ab1fa080dbfaaac0ffbab85ce2641800a9b44 Mon Sep 17 00:00:00 2001 From: reforge Date: Thu, 27 Aug 2026 23:34:31 -0300 Subject: [PATCH 1/8] Add Azure Service Bus queue connection settings Adds the AzureServiceBus broker type to the queue connection string, with its shared-access, Entra ID, and passwordless authentication settings. The serializer omits the credential members that are not set, which is what the server expects when it picks the authentication method. The reference connection-validation helpers stay out: the sibling Azure Queue Storage and Amazon SQS settings do not carry them either, because validation runs on the server. Reforge-Run: 20260901T212227Z-1941916-reforge --- .../azure_service_bus_connection_settings.py | 80 +++++++++++++ .../operations/etl/queue/connection.py | 14 +++ ...t_azure_service_bus_connection_settings.py | 109 ++++++++++++++++++ 3 files changed, 203 insertions(+) create mode 100644 ravendb/documents/operations/etl/queue/azure_service_bus_connection_settings.py create mode 100644 ravendb/tests/operations_tests/test_azure_service_bus_connection_settings.py diff --git a/ravendb/documents/operations/etl/queue/azure_service_bus_connection_settings.py b/ravendb/documents/operations/etl/queue/azure_service_bus_connection_settings.py new file mode 100644 index 00000000..22ef5fe3 --- /dev/null +++ b/ravendb/documents/operations/etl/queue/azure_service_bus_connection_settings.py @@ -0,0 +1,80 @@ +from typing import Optional, Dict, Any + + +class AzureServiceBusEntraId: + def __init__( + self, + namespace: Optional[str] = None, + tenant_id: Optional[str] = None, + client_id: Optional[str] = None, + client_secret: Optional[str] = None, + ): + self.namespace = namespace + self.tenant_id = tenant_id + self.client_id = client_id + self.client_secret = client_secret + + def to_json(self) -> Dict[str, Any]: + return { + "Namespace": self.namespace, + "TenantId": self.tenant_id, + "ClientId": self.client_id, + "ClientSecret": self.client_secret, + } + + @classmethod + def from_json(cls, json_dict: Optional[Dict[str, Any]]): + return cls( + namespace=json_dict.get("Namespace"), + tenant_id=json_dict.get("TenantId"), + client_id=json_dict.get("ClientId"), + client_secret=json_dict.get("ClientSecret"), + ) + + +class AzureServiceBusPasswordless: + def __init__(self, namespace: Optional[str] = None): + self.namespace = namespace + + def to_json(self) -> Dict[str, Any]: + return { + "Namespace": self.namespace, + } + + @classmethod + def from_json(cls, json_dict: Optional[Dict[str, Any]]): + return cls( + namespace=json_dict.get("Namespace"), + ) + + +class AzureServiceBusConnectionSettings: + def __init__( + self, + connection_string: Optional[str] = None, + entra_id: Optional[AzureServiceBusEntraId] = None, + passwordless: Optional[AzureServiceBusPasswordless] = None, + ): + self.connection_string = connection_string + self.entra_id = entra_id + self.passwordless = passwordless + + def to_json(self) -> Dict[str, Any]: + json_dict = {} + if self.connection_string: + json_dict["ConnectionString"] = self.connection_string + if self.entra_id: + json_dict["EntraId"] = self.entra_id.to_json() + if self.passwordless: + json_dict["Passwordless"] = self.passwordless.to_json() + return json_dict + + @classmethod + def from_json(cls, json_dict: Optional[Dict[str, Any]]): + entra_id_dict = json_dict.get("EntraId") + passwordless_dict = json_dict.get("Passwordless") + return cls( + connection_string=json_dict.get("ConnectionString"), + entra_id=AzureServiceBusEntraId.from_json(entra_id_dict) if entra_id_dict else None, + passwordless=AzureServiceBusPasswordless.from_json(passwordless_dict) if passwordless_dict else None, + ) diff --git a/ravendb/documents/operations/etl/queue/connection.py b/ravendb/documents/operations/etl/queue/connection.py index 985d65d0..02709c9a 100644 --- a/ravendb/documents/operations/etl/queue/connection.py +++ b/ravendb/documents/operations/etl/queue/connection.py @@ -6,6 +6,9 @@ from ravendb.documents.operations.etl.queue.azure_queue_storage_connection_settings import ( AzureQueueStorageConnectionSettings, ) +from ravendb.documents.operations.etl.queue.azure_service_bus_connection_settings import ( + AzureServiceBusConnectionSettings, +) from ravendb.documents.operations.etl.queue.kafka_connection_settings import KafkaConnectionSettings from ravendb.documents.operations.etl.queue.rabbit_mq_connection_settings import RabbitMqConnectionSettings @@ -16,6 +19,7 @@ class QueueBrokerType(Enum): RABBIT_MQ = "RabbitMq" AZURE_QUEUE_STORAGE = "AzureQueueStorage" AMAZON_SQS = "AmazonSqs" + AZURE_SERVICE_BUS = "AzureServiceBus" class QueueConnectionString(ConnectionString): @@ -27,6 +31,7 @@ def __init__( rabbit_mq_settings: RabbitMqConnectionSettings = None, azure_queue_storage_settings: AzureQueueStorageConnectionSettings = None, amazon_sqs_settings: AmazonSqsConnectionSettings = None, + azure_service_bus_settings: AzureServiceBusConnectionSettings = None, ): super().__init__(name) self.broker_type = broker_type @@ -34,6 +39,7 @@ def __init__( self.rabbit_mq_settings = rabbit_mq_settings self.azure_queue_storage_settings = azure_queue_storage_settings self.amazon_sqs_settings = amazon_sqs_settings + self.azure_service_bus_settings = azure_service_bus_settings @property def get_type(self): @@ -49,6 +55,9 @@ def to_json(self): self.azure_queue_storage_settings.to_json() if self.azure_queue_storage_settings else None ), "AmazonSqsConnectionSettings": self.amazon_sqs_settings.to_json() if self.amazon_sqs_settings else None, + "AzureServiceBusConnectionSettings": ( + self.azure_service_bus_settings.to_json() if self.azure_service_bus_settings else None + ), "Type": ravendb.serverwide.server_operation_executor.ConnectionStringType.QUEUE, } @@ -77,4 +86,9 @@ def from_json(cls, json_dict: dict) -> "QueueConnectionString": if json_dict["AmazonSqsConnectionSettings"] else None ), + azure_service_bus_settings=( + AzureServiceBusConnectionSettings.from_json(json_dict["AzureServiceBusConnectionSettings"]) + if json_dict.get("AzureServiceBusConnectionSettings") + else None + ), ) diff --git a/ravendb/tests/operations_tests/test_azure_service_bus_connection_settings.py b/ravendb/tests/operations_tests/test_azure_service_bus_connection_settings.py new file mode 100644 index 00000000..2b887188 --- /dev/null +++ b/ravendb/tests/operations_tests/test_azure_service_bus_connection_settings.py @@ -0,0 +1,109 @@ +"""Azure Service Bus queue connection string tests.""" + +import os +import unittest + +from ravendb.documents.operations.connection_string.put_connection_string_operation import ( + PutConnectionStringOperation, +) +from ravendb.documents.operations.connection_string.get_connection_string_operation import ( + GetConnectionStringsOperation, +) +from ravendb.documents.operations.etl.queue.azure_service_bus_connection_settings import ( + AzureServiceBusConnectionSettings, + AzureServiceBusEntraId, + AzureServiceBusPasswordless, +) +from ravendb.documents.operations.etl.queue.connection import ( + QueueBrokerType, + QueueConnectionString, +) +from ravendb.serverwide.server_operation_executor import ConnectionStringType +from ravendb.tests.test_base import TestBase + + +class TestAzureServiceBusConnectionSettings(unittest.TestCase): + def test_to_json_omits_empty_connection_string(self): + settings = AzureServiceBusConnectionSettings(connection_string="") + self.assertEqual({}, settings.to_json()) + + def test_to_json_with_connection_string(self): + settings = AzureServiceBusConnectionSettings( + connection_string="Endpoint=sb://ns.servicebus.windows.net/;SharedAccessKeyName=k;SharedAccessKey=s=" + ) + self.assertEqual( + settings.to_json(), + {"ConnectionString": "Endpoint=sb://ns.servicebus.windows.net/;SharedAccessKeyName=k;SharedAccessKey=s="}, + ) + + def test_to_json_with_entra_id_and_passwordless(self): + settings = AzureServiceBusConnectionSettings( + entra_id=AzureServiceBusEntraId( + namespace="ns.servicebus.windows.net", + tenant_id="t", + client_id="c", + client_secret="s", + ), + passwordless=AzureServiceBusPasswordless(namespace="ns.servicebus.windows.net"), + ) + payload = settings.to_json() + self.assertEqual( + payload["EntraId"], + { + "Namespace": "ns.servicebus.windows.net", + "TenantId": "t", + "ClientId": "c", + "ClientSecret": "s", + }, + ) + self.assertEqual(payload["Passwordless"], {"Namespace": "ns.servicebus.windows.net"}) + + def test_queue_connection_string_wire(self): + connection_string = QueueConnectionString( + name="asb-cs", + broker_type=QueueBrokerType.AZURE_SERVICE_BUS, + azure_service_bus_settings=AzureServiceBusConnectionSettings( + passwordless=AzureServiceBusPasswordless(namespace="ns.servicebus.windows.net") + ), + ) + payload = connection_string.to_json() + self.assertEqual("AzureServiceBus", payload["BrokerType"]) + self.assertEqual( + {"Passwordless": {"Namespace": "ns.servicebus.windows.net"}}, + payload["AzureServiceBusConnectionSettings"], + ) + parsed = QueueConnectionString.from_json(payload) + self.assertEqual(QueueBrokerType.AZURE_SERVICE_BUS, parsed.broker_type) + self.assertEqual( + "ns.servicebus.windows.net", + parsed.azure_service_bus_settings.passwordless.namespace, + ) + + +@unittest.skipIf( + os.environ.get("RAVENDB_LICENSE") is None and os.environ.get("RAVEN_License") is None, + "Insufficient license permissions. Skipping on CI/CD.", +) +class TestAzureServiceBusRoundTrip(TestBase): + def test_put_and_get_queue_connection_string(self): + connection_string = QueueConnectionString( + name="asb-cs", + broker_type=QueueBrokerType.AZURE_SERVICE_BUS, + azure_service_bus_settings=AzureServiceBusConnectionSettings( + passwordless=AzureServiceBusPasswordless(namespace="ns.servicebus.windows.net") + ), + ) + self.store.maintenance.send(PutConnectionStringOperation(connection_string)) + + result = self.store.maintenance.send( + GetConnectionStringsOperation( + connection_string_name="asb-cs", + connection_string_type=ConnectionStringType.QUEUE, + ) + ) + parsed = result.queue_connection_strings["asb-cs"] + self.assertEqual(QueueBrokerType.AZURE_SERVICE_BUS, parsed.broker_type) + self.assertEqual( + "ns.servicebus.windows.net", + parsed.azure_service_bus_settings.passwordless.namespace, + ) From 68d196b38a087c115950ef1c4d93081fd021ba7a Mon Sep 17 00:00:00 2001 From: reforge Date: Thu, 27 Aug 2026 23:34:31 -0300 Subject: [PATCH 2/8] Add server-wide connection string operations Adds put, get and remove operations for connection strings that the server stores once and copies into every database it creates. A stored entry can name the databases it must not reach, and the get accepts an optional name and type filter, named after the database-scoped getter. Reforge-Run: 20260901T212227Z-1941916-reforge --- ravendb/__init__.py | 9 + .../operations/connection_strings.py | 232 ++++++++++++++++++ .../test_server_wide_connection_strings.py | 151 ++++++++++++ 3 files changed, 392 insertions(+) create mode 100644 ravendb/serverwide/operations/connection_strings.py create mode 100644 ravendb/tests/operations_tests/test_server_wide_connection_strings.py diff --git a/ravendb/__init__.py b/ravendb/__init__.py index 172d57b8..50443bbb 100644 --- a/ravendb/__init__.py +++ b/ravendb/__init__.py @@ -365,6 +365,15 @@ CreateDatabaseOperation, GetDatabaseRecordOperation, ) +from ravendb.serverwide.operations.connection_strings import ( + GetServerWideConnectionStringsOperation, + GetServerWideConnectionStringsResult, + PutServerWideConnectionStringOperation, + PutServerWideConnectionStringResult, + RemoveServerWideConnectionStringOperation, + RemoveServerWideConnectionStringResult, + ServerWideConnectionString, +) from ravendb.documents.identity.hilo import ( HiLoIdGenerator, diff --git a/ravendb/serverwide/operations/connection_strings.py b/ravendb/serverwide/operations/connection_strings.py new file mode 100644 index 00000000..3067d166 --- /dev/null +++ b/ravendb/serverwide/operations/connection_strings.py @@ -0,0 +1,232 @@ +from __future__ import annotations + +import json +from typing import TYPE_CHECKING, Any, Dict, List, Optional + +import requests + +from ravendb.documents.operations.connection_strings import ConnectionString +from ravendb.documents.operations.etl.configuration import RavenConnectionString +from ravendb.documents.operations.etl.elastic_search.connection import ElasticSearchConnectionString +from ravendb.documents.operations.etl.olap.connection import OlapConnectionString +from ravendb.documents.operations.etl.queue.connection import QueueConnectionString +from ravendb.documents.operations.etl.snowflake.connection import SnowflakeConnectionString +from ravendb.documents.operations.etl.sql import SqlConnectionString +from ravendb.documents.operations.ai.ai_connection_string import AiConnectionString +from ravendb.http.raven_command import RavenCommand +from ravendb.http.server_node import ServerNode +from ravendb.http.topology import RaftCommand +from ravendb.serverwide.operations.common import ServerOperation +from ravendb.serverwide.server_operation_executor import ConnectionStringType +from ravendb.tools.utils import Utils +from ravendb.util.util import RaftIdGenerator + +if TYPE_CHECKING: + from ravendb.documents.conventions import DocumentConventions + + +def _deserialize_connection_string(json_dict: Dict[str, Any], connection_string_type: ConnectionStringType): + if connection_string_type == ConnectionStringType.RAVEN: + return RavenConnectionString.from_json(json_dict) + if connection_string_type == ConnectionStringType.SQL: + return SqlConnectionString.from_json(json_dict) + if connection_string_type == ConnectionStringType.OLAP: + return OlapConnectionString.from_json(json_dict) + if connection_string_type == ConnectionStringType.ELASTIC_SEARCH: + return ElasticSearchConnectionString.from_json(json_dict) + if connection_string_type == ConnectionStringType.QUEUE: + return QueueConnectionString.from_json(json_dict) + if connection_string_type == ConnectionStringType.SNOWFLAKE: + return SnowflakeConnectionString.from_json(json_dict) + if connection_string_type == ConnectionStringType.AI: + return AiConnectionString.from_json(json_dict) + raise ValueError(f"Unknown connection string type: {connection_string_type}") + + +class ServerWideConnectionString: + def __init__(self, connection_string: ConnectionString = None, excluded_databases: Optional[List[str]] = None): + self.connection_string = connection_string + self.excluded_databases = excluded_databases + + @property + def name(self) -> Optional[str]: + return self.connection_string.name if self.connection_string else None + + @property + def type(self) -> Optional[ConnectionStringType]: + return ConnectionStringType(self.connection_string.get_type) if self.connection_string else None + + def to_json(self) -> Dict[str, Any]: + json_dict = self.connection_string.to_json() if self.connection_string else {} + json_dict["Type"] = self.type + json_dict["ExcludedDatabases"] = self.excluded_databases + return json_dict + + @classmethod + def from_json(cls, json_dict: Optional[Dict[str, Any]]) -> Optional["ServerWideConnectionString"]: + if json_dict is None: + return None + + type_raw = json_dict.get("Type") + if type_raw is None: + return None + + return cls( + connection_string=_deserialize_connection_string(json_dict, ConnectionStringType(type_raw)), + excluded_databases=json_dict.get("ExcludedDatabases"), + ) + + +class GetServerWideConnectionStringsResult: + def __init__(self, results: Optional[List[ServerWideConnectionString]] = None): + self.results = results if results is not None else [] + + @classmethod + def from_json(cls, json_dict: Dict[str, Any]) -> "GetServerWideConnectionStringsResult": + results = [] + for item in json_dict.get("Results") or []: + parsed = ServerWideConnectionString.from_json(item) + if parsed is not None: + results.append(parsed) + return cls(results) + + +class GetServerWideConnectionStringsOperation(ServerOperation[GetServerWideConnectionStringsResult]): + def __init__( + self, + connection_string_name: Optional[str] = None, + connection_string_type: Optional[ConnectionStringType] = None, + ): + if connection_string_name is not None and not connection_string_name.strip(): + raise ValueError("Connection string name must not be null or empty.") + + self._connection_string_name = connection_string_name + self._type = connection_string_type + + def get_command(self, conventions: "DocumentConventions") -> RavenCommand[GetServerWideConnectionStringsResult]: + return GetServerWideConnectionStringsOperation._GetServerWideConnectionStringsCommand( + self._connection_string_name, self._type + ) + + class _GetServerWideConnectionStringsCommand(RavenCommand[GetServerWideConnectionStringsResult]): + def __init__( + self, + connection_string_name: Optional[str] = None, + connection_string_type: Optional[ConnectionStringType] = None, + ): + super().__init__(GetServerWideConnectionStringsResult) + self._connection_string_name = connection_string_name + self._type = connection_string_type + + def is_read_request(self) -> bool: + return True + + def create_request(self, node: ServerNode) -> requests.Request: + url = f"{node.url}/admin/configuration/server-wide/connection-strings" + + query_params = [] + if self._connection_string_name is not None: + query_params.append(f"name={Utils.quote_key(self._connection_string_name)}") + if self._type is not None: + query_params.append(f"type={self._type.value}") + + if query_params: + url += f"?{'&'.join(query_params)}" + + return requests.Request("GET", url) + + def set_response(self, response: Optional[str], from_cache: bool) -> None: + if response is None: + self._throw_invalid_response() + self.result = GetServerWideConnectionStringsResult.from_json(json.loads(response)) + + +class PutServerWideConnectionStringResult: + def __init__(self, raft_command_index: Optional[int] = None): + self.raft_command_index = raft_command_index + + @classmethod + def from_json(cls, json_dict: Dict[str, Any]) -> "PutServerWideConnectionStringResult": + return cls(json_dict["RaftCommandIndex"]) + + +class PutServerWideConnectionStringOperation(ServerOperation[PutServerWideConnectionStringResult]): + def __init__(self, connection_string: ServerWideConnectionString): + if connection_string is None: + raise ValueError("connection_string cannot be None") + if connection_string.connection_string is None: + raise ValueError("ServerWideConnectionString.connection_string must not be null.") + + self._connection_string = connection_string + + def get_command(self, conventions: "DocumentConventions") -> RavenCommand[PutServerWideConnectionStringResult]: + return PutServerWideConnectionStringOperation._PutServerWideConnectionStringCommand(self._connection_string) + + class _PutServerWideConnectionStringCommand(RavenCommand[PutServerWideConnectionStringResult], RaftCommand): + def __init__(self, connection_string: ServerWideConnectionString): + super().__init__(PutServerWideConnectionStringResult) + self._connection_string = connection_string + + def is_read_request(self) -> bool: + return False + + def create_request(self, node: ServerNode) -> requests.Request: + url = f"{node.url}/admin/configuration/server-wide/connection-strings" + request = requests.Request("PUT", url) + request.data = self._connection_string.to_json() + return request + + def set_response(self, response: Optional[str], from_cache: bool) -> None: + if response is None: + self._throw_invalid_response() + self.result = PutServerWideConnectionStringResult.from_json(json.loads(response)) + + def get_raft_unique_request_id(self) -> str: + return RaftIdGenerator.new_id() + + +class RemoveServerWideConnectionStringResult: + def __init__(self, raft_command_index: Optional[int] = None): + self.raft_command_index = raft_command_index + + @classmethod + def from_json(cls, json_dict: Dict[str, Any]) -> "RemoveServerWideConnectionStringResult": + return cls(json_dict["RaftCommandIndex"]) + + +class RemoveServerWideConnectionStringOperation(ServerOperation[RemoveServerWideConnectionStringResult]): + def __init__(self, connection_string: ConnectionString): + if connection_string is None: + raise ValueError("connection_string cannot be None") + if not connection_string.name or not connection_string.name.strip(): + raise ValueError("Connection string name must not be null or empty.") + + self._connection_string = connection_string + + def get_command(self, conventions: "DocumentConventions") -> RavenCommand[RemoveServerWideConnectionStringResult]: + return RemoveServerWideConnectionStringOperation._RemoveServerWideConnectionStringCommand( + self._connection_string + ) + + class _RemoveServerWideConnectionStringCommand(RavenCommand[RemoveServerWideConnectionStringResult], RaftCommand): + def __init__(self, connection_string: ConnectionString): + super().__init__(RemoveServerWideConnectionStringResult) + self._connection_string = connection_string + + def is_read_request(self) -> bool: + return False + + def create_request(self, node: ServerNode) -> requests.Request: + url = ( + f"{node.url}/admin/configuration/server-wide/connection-strings" + f"?name={Utils.quote_key(self._connection_string.name)}&type={self._connection_string.get_type}" + ) + return requests.Request("DELETE", url) + + def set_response(self, response: Optional[str], from_cache: bool) -> None: + if response is None: + self._throw_invalid_response() + self.result = RemoveServerWideConnectionStringResult.from_json(json.loads(response)) + + def get_raft_unique_request_id(self) -> str: + return RaftIdGenerator.new_id() diff --git a/ravendb/tests/operations_tests/test_server_wide_connection_strings.py b/ravendb/tests/operations_tests/test_server_wide_connection_strings.py new file mode 100644 index 00000000..7311faba --- /dev/null +++ b/ravendb/tests/operations_tests/test_server_wide_connection_strings.py @@ -0,0 +1,151 @@ +"""Server-wide connection string tests, ported from RavenDB_24310. + +The live tests need a license that includes the ServerWideConnectionStrings +feature; under a license that gates it they skip on the server's own refusal. +""" + +import os +import unittest + +from ravendb.documents.operations.etl.configuration import RavenConnectionString +from ravendb.exceptions.raven_exceptions import RavenException +from ravendb.http.server_node import ServerNode +from ravendb.serverwide.operations.connection_strings import ( + GetServerWideConnectionStringsOperation, + PutServerWideConnectionStringOperation, + RemoveServerWideConnectionStringOperation, + ServerWideConnectionString, +) +from ravendb.serverwide.database_record import DatabaseRecord +from ravendb.serverwide.operations.common import ( + CreateDatabaseOperation, + DeleteDatabaseOperation, + GetDatabaseRecordOperation, +) +from ravendb.serverwide.server_operation_executor import ConnectionStringType +from ravendb.tests.test_base import TestBase + +CONNECTION_STRING_NAME = "MyRavenCS" + + +def _server_wide_connection_string(database="TargetDb"): + return ServerWideConnectionString( + connection_string=RavenConnectionString( + name=CONNECTION_STRING_NAME, + database=database, + topology_discovery_urls=["http://localhost:8080"], + ) + ) + + +class TestServerWideConnectionStringWireShape(unittest.TestCase): + def test_get_url_with_name_and_type(self): + operation = GetServerWideConnectionStringsOperation(CONNECTION_STRING_NAME, ConnectionStringType.RAVEN) + request = operation.get_command(None).create_request(ServerNode("http://localhost:8080", "db")) + self.assertEqual( + request.url, + "http://localhost:8080/admin/configuration/server-wide/connection-strings" + f"?name={CONNECTION_STRING_NAME}&type=Raven", + ) + + def test_get_url_without_filters(self): + operation = GetServerWideConnectionStringsOperation() + request = operation.get_command(None).create_request(ServerNode("http://localhost:8080", "db")) + self.assertEqual( + request.url, + "http://localhost:8080/admin/configuration/server-wide/connection-strings", + ) + + def test_put_payload(self): + operation = PutServerWideConnectionStringOperation(_server_wide_connection_string()) + request = operation.get_command(None).create_request(ServerNode("http://localhost:8080", "db")) + self.assertEqual("PUT", request.method) + self.assertEqual( + request.url, + "http://localhost:8080/admin/configuration/server-wide/connection-strings", + ) + self.assertEqual( + request.data, + { + "Name": CONNECTION_STRING_NAME, + "Database": "TargetDb", + "TopologyDiscoveryUrls": ["http://localhost:8080"], + "Type": ConnectionStringType.RAVEN, + "ExcludedDatabases": None, + }, + ) + + def test_delete_url(self): + operation = RemoveServerWideConnectionStringOperation(RavenConnectionString(name=CONNECTION_STRING_NAME)) + request = operation.get_command(None).create_request(ServerNode("http://localhost:8080", "db")) + self.assertEqual("DELETE", request.method) + self.assertEqual( + request.url, + "http://localhost:8080/admin/configuration/server-wide/connection-strings" + f"?name={CONNECTION_STRING_NAME}&type=Raven", + ) + + +@unittest.skipIf( + os.environ.get("RAVENDB_LICENSE") is None and os.environ.get("RAVEN_License") is None, + "Insufficient license permissions. Skipping on CI/CD.", +) +class TestServerWideConnectionStrings(TestBase): + def setUp(self): + super().setUp() + + def _run(self, operation): + try: + return self.store.maintenance.server.send(operation) + except RavenException as e: + # The ServerWideConnectionStrings license feature gates the endpoint; + # skip when the deployed license does not include it. + if "doesn't support" in str(e) and "connection string" in str(e).lower(): + self.skipTest("License does not support server-wide connection strings") + raise + + def _put(self, database="TargetDb"): + result = self._run(PutServerWideConnectionStringOperation(_server_wide_connection_string(database))) + self.assertGreater(result.raft_command_index, 0) + return result + + def _get(self): + return self._run(GetServerWideConnectionStringsOperation(CONNECTION_STRING_NAME, ConnectionStringType.RAVEN)) + + def test_can_create_and_get_server_wide_connection_string(self): + self._put() + + get_result = self._get() + self.assertEqual(1, len(get_result.results)) + self.assertEqual(CONNECTION_STRING_NAME, get_result.results[0].name) + self.assertEqual(ConnectionStringType.RAVEN, get_result.results[0].type) + self.assertEqual("TargetDb", get_result.results[0].connection_string.database) + self.assertEqual( + ["http://localhost:8080"], + get_result.results[0].connection_string.topology_discovery_urls, + ) + + def test_can_delete_server_wide_connection_string(self): + self._put() + delete_result = self._run( + RemoveServerWideConnectionStringOperation(RavenConnectionString(name=CONNECTION_STRING_NAME)) + ) + self.assertGreater(delete_result.raft_command_index, 0) + self.assertEqual(0, len(self._get().results)) + + def test_deleting_non_existent_server_wide_connection_string_is_no_op(self): + delete_result = self._run(RemoveServerWideConnectionStringOperation(RavenConnectionString(name="DoesNotExist"))) + self.assertGreater(delete_result.raft_command_index, 0) + self.assertEqual(0, len(self._get().results)) + + def test_server_wide_connection_string_propagated_to_new_database(self): + self._put() + new_db = self.store.database + "-swcs" + self.store.maintenance.server.send(CreateDatabaseOperation(DatabaseRecord(new_db))) + try: + record = self.store.maintenance.server.send(GetDatabaseRecordOperation(new_db)) + expected_name = "Server Wide Connection String, " + CONNECTION_STRING_NAME + self.assertIn(expected_name, record.raven_connection_strings) + self.assertEqual("TargetDb", record.raven_connection_strings[expected_name]["Database"]) + finally: + self.store.maintenance.server.send(DeleteDatabaseOperation(new_db, hard_delete=True)) From dfe6a36142cf5bf77210c0d6e92453c25440b5af Mon Sep 17 00:00:00 2001 From: reforge Date: Thu, 27 Aug 2026 23:34:31 -0300 Subject: [PATCH 3/8] Add GetConversationMessagesOperation for AI agents Reads the messages of an AI agent conversation, with optional timestamp paging and a detail level, and exposes it as store.ai.get_conversation_messages. The operation accepts either a conversation id or an options object, which mirrors the two reference overloads, and tests pin both forms to the same wire shape. A missing conversation returns None instead of raising, because the server answers an unknown conversation id with 404 (RavenDB-24609). The default page size is the int32 maximum the server expects when a caller asks for every message. Reforge-Run: 20260901T212227Z-1941916-reforge --- ravendb/__init__.py | 7 + ravendb/documents/ai/ai_operations.py | 22 +- .../operations/ai/agents/__init__.py | 17 ++ .../get_conversation_messages_operation.py | 249 ++++++++++++++++++ .../test_ai_agent_conversation_messages.py | 182 +++++++++++++ 5 files changed, 476 insertions(+), 1 deletion(-) create mode 100644 ravendb/documents/operations/ai/agents/get_conversation_messages_operation.py create mode 100644 ravendb/tests/ai_agent_tests/test_ai_agent_conversation_messages.py diff --git a/ravendb/__init__.py b/ravendb/__init__.py index 50443bbb..1727e3ea 100644 --- a/ravendb/__init__.py +++ b/ravendb/__init__.py @@ -115,6 +115,13 @@ GetAiAgentsResponse, AddOrUpdateAiAgentOperation, DeleteAiAgentOperation, + AiConversationDetailLevel, + AiConversationMessage, + AiConversationMessagesResult, + AiMessageRole, + AiToolCallResult, + GetConversationMessagesOperation, + GetConversationMessagesOptions, ) from ravendb.documents.operations.ai import ( ChunkingOptions, diff --git a/ravendb/documents/ai/ai_operations.py b/ravendb/documents/ai/ai_operations.py index 77bf1ad8..42786db6 100644 --- a/ravendb/documents/ai/ai_operations.py +++ b/ravendb/documents/ai/ai_operations.py @@ -1,5 +1,5 @@ from __future__ import annotations -from typing import TYPE_CHECKING, Dict, Any, Optional, Type +from typing import TYPE_CHECKING, Dict, Any, Optional, Type, Union import warnings @@ -11,7 +11,9 @@ from ravendb.documents.operations.ai.agents import ( AiAgentConfiguration, AiAgentConfigurationResult, + AiConversationMessagesResult, GetAiAgentsResponse, + GetConversationMessagesOptions, ) @@ -115,3 +117,21 @@ def conversation_with_id(self, conversation_id: str, change_vector: str = None) from ravendb.documents.ai.ai_conversation import AiConversation return AiConversation.with_conversation_id(self._store, conversation_id, change_vector) + + def get_conversation_messages( + self, conversation_id_or_options: Union[str, "GetConversationMessagesOptions"] + ) -> Optional["AiConversationMessagesResult"]: + """ + Reads messages from an AI conversation. Returns the most recent messages by default. + + Args: + conversation_id_or_options: The conversation document ID, or a + GetConversationMessagesOptions for paging and filtering control. + + Returns: + The conversation messages, or None when the conversation does not exist. + """ + from ravendb.documents.operations.ai.agents import GetConversationMessagesOperation + + operation = GetConversationMessagesOperation(conversation_id_or_options) + return self._store.maintenance.send(operation) diff --git a/ravendb/documents/operations/ai/agents/__init__.py b/ravendb/documents/operations/ai/agents/__init__.py index d8099d41..18d2271d 100644 --- a/ravendb/documents/operations/ai/agents/__init__.py +++ b/ravendb/documents/operations/ai/agents/__init__.py @@ -39,6 +39,16 @@ AiConversationParameterOptions, ) +from .get_conversation_messages_operation import ( + AiConversationDetailLevel, + AiConversationMessage, + AiConversationMessagesResult, + AiMessageRole, + AiToolCallResult, + GetConversationMessagesOperation, + GetConversationMessagesOptions, +) + __all__ = [ "AiAgentConfiguration", "AiAgentConfigurationResult", @@ -68,4 +78,11 @@ "GetAiAgentsResponse", "AddOrUpdateAiAgentOperation", "DeleteAiAgentOperation", + "AiConversationDetailLevel", + "AiConversationMessage", + "AiConversationMessagesResult", + "AiMessageRole", + "AiToolCallResult", + "GetConversationMessagesOperation", + "GetConversationMessagesOptions", ] diff --git a/ravendb/documents/operations/ai/agents/get_conversation_messages_operation.py b/ravendb/documents/operations/ai/agents/get_conversation_messages_operation.py new file mode 100644 index 00000000..870adc8f --- /dev/null +++ b/ravendb/documents/operations/ai/agents/get_conversation_messages_operation.py @@ -0,0 +1,249 @@ +from __future__ import annotations + +import enum +import json +from datetime import datetime +from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union + +import requests + +from ravendb.documents.operations.definitions import MaintenanceOperation +from ravendb.http.raven_command import RavenCommand +from ravendb.http.server_node import ServerNode +from ravendb.tools.utils import Utils + +if TYPE_CHECKING: + from ravendb.documents.conventions import DocumentConventions + from ravendb.documents.operations.ai.agents.run_conversation_operation import AiUsage + + +class AiMessageRole(enum.Enum): + SYSTEM = "System" + USER = "User" + ASSISTANT = "Assistant" + SUMMARY = "Summary" + INTERNAL = "Internal" + + +class AiConversationDetailLevel(enum.Enum): + SIMPLE = "Simple" + DETAILED = "Detailed" + FULL = "Full" + + +class AiToolCallResult: + def __init__( + self, + id: Optional[str] = None, + name: Optional[str] = None, + arguments: Optional[str] = None, + result: Optional[str] = None, + sub_conversation_id: Optional[str] = None, + ): + self.id = id + self.name = name + self.arguments = arguments + self.result = result + self.sub_conversation_id = sub_conversation_id + + def to_json(self) -> Dict[str, Any]: + return { + "Id": self.id, + "Name": self.name, + "Arguments": self.arguments, + "Result": self.result, + "SubConversationId": self.sub_conversation_id, + } + + @classmethod + def from_json(cls, json_dict: Dict[str, Any]) -> "AiToolCallResult": + return cls( + id=json_dict.get("Id"), + name=json_dict.get("Name"), + arguments=json_dict.get("Arguments"), + result=json_dict.get("Result"), + sub_conversation_id=json_dict.get("SubConversationId"), + ) + + +class AiConversationMessage: + def __init__( + self, + role: Optional[AiMessageRole] = None, + content: Optional[str] = None, + attachments: Optional[List[str]] = None, + timestamp: Optional[datetime] = None, + tool_calls: Optional[List[AiToolCallResult]] = None, + usage: Optional["AiUsage"] = None, + sub_conversation_id: Optional[str] = None, + ): + self.role = role + self.content = content + self.attachments = attachments + self.timestamp = timestamp + self.tool_calls = tool_calls + self.usage = usage + self.sub_conversation_id = sub_conversation_id + + def to_json(self) -> Dict[str, Any]: + return { + "Role": self.role.value if self.role else None, + "Content": self.content, + "Attachments": self.attachments, + "Timestamp": Utils.datetime_to_string(self.timestamp) if self.timestamp is not None else None, + "ToolCalls": ( + [tool_call.to_json() for tool_call in self.tool_calls] if self.tool_calls is not None else None + ), + "Usage": self.usage.to_json() if self.usage else None, + "SubConversationId": self.sub_conversation_id, + } + + @classmethod + def from_json(cls, json_dict: Dict[str, Any]) -> "AiConversationMessage": + from ravendb.documents.operations.ai.agents.run_conversation_operation import AiUsage + + role_str = json_dict.get("Role") + return cls( + role=AiMessageRole(role_str) if role_str else None, + content=json_dict.get("Content"), + attachments=json_dict.get("Attachments") or [], + timestamp=Utils.string_to_datetime(json_dict.get("Timestamp")), + tool_calls=( + [AiToolCallResult.from_json(tool_call) for tool_call in json_dict["ToolCalls"]] + if json_dict.get("ToolCalls") + else [] + ), + usage=AiUsage.from_json(json_dict["Usage"]) if json_dict.get("Usage") else None, + sub_conversation_id=json_dict.get("SubConversationId"), + ) + + +class AiConversationMessagesResult: + def __init__( + self, + conversation_id: Optional[str] = None, + agent: Optional[str] = None, + parameters: Optional[Dict[str, Any]] = None, + total_usage: Optional["AiUsage"] = None, + last_message_at: Optional[datetime] = None, + messages: Optional[List[AiConversationMessage]] = None, + has_more_messages: bool = False, + sub_conversation_ids: Optional[List[str]] = None, + attachments: Optional[List[str]] = None, + ): + self.conversation_id = conversation_id + self.agent = agent + self.parameters = parameters + self.total_usage = total_usage + self.last_message_at = last_message_at + self.messages = messages + self.has_more_messages = has_more_messages + self.sub_conversation_ids = sub_conversation_ids + self.attachments = attachments + + def to_json(self) -> Dict[str, Any]: + return { + "ConversationId": self.conversation_id, + "Agent": self.agent, + "Parameters": self.parameters, + "TotalUsage": self.total_usage.to_json() if self.total_usage else None, + "LastMessageAt": ( + Utils.datetime_to_string(self.last_message_at) if self.last_message_at is not None else None + ), + "HasMoreMessages": self.has_more_messages, + "SubConversationIds": self.sub_conversation_ids, + "Attachments": self.attachments, + "Messages": [message.to_json() for message in self.messages] if self.messages is not None else None, + } + + @classmethod + def from_json(cls, json_dict: Dict[str, Any]) -> "AiConversationMessagesResult": + from ravendb.documents.operations.ai.agents.run_conversation_operation import AiUsage + + return cls( + conversation_id=json_dict.get("ConversationId"), + agent=json_dict.get("Agent"), + parameters=json_dict.get("Parameters"), + total_usage=AiUsage.from_json(json_dict["TotalUsage"]) if json_dict.get("TotalUsage") else None, + last_message_at=Utils.string_to_datetime(json_dict.get("LastMessageAt")), + messages=( + [AiConversationMessage.from_json(message) for message in json_dict["Messages"]] + if json_dict.get("Messages") + else None + ), + has_more_messages=json_dict.get("HasMoreMessages", False), + sub_conversation_ids=json_dict.get("SubConversationIds"), + attachments=json_dict.get("Attachments"), + ) + + +class GetConversationMessagesOptions: + def __init__( + self, + conversation_id: Optional[str] = None, + before: Optional[datetime] = None, + after: Optional[datetime] = None, + page_size: int = 2147483647, + detail_level: AiConversationDetailLevel = AiConversationDetailLevel.SIMPLE, + ): + self.conversation_id = conversation_id + self.before = before + self.after = after + self.page_size = page_size + self.detail_level = detail_level + + def validate(self) -> None: + if not self.conversation_id: + raise ValueError("conversation_id cannot be None or empty") + + if self.before is not None and self.after is not None: + raise ValueError("before and after cannot both be specified.") + + if self.page_size <= 0: + raise ValueError("page_size must be greater than 0.") + + +class GetConversationMessagesOperation(MaintenanceOperation[AiConversationMessagesResult]): + def __init__(self, conversation_id_or_options: Union[str, GetConversationMessagesOptions]): + if isinstance(conversation_id_or_options, GetConversationMessagesOptions): + parameters = conversation_id_or_options + elif isinstance(conversation_id_or_options, str): + parameters = GetConversationMessagesOptions(conversation_id=conversation_id_or_options) + else: + raise TypeError("conversation_id_or_options must be a conversation id or GetConversationMessagesOptions") + + parameters.validate() + self._parameters = parameters + + def get_command(self, conventions: "DocumentConventions") -> RavenCommand[AiConversationMessagesResult]: + return GetConversationMessagesOperation._GetConversationMessagesCommand(self._parameters) + + class _GetConversationMessagesCommand(RavenCommand[AiConversationMessagesResult]): + def __init__(self, parameters: GetConversationMessagesOptions): + super().__init__(AiConversationMessagesResult) + self._parameters = parameters + + def is_read_request(self) -> bool: + return True + + def create_request(self, node: ServerNode) -> requests.Request: + url = ( + f"{node.url}/databases/{node.database}/ai/agent/conversation/messages" + f"?conversationId={Utils.quote_key(self._parameters.conversation_id, True)}" + ) + + if self._parameters.before is not None: + url += f"&before={Utils.quote_key(Utils.datetime_to_string(self._parameters.before))}" + if self._parameters.after is not None: + url += f"&after={Utils.quote_key(Utils.datetime_to_string(self._parameters.after))}" + + url += f"&pageSize={self._parameters.page_size}" + url += f"&detailLevel={self._parameters.detail_level.value}" + + return requests.Request("GET", url) + + def set_response(self, response: Optional[str], from_cache: bool) -> None: + # A 404 for an unknown conversation yields a null result, not an exception. + if response is None: + return + self.result = AiConversationMessagesResult.from_json(json.loads(response)) diff --git a/ravendb/tests/ai_agent_tests/test_ai_agent_conversation_messages.py b/ravendb/tests/ai_agent_tests/test_ai_agent_conversation_messages.py new file mode 100644 index 00000000..2c9b292f --- /dev/null +++ b/ravendb/tests/ai_agent_tests/test_ai_agent_conversation_messages.py @@ -0,0 +1,182 @@ +"""AI conversation messages tests, ported from AiAgentGetConversationMessages.""" + +import os +import unittest +from datetime import datetime + +from ravendb.documents.operations.ai.agents import ( + AiConversationDetailLevel, + AiConversationMessage, + AiConversationMessagesResult, + AiMessageRole, + AiToolCallResult, + GetConversationMessagesOperation, + GetConversationMessagesOptions, +) +from ravendb.http.server_node import ServerNode +from ravendb.tests.test_base import TestBase + + +class TestGetConversationMessagesWireShape(unittest.TestCase): + def _url(self, options: GetConversationMessagesOptions) -> str: + operation = GetConversationMessagesOperation(options) + command = operation.get_command(None) + request = command.create_request(ServerNode("http://localhost:8080", "db")) + return request.url + + def test_url_has_all_parameters(self): + url = self._url(GetConversationMessagesOptions(conversation_id="conversations/1")) + self.assertEqual( + url, + "http://localhost:8080/databases/db/ai/agent/conversation/messages" + "?conversationId=conversations/1&pageSize=2147483647&detailLevel=Simple", + ) + + def test_operation_accepts_plain_conversation_id(self): + # The string form must build the same URL as the options form with defaults. + request = ( + GetConversationMessagesOperation("conversations/1") + .get_command(None) + .create_request(ServerNode("http://localhost:8080", "db")) + ) + self.assertEqual( + request.url, + "http://localhost:8080/databases/db/ai/agent/conversation/messages" + "?conversationId=conversations/1&pageSize=2147483647&detailLevel=Simple", + ) + + def test_url_includes_before_and_detail_level(self): + options = GetConversationMessagesOptions( + conversation_id="conversations/1", + before=datetime(2026, 6, 1, 12, 0, 0), + page_size=50, + detail_level=AiConversationDetailLevel.DETAILED, + ) + url = self._url(options) + self.assertIn("&before=2026-06-01T12%3A00%3A00.0000000", url) + self.assertIn("&pageSize=50", url) + self.assertIn("&detailLevel=Detailed", url) + + def test_validation(self): + with self.assertRaises(ValueError): + GetConversationMessagesOperation(GetConversationMessagesOptions()) + with self.assertRaises(ValueError): + GetConversationMessagesOperation( + GetConversationMessagesOptions( + conversation_id="conversations/1", + before=datetime(2026, 1, 1), + after=datetime(2026, 1, 2), + ) + ) + with self.assertRaises(ValueError): + GetConversationMessagesOperation(GetConversationMessagesOptions(conversation_id="c", page_size=0)) + + +class TestGetConversationMessagesResponseParse(unittest.TestCase): + """Response parsing, pinned to the reference AiConversationMessagesResult wire.""" + + def _sample_payload(self): + return { + "ConversationId": "conversations/1", + "Agent": "agents/1", + "Parameters": {"temperature": 0.5}, + "TotalUsage": { + "PromptTokens": 10, + "CompletionTokens": 20, + "TotalTokens": 30, + "CachedTokens": 0, + "ReasoningTokens": 5, + }, + "LastMessageAt": "2026-06-01T12:00:00.0000000", + "Messages": [ + { + "Role": "User", + "Content": "hello", + "Attachments": ["a.txt"], + "Timestamp": "2026-06-01T11:59:00.0000000", + "ToolCalls": None, + "Usage": None, + "SubConversationId": None, + }, + { + "Role": "Assistant", + "Content": "hi", + "Attachments": [], + "Timestamp": "2026-06-01T12:00:00.0000000", + "ToolCalls": [ + { + "Id": "call-1", + "Name": "get_weather", + "Arguments": "{}", + "Result": "sunny", + "SubConversationId": None, + } + ], + "Usage": { + "PromptTokens": 1, + "CompletionTokens": 2, + "TotalTokens": 3, + "CachedTokens": 0, + "ReasoningTokens": 0, + }, + "SubConversationId": None, + }, + ], + "HasMoreMessages": True, + "SubConversationIds": ["conversations/1/sub/1"], + "Attachments": ["b.txt"], + } + + def test_result_from_json_binds_all_fields(self): + result = AiConversationMessagesResult.from_json(self._sample_payload()) + self.assertEqual("conversations/1", result.conversation_id) + self.assertEqual("agents/1", result.agent) + self.assertTrue(result.has_more_messages) + self.assertEqual(["conversations/1/sub/1"], result.sub_conversation_ids) + self.assertEqual(["b.txt"], result.attachments) + self.assertEqual(2, len(result.messages)) + self.assertEqual(datetime(2026, 6, 1, 12, 0, 0), result.last_message_at) + self.assertEqual(0.5, result.parameters["temperature"]) + + def test_message_from_json_binds_role_content_and_tool_calls(self): + result = AiConversationMessagesResult.from_json(self._sample_payload()) + messages = result.messages + first = messages[0] + self.assertEqual(AiMessageRole.USER, first.role) + self.assertEqual("hello", first.content) + self.assertEqual(["a.txt"], first.attachments) + self.assertEqual([], first.tool_calls) + + second = messages[1] + self.assertEqual(AiMessageRole.ASSISTANT, second.role) + self.assertEqual(1, len(second.tool_calls)) + tool_call = second.tool_calls[0] + self.assertIsInstance(tool_call, AiToolCallResult) + self.assertEqual("call-1", tool_call.id) + self.assertEqual("get_weather", tool_call.name) + self.assertEqual("sunny", tool_call.result) + self.assertIsNotNone(second.usage) + self.assertEqual(2, second.usage.completion_tokens) + self.assertIsNotNone(result.total_usage) + self.assertEqual(5, result.total_usage.reasoning_tokens) + + def test_message_timestamp_parsed(self): + messages = AiConversationMessagesResult.from_json(self._sample_payload()).messages + self.assertEqual(datetime(2026, 6, 1, 11, 59, 0), messages[0].timestamp) + + +@unittest.skipIf( + os.environ.get("RAVENDB_LICENSE") is None and os.environ.get("RAVEN_License") is None, + "Insufficient license permissions. Skipping on CI/CD.", +) +class TestGetConversationMessages(TestBase): + def test_missing_conversation_returns_none(self): + # A 404 for an unknown conversation must yield None, not an exception. + result = self.store.ai.get_conversation_messages("conversations/does-not-exist") + self.assertIsNone(result) + + def test_missing_conversation_returns_none_with_options(self): + result = self.store.ai.get_conversation_messages( + GetConversationMessagesOptions(conversation_id="conversations/does-not-exist") + ) + self.assertIsNone(result) From c4bb1ee611c7e42292b0a4fdec8e57bf7d2043b4 Mon Sep 17 00:00:00 2001 From: reforge Date: Thu, 27 Aug 2026 23:34:31 -0300 Subject: [PATCH 4/8] Add SSO fields to client certificate editing and metadata EditClientCertificateOperation.Parameters accepts the SSO server pinning hashes, the allow-any-SSO-server flag and the SSO identifiers. Each of the three is written to the request body only when the caller supplies it, so a partial edit leaves the stored SSO configuration alone; an empty list is a value and clears the stored one. An SSO identifier omits its domain when it has none. The certificate read models bind the new Usage and SSO members the server writes. Usage is a numeric enum whose member name is the wire value. Reforge-Run: 20260901T212227Z-1941916-reforge --- ravendb/__init__.py | 3 + ravendb/serverwide/operations/certificates.py | 119 +++++++++++++++- .../operations_tests/test_sso_certificates.py | 127 ++++++++++++++++++ 3 files changed, 246 insertions(+), 3 deletions(-) create mode 100644 ravendb/tests/operations_tests/test_sso_certificates.py diff --git a/ravendb/__init__.py b/ravendb/__init__.py index 1727e3ea..ce62446b 100644 --- a/ravendb/__init__.py +++ b/ravendb/__init__.py @@ -363,6 +363,9 @@ GetCertificatesResponse, PutClientCertificateOperation, SecurityClearance, + SsoProvider, + SsoIdentifier, + CertificateUsage, ) from ravendb.serverwide.operations.common import ( BuildNumber, diff --git a/ravendb/serverwide/operations/certificates.py b/ravendb/serverwide/operations/certificates.py index 29a22cba..1655748c 100644 --- a/ravendb/serverwide/operations/certificates.py +++ b/ravendb/serverwide/operations/certificates.py @@ -4,7 +4,7 @@ import enum import json from datetime import datetime -from typing import TYPE_CHECKING, Dict, List +from typing import TYPE_CHECKING, Dict, List, Optional import requests @@ -39,6 +39,47 @@ def __str__(self): return self.value +class SsoProvider(enum.Enum): + GITHUB = "Github" + GOOGLE = "Google" + MICROSOFT = "Microsoft" + WINDOWS = "Windows" + + def __str__(self): + return self.value + + +class CertificateUsage(enum.Enum): + RAVEN_SERVER = 0 + RAVEN_SERVER_FOR_COMMUNICATION = 1 + CLIENT = 2 + SSO_SERVER = 3 + SSO_CLIENT = 4 + WELL_KNOWN_ISSUER = 5 + + +class SsoIdentifier: + def __init__(self, provider: SsoProvider = None, domain: str = None, identifier: str = None): + self.provider = provider + self.domain = domain + self.identifier = identifier + + def to_json(self) -> dict: + json_dict = {"Provider": self.provider.value if self.provider else None, "Identifier": self.identifier} + if self.domain: + json_dict["Domain"] = self.domain + return json_dict + + @classmethod + def from_json(cls, json_dict: dict) -> "SsoIdentifier": + provider_raw = json_dict.get("Provider") + return cls( + provider=SsoProvider(provider_raw) if provider_raw else None, + domain=json_dict.get("Domain"), + identifier=json_dict.get("Identifier"), + ) + + class CertificateRawData: def __init__(self, raw_data: bytes = None): self.raw_data = raw_data @@ -57,6 +98,10 @@ def __init__( public_key_pinning_hash: str = None, not_before: datetime = None, disabled: bool = False, + usage: CertificateUsage = None, + sso_server_public_key_pinning_hashes: List[str] = None, + allow_any_sso_server: bool = None, + sso_identifiers: List[SsoIdentifier] = None, ): self.name = name self.security_clearance = security_clearance @@ -68,6 +113,23 @@ def __init__( self.public_key_pinning_hash = public_key_pinning_hash self.not_before = not_before self.disabled = disabled + self.usage = usage + self.sso_server_public_key_pinning_hashes = sso_server_public_key_pinning_hashes + self.allow_any_sso_server = allow_any_sso_server + self.sso_identifiers = sso_identifiers + + @staticmethod + def _parse_usage(usage_raw) -> Optional[CertificateUsage]: + # The server writes the member name of a numeric enum, so the name is the wire value. + if usage_raw is None: + return None + return CertificateUsage[Utils.convert_to_snake_case(usage_raw).upper()] + + @staticmethod + def _parse_sso_identifiers(identifiers_raw) -> Optional[List[SsoIdentifier]]: + if identifiers_raw is None: + return None + return [SsoIdentifier.from_json(identifier) for identifier in identifiers_raw] @classmethod def from_json(cls, json_dict: dict) -> CertificateMetadata: @@ -82,6 +144,10 @@ def from_json(cls, json_dict: dict) -> CertificateMetadata: json_dict.get("PublicKeyPinningHash", None), Utils.string_to_datetime(json_dict["NotBefore"]) if "NotBefore" in json_dict else None, json_dict.get("Disabled", False), + cls._parse_usage(json_dict.get("Usage")), + json_dict.get("SsoServerPublicKeyPinningHashes", None), + json_dict.get("AllowAnySsoServer", None), + cls._parse_sso_identifiers(json_dict.get("SsoIdentifiers")), ) @@ -99,6 +165,10 @@ def __init__( collection_primary_key: str = None, public_key_pinning_hash: str = None, disabled: bool = False, + usage: CertificateUsage = None, + sso_server_public_key_pinning_hashes: List[str] = None, + allow_any_sso_server: bool = None, + sso_identifiers: List[SsoIdentifier] = None, ): super().__init__( name, @@ -110,6 +180,10 @@ def __init__( collection_primary_key, public_key_pinning_hash, disabled=disabled, + usage=usage, + sso_server_public_key_pinning_hashes=sso_server_public_key_pinning_hashes, + allow_any_sso_server=allow_any_sso_server, + sso_identifiers=sso_identifiers, ) self.certificate = certificate self.password = password @@ -145,6 +219,10 @@ def from_json(cls, json_dict: dict) -> CertificateDefinition: json_dict["CollectionPrimaryKey"], json_dict["PublicKeyPinningHash"], disabled=json_dict.get("Disabled", False), + usage=cls._parse_usage(json_dict.get("Usage")), + sso_server_public_key_pinning_hashes=json_dict.get("SsoServerPublicKeyPinningHashes"), + allow_any_sso_server=json_dict.get("AllowAnySsoServer"), + sso_identifiers=cls._parse_sso_identifiers(json_dict.get("SsoIdentifiers")), ) @@ -456,12 +534,18 @@ def __init__( name: str, clearance: SecurityClearance, disabled: bool = False, + sso_server_public_key_pinning_hashes: List[str] = None, + allow_any_sso_server: bool = None, + sso_identifiers: List[SsoIdentifier] = None, ): self.thumbprint = thumbprint self.permissions = permissions self.name = name self.clearance = clearance self.disabled = disabled + self.sso_server_public_key_pinning_hashes = sso_server_public_key_pinning_hashes + self.allow_any_sso_server = allow_any_sso_server + self.sso_identifiers = sso_identifiers def __init__(self, parameters: Parameters): if parameters is None: @@ -481,10 +565,20 @@ def __init__(self, parameters: Parameters): self.__permissions = parameters.permissions self.__clearance = parameters.clearance self.__disabled = parameters.disabled + self.__sso_server_public_key_pinning_hashes = parameters.sso_server_public_key_pinning_hashes + self.__allow_any_sso_server = parameters.allow_any_sso_server + self.__sso_identifiers = parameters.sso_identifiers def get_command(self, conventions: "DocumentConventions") -> "VoidRavenCommand": return self.__EditCertificateClientCommand( - self.__thumbprint, self.__name, self.__permissions, self.__clearance, self.__disabled + self.__thumbprint, + self.__name, + self.__permissions, + self.__clearance, + self.__disabled, + self.__sso_server_public_key_pinning_hashes, + self.__allow_any_sso_server, + self.__sso_identifiers, ) class __EditCertificateClientCommand(VoidRavenCommand, RaftCommand): @@ -495,6 +589,9 @@ def __init__( permissions: Dict[str, DatabaseAccess], clearance: SecurityClearance, disabled: bool, + sso_server_public_key_pinning_hashes: List[str], + allow_any_sso_server: bool, + sso_identifiers: List[SsoIdentifier], ): super().__init__() self.__thumbprint = thumbprint @@ -502,6 +599,9 @@ def __init__( self.__permissions = permissions self.__clearance = clearance self.__disabled = disabled + self.__sso_server_public_key_pinning_hashes = sso_server_public_key_pinning_hashes + self.__allow_any_sso_server = allow_any_sso_server + self.__sso_identifiers = sso_identifiers def is_read_request(self) -> bool: return False @@ -517,7 +617,20 @@ def create_request(self, node: ServerNode) -> requests.Request: definition.disabled = self.__disabled request = requests.Request("POST", url) - request.data = definition.to_json() + body = definition.to_json() + + # SSO fields are written only when explicitly provided so the server + # leaves the existing SSO configuration untouched on a partial edit. + if self.__sso_server_public_key_pinning_hashes is not None: + body["SsoServerPublicKeyPinningHashes"] = self.__sso_server_public_key_pinning_hashes + + if self.__allow_any_sso_server is not None: + body["AllowAnySsoServer"] = self.__allow_any_sso_server + + if self.__sso_identifiers is not None: + body["SsoIdentifiers"] = [identifier.to_json() for identifier in self.__sso_identifiers] + + request.data = body return request diff --git a/ravendb/tests/operations_tests/test_sso_certificates.py b/ravendb/tests/operations_tests/test_sso_certificates.py new file mode 100644 index 00000000..a3678a22 --- /dev/null +++ b/ravendb/tests/operations_tests/test_sso_certificates.py @@ -0,0 +1,127 @@ +"""SSO certificate editing and certificate metadata tests (RavenDB-26155).""" + +import unittest + +from ravendb.http.server_node import ServerNode +from ravendb.serverwide.operations.certificates import ( + CertificateMetadata, + CertificateUsage, + DatabaseAccess, + EditClientCertificateOperation, + SecurityClearance, + SsoIdentifier, + SsoProvider, +) + + +def _edit_body(parameters: EditClientCertificateOperation.Parameters) -> dict: + operation = EditClientCertificateOperation(parameters) + command = operation.get_command(None) + request = command.create_request(ServerNode("http://localhost:8080", "db")) + return request.data + + +class TestSsoCertificateEditing(unittest.TestCase): + def _parameters(self, **kwargs): + return EditClientCertificateOperation.Parameters( + thumbprint="A" * 64, + permissions={"db": DatabaseAccess.READ_WRITE}, + name="cert-name", + clearance=SecurityClearance.VALID_USER, + **kwargs, + ) + + def test_no_sso_fields_when_not_provided(self): + body = _edit_body(self._parameters()) + self.assertNotIn("SsoServerPublicKeyPinningHashes", body) + self.assertNotIn("AllowAnySsoServer", body) + self.assertNotIn("SsoIdentifiers", body) + + def test_sso_fields_written_when_provided(self): + body = _edit_body( + self._parameters( + sso_server_public_key_pinning_hashes=["hash1"], + allow_any_sso_server=True, + sso_identifiers=[ + SsoIdentifier( + provider=SsoProvider.GITHUB, + domain="example.com", + identifier="user", + ) + ], + ) + ) + self.assertEqual(["hash1"], body["SsoServerPublicKeyPinningHashes"]) + self.assertTrue(body["AllowAnySsoServer"]) + self.assertEqual( + [{"Provider": "Github", "Domain": "example.com", "Identifier": "user"}], + body["SsoIdentifiers"], + ) + + def test_empty_list_clears_and_domain_omitted_when_empty(self): + body = _edit_body( + self._parameters( + sso_server_public_key_pinning_hashes=[], + sso_identifiers=[SsoIdentifier(provider=SsoProvider.GOOGLE, identifier="user")], + ) + ) + self.assertEqual([], body["SsoServerPublicKeyPinningHashes"]) + self.assertEqual([{"Provider": "Google", "Identifier": "user"}], body["SsoIdentifiers"]) + + def test_usage_enum_values(self): + self.assertEqual(0, CertificateUsage.RAVEN_SERVER.value) + self.assertEqual(2, CertificateUsage.CLIENT.value) + self.assertEqual(3, CertificateUsage.SSO_SERVER.value) + self.assertEqual(4, CertificateUsage.SSO_CLIENT.value) + self.assertEqual(5, CertificateUsage.WELL_KNOWN_ISSUER.value) + + +class TestCertificateMetadataSsoFields(unittest.TestCase): + """GetCertificateMetadataOperation reads the SSO fields the server writes + on CertificateMetadata (reference CertificateMetadata.ToJson).""" + + _METADATA = { + "Name": "cert-name", + "SecurityClearance": "ValidUser", + "Thumbprint": "A" * 64, + "Permissions": {"db": "ReadWrite"}, + "Disabled": False, + "Usage": "SsoClient", + "SsoServerPublicKeyPinningHashes": ["hash1", "hash2"], + "AllowAnySsoServer": True, + "SsoIdentifiers": [ + {"Provider": "Github", "Domain": "example.com", "Identifier": "user"}, + {"Provider": "Windows", "Identifier": "win-user"}, + ], + } + + def test_from_json_binds_usage_and_sso_fields(self): + metadata = CertificateMetadata.from_json(self._METADATA) + self.assertEqual(CertificateUsage.SSO_CLIENT, metadata.usage) + self.assertEqual(["hash1", "hash2"], metadata.sso_server_public_key_pinning_hashes) + self.assertTrue(metadata.allow_any_sso_server) + self.assertEqual(2, len(metadata.sso_identifiers)) + first = metadata.sso_identifiers[0] + self.assertEqual(SsoProvider.GITHUB, first.provider) + self.assertEqual("example.com", first.domain) + self.assertEqual("user", first.identifier) + self.assertEqual(SsoProvider.WINDOWS, metadata.sso_identifiers[1].provider) + + def test_absent_usage_and_sso_fields_are_none(self): + metadata = CertificateMetadata.from_json( + { + "Name": "cert-name", + "SecurityClearance": "ValidUser", + "Thumbprint": "A" * 64, + } + ) + self.assertIsNone(metadata.usage) + self.assertIsNone(metadata.sso_server_public_key_pinning_hashes) + self.assertIsNone(metadata.allow_any_sso_server) + self.assertIsNone(metadata.sso_identifiers) + + def test_sso_provider_enum_values(self): + self.assertEqual("Github", SsoProvider.GITHUB.value) + self.assertEqual("Google", SsoProvider.GOOGLE.value) + self.assertEqual("Microsoft", SsoProvider.MICROSOFT.value) + self.assertEqual("Windows", SsoProvider.WINDOWS.value) From d60022c2db218925f489b48931bf42e69b7e3b64 Mon Sep 17 00:00:00 2001 From: reforge Date: Thu, 27 Aug 2026 23:34:31 -0300 Subject: [PATCH 5/8] Add disable_checksum_validation to S3 settings The S3 settings of remote attachments and of backups accept disable_checksum_validation, which turns off the checksum the S3 client sends. An unset value serializes as null, like the sibling flags, and the server reads null as false. The backup S3 settings deserializer tolerates missing optional keys, matching the sibling remote attachment settings. Reforge-Run: 20260901T212227Z-1941916-reforge --- .../operations/attachments/__init__.py | 4 ++ .../documents/operations/backups/settings.py | 29 ++++++---- .../test_s3_checksum_and_pull_replication.py | 56 +++++++++++++++++++ 3 files changed, 78 insertions(+), 11 deletions(-) create mode 100644 ravendb/tests/operations_tests/test_s3_checksum_and_pull_replication.py diff --git a/ravendb/documents/operations/attachments/__init__.py b/ravendb/documents/operations/attachments/__init__.py index 1dfdcd40..9fedbffe 100644 --- a/ravendb/documents/operations/attachments/__init__.py +++ b/ravendb/documents/operations/attachments/__init__.py @@ -404,6 +404,7 @@ def __init__( custom_server_url: str = None, force_path_style: bool = None, storage_class: Optional[S3StorageClass] = None, + disable_checksum_validation: bool = None, ): self.aws_access_key = aws_access_key self.aws_secret_key = aws_secret_key @@ -414,6 +415,7 @@ def __init__( self.custom_server_url = custom_server_url self.force_path_style = force_path_style self.storage_class = storage_class + self.disable_checksum_validation = disable_checksum_validation @classmethod def from_json(cls, json_dict: dict) -> RemoteAttachmentsS3Settings: @@ -428,6 +430,7 @@ def from_json(cls, json_dict: dict) -> RemoteAttachmentsS3Settings: json_dict.get("CustomServerUrl"), json_dict.get("ForcePathStyle"), S3StorageClass(storage_class_raw) if storage_class_raw is not None else None, + json_dict.get("DisableChecksumValidation"), ) def to_json(self) -> dict: @@ -440,6 +443,7 @@ def to_json(self) -> dict: "BucketName": self.bucket_name, "CustomServerUrl": self.custom_server_url, "ForcePathStyle": self.force_path_style, + "DisableChecksumValidation": self.disable_checksum_validation, } if self.storage_class is not None: result["StorageClass"] = self.storage_class.value diff --git a/ravendb/documents/operations/backups/settings.py b/ravendb/documents/operations/backups/settings.py index 565e3df8..4e9887cc 100644 --- a/ravendb/documents/operations/backups/settings.py +++ b/ravendb/documents/operations/backups/settings.py @@ -130,6 +130,7 @@ def __init__( bucket_name: str = None, custom_server_url: str = None, force_path_style: bool = None, + disable_checksum_validation: bool = None, ): super().__init__( disabled, @@ -143,26 +144,31 @@ def __init__( self.bucket_name = bucket_name self.custom_server_url = custom_server_url self.force_path_style = force_path_style + self.disable_checksum_validation = disable_checksum_validation @classmethod def from_json(cls, json_dict: Dict[str, Any]) -> S3Settings: + script_dict = json_dict.get("GetBackupConfigurationScript") return cls( - json_dict["Disabled"], - GetBackupConfigurationScript.from_json(json_dict["GetBackupConfigurationScript"]), - json_dict["AwsAccessKey"], - json_dict["AwsSecretKey"], - json_dict["AwsSessionToken"], - json_dict["AwsRegionName"], - json_dict["RemoteFolderName"], - json_dict["BucketName"], - json_dict["CustomServerUrl"], - json_dict["ForcePathStyle"], + json_dict.get("Disabled"), + GetBackupConfigurationScript.from_json(script_dict) if script_dict else None, + json_dict.get("AwsAccessKey"), + json_dict.get("AwsSecretKey"), + json_dict.get("AwsSessionToken"), + json_dict.get("AwsRegionName"), + json_dict.get("RemoteFolderName"), + json_dict.get("BucketName"), + json_dict.get("CustomServerUrl"), + json_dict.get("ForcePathStyle"), + json_dict.get("DisableChecksumValidation"), ) def to_json(self) -> Dict[str, Any]: return { "Disabled": self.disabled, - "GetBackupConfigurationScript": self.get_backup_configuration_script.to_json(), + "GetBackupConfigurationScript": ( + self.get_backup_configuration_script.to_json() if self.get_backup_configuration_script else None + ), "AwsAccessKey": self.aws_access_key, "AwsSecretKey": self.aws_secret_key, "AwsSessionToken": self.aws_session_token, @@ -171,6 +177,7 @@ def to_json(self) -> Dict[str, Any]: "BucketName": self.bucket_name, "CustomServerUrl": self.custom_server_url, "ForcePathStyle": self.force_path_style, + "DisableChecksumValidation": self.disable_checksum_validation, } diff --git a/ravendb/tests/operations_tests/test_s3_checksum_and_pull_replication.py b/ravendb/tests/operations_tests/test_s3_checksum_and_pull_replication.py new file mode 100644 index 00000000..527f84d7 --- /dev/null +++ b/ravendb/tests/operations_tests/test_s3_checksum_and_pull_replication.py @@ -0,0 +1,56 @@ +"""S3 checksum toggle wire tests.""" + +import unittest + +from ravendb.documents.operations.attachments import RemoteAttachmentsS3Settings +from ravendb.documents.operations.backups.settings import S3Settings + + +class TestS3ChecksumToggleWireShape(unittest.TestCase): + def test_remote_attachments_s3_settings_to_json_writes_checksum_flag(self): + settings = RemoteAttachmentsS3Settings( + aws_access_key="ak", + aws_secret_key="sk", + bucket_name="bucket", + disable_checksum_validation=True, + ) + payload = settings.to_json() + self.assertTrue(payload["DisableChecksumValidation"]) + self.assertIn("DisableChecksumValidation", payload) + + def test_remote_attachments_s3_settings_from_json_round_trip(self): + parsed = RemoteAttachmentsS3Settings.from_json( + { + "AwsAccessKey": "ak", + "AwsSecretKey": "sk", + "BucketName": "bucket", + "DisableChecksumValidation": True, + } + ) + self.assertTrue(parsed.disable_checksum_validation) + self.assertEqual("bucket", parsed.bucket_name) + + def test_remote_attachments_s3_settings_default_is_false(self): + settings = RemoteAttachmentsS3Settings(bucket_name="bucket") + self.assertFalse(settings.disable_checksum_validation) + self.assertFalse(settings.to_json()["DisableChecksumValidation"]) + + def test_backup_s3_settings_to_json_writes_checksum_flag(self): + settings = S3Settings(bucket_name="bucket", disable_checksum_validation=True) + payload = settings.to_json() + self.assertTrue(payload["DisableChecksumValidation"]) + + def test_backup_s3_settings_from_json_round_trip(self): + parsed = S3Settings.from_json( + { + "Disabled": False, + "GetBackupConfigurationScript": { + "Exec": None, + "Arguments": None, + "TimeoutInMs": 10000, + }, + "BucketName": "bucket", + "DisableChecksumValidation": True, + } + ) + self.assertTrue(parsed.disable_checksum_validation) From f4528815a1edd4957134e7a0e999dac6a98e6f0f Mon Sep 17 00:00:00 2001 From: reforge Date: Thu, 27 Aug 2026 23:34:31 -0300 Subject: [PATCH 6/8] Add hub and sink cursors to pull replication sink task info The pull replication sink task info carries the hub and sink cursors the server now reports. Reforge-Run: 20260901T212227Z-1941916-reforge --- ravendb/documents/operations/ongoing_tasks.py | 8 ++++ .../test_s3_checksum_and_pull_replication.py | 37 ++++++++++++++++++- 2 files changed, 44 insertions(+), 1 deletion(-) diff --git a/ravendb/documents/operations/ongoing_tasks.py b/ravendb/documents/operations/ongoing_tasks.py index 62b869c6..027ead01 100644 --- a/ravendb/documents/operations/ongoing_tasks.py +++ b/ravendb/documents/operations/ongoing_tasks.py @@ -392,6 +392,8 @@ def __init__( access_name: Optional[str] = None, allowed_hub_to_sink_paths: Optional[list] = None, allowed_sink_to_hub_paths: Optional[list] = None, + hub_cursor: Optional[str] = None, + sink_cursor: Optional[str] = None, ): super().__init__( task_id=task_id, @@ -414,6 +416,8 @@ def __init__( self.access_name = access_name self.allowed_hub_to_sink_paths = allowed_hub_to_sink_paths self.allowed_sink_to_hub_paths = allowed_sink_to_hub_paths + self.hub_cursor = hub_cursor + self.sink_cursor = sink_cursor def to_json(self) -> dict: result = super().to_json() @@ -427,6 +431,8 @@ def to_json(self) -> dict: result["AccessName"] = self.access_name result["AllowedHubToSinkPaths"] = self.allowed_hub_to_sink_paths result["AllowedSinkToHubPaths"] = self.allowed_sink_to_hub_paths + result["HubCursor"] = self.hub_cursor + result["SinkCursor"] = self.sink_cursor return result @classmethod @@ -457,6 +463,8 @@ def from_json(cls, json_dict: dict) -> Optional["OngoingTaskPullReplicationAsSin access_name=json_dict.get("AccessName"), allowed_hub_to_sink_paths=json_dict.get("AllowedHubToSinkPaths"), allowed_sink_to_hub_paths=json_dict.get("AllowedSinkToHubPaths"), + hub_cursor=json_dict.get("HubCursor"), + sink_cursor=json_dict.get("SinkCursor"), ) diff --git a/ravendb/tests/operations_tests/test_s3_checksum_and_pull_replication.py b/ravendb/tests/operations_tests/test_s3_checksum_and_pull_replication.py index 527f84d7..a7f67abe 100644 --- a/ravendb/tests/operations_tests/test_s3_checksum_and_pull_replication.py +++ b/ravendb/tests/operations_tests/test_s3_checksum_and_pull_replication.py @@ -1,9 +1,13 @@ -"""S3 checksum toggle wire tests.""" +"""S3 checksum toggle and pull-replication cursor wire tests.""" import unittest from ravendb.documents.operations.attachments import RemoteAttachmentsS3Settings from ravendb.documents.operations.backups.settings import S3Settings +from ravendb.documents.operations.ongoing_tasks import ( + OngoingTaskPullReplicationAsSink, + OngoingTaskType, +) class TestS3ChecksumToggleWireShape(unittest.TestCase): @@ -54,3 +58,34 @@ def test_backup_s3_settings_from_json_round_trip(self): } ) self.assertTrue(parsed.disable_checksum_validation) + + +class TestPullReplicationCursorFields(unittest.TestCase): + def test_sink_task_info_from_json_reads_cursors(self): + task = OngoingTaskPullReplicationAsSink.from_json( + { + "TaskId": 1, + "TaskName": "sink", + "TaskType": "PullReplicationAsSink", + "HubName": "hub", + "HubCursor": "hub-cursor-value", + "SinkCursor": "sink-cursor-value", + } + ) + self.assertIsNotNone(task) + self.assertEqual("hub-cursor-value", task.hub_cursor) + self.assertEqual("sink-cursor-value", task.sink_cursor) + self.assertEqual(OngoingTaskType.PULL_REPLICATION_AS_SINK, task.task_type) + + def test_sink_task_info_to_json_writes_cursors(self): + task = OngoingTaskPullReplicationAsSink(task_id=1, hub_name="hub") + task.hub_cursor = "hub-cursor-value" + task.sink_cursor = "sink-cursor-value" + payload = task.to_json() + self.assertEqual("hub-cursor-value", payload["HubCursor"]) + self.assertEqual("sink-cursor-value", payload["SinkCursor"]) + + def test_sink_task_info_without_cursors(self): + task = OngoingTaskPullReplicationAsSink.from_json({"TaskId": 1, "TaskName": "sink"}) + self.assertIsNone(task.hub_cursor) + self.assertIsNone(task.sink_cursor) From a28821d511d8654a57209b08736aa7094f84a114 Mon Sep 17 00:00:00 2001 From: reforge Date: Thu, 27 Aug 2026 23:34:31 -0300 Subject: [PATCH 7/8] Add the CDC Sink client API Adds the CDC Sink task configuration (tables, column mappings, embedded and linked tables, Postgres publication settings), the add and update operations, and the CdcSink ongoing task type with its task info. The update operation carries the task id in the query string. The server applies an update as a delete followed by an add, so it answers with a new task id and a caller that wants the task afterwards has to read it by name. Both operations reuse the ETL add and update result types, which already carry the raft command index and the task id the server returns; the AI task operations reuse them the same way. validate() mirrors the checks and the messages of the server side configuration, so a caller can find a broken mapping before sending it. Reference surface the target has no home for stays out: the schema and test-mapping families, the task and process state documents, and the CdcSinks list of the database record, whose sibling QueueSinks list is not ported either. Reforge-Run: 20260901T212227Z-1941916-reforge --- ravendb/__init__.py | 14 + .../documents/operations/cdc_sink/__init__.py | 29 + .../operations/cdc_sink/configuration.py | 498 +++++++++++++++ .../operations/cdc_sink/operations.py | 84 +++ ravendb/documents/operations/ongoing_tasks.py | 98 ++- .../tests/operations_tests/test_cdc_sink.py | 575 ++++++++++++++++++ 6 files changed, 1297 insertions(+), 1 deletion(-) create mode 100644 ravendb/documents/operations/cdc_sink/__init__.py create mode 100644 ravendb/documents/operations/cdc_sink/configuration.py create mode 100644 ravendb/documents/operations/cdc_sink/operations.py create mode 100644 ravendb/tests/operations_tests/test_cdc_sink.py diff --git a/ravendb/__init__.py b/ravendb/__init__.py index ce62446b..fb88cd28 100644 --- a/ravendb/__init__.py +++ b/ravendb/__init__.py @@ -195,6 +195,20 @@ from ravendb.documents.operations.ongoing_tasks import ( OngoingTaskPullReplicationAsSink, OngoingTaskPullReplicationAsHub, + OngoingTaskCdcSink, +) +from ravendb.documents.operations.cdc_sink import ( + AddCdcSinkOperation, + CdcColumnMapping, + CdcColumnType, + CdcSinkConfiguration, + CdcSinkEmbeddedTableConfig, + CdcSinkLinkedTableConfig, + CdcSinkOnDeleteConfig, + CdcSinkPostgresSettings, + CdcSinkRelationType, + CdcSinkTableConfig, + UpdateCdcSinkOperation, ) from ravendb.documents.operations.revisions import ( RevisionsCollectionConfiguration, diff --git a/ravendb/documents/operations/cdc_sink/__init__.py b/ravendb/documents/operations/cdc_sink/__init__.py new file mode 100644 index 00000000..044770a0 --- /dev/null +++ b/ravendb/documents/operations/cdc_sink/__init__.py @@ -0,0 +1,29 @@ +from .configuration import ( + CdcColumnMapping, + CdcColumnType, + CdcSinkConfiguration, + CdcSinkEmbeddedTableConfig, + CdcSinkLinkedTableConfig, + CdcSinkOnDeleteConfig, + CdcSinkPostgresSettings, + CdcSinkRelationType, + CdcSinkTableConfig, +) +from .operations import ( + AddCdcSinkOperation, + UpdateCdcSinkOperation, +) + +__all__ = [ + "AddCdcSinkOperation", + "CdcColumnMapping", + "CdcColumnType", + "CdcSinkConfiguration", + "CdcSinkEmbeddedTableConfig", + "CdcSinkLinkedTableConfig", + "CdcSinkOnDeleteConfig", + "CdcSinkPostgresSettings", + "CdcSinkRelationType", + "CdcSinkTableConfig", + "UpdateCdcSinkOperation", +] diff --git a/ravendb/documents/operations/cdc_sink/configuration.py b/ravendb/documents/operations/cdc_sink/configuration.py new file mode 100644 index 00000000..1da9061d --- /dev/null +++ b/ravendb/documents/operations/cdc_sink/configuration.py @@ -0,0 +1,498 @@ +from __future__ import annotations + +import enum +from typing import Any, Dict, List, Optional + + +def _is_blank(value: Optional[str]) -> bool: + return value is None or (isinstance(value, str) and value.strip() == "") + + +def _add_case_insensitive(keys: set, value: str) -> bool: + key = value.lower() + if key in keys: + return False + keys.add(key) + return True + + +class CdcColumnType(enum.Enum): + DEFAULT = "Default" + JSON = "Json" + ATTACHMENT = "Attachment" + + +class CdcSinkRelationType(enum.Enum): + ARRAY = "Array" + MAP = "Map" + VALUE = "Value" + + +class CdcColumnMapping: + def __init__(self, column: Optional[str] = None, name: Optional[str] = None, type: CdcColumnType = None): + self.column = column + self.name = name + self.type = type if type is not None else CdcColumnType.DEFAULT + + def to_json(self) -> Dict[str, Any]: + json_dict = {"Column": self.column, "Name": self.name} + if self.type != CdcColumnType.DEFAULT: + json_dict["Type"] = self.type.value + return json_dict + + @classmethod + def from_json(cls, json_dict: Dict[str, Any]) -> "CdcColumnMapping": + type_raw = json_dict.get("Type") + return cls( + column=json_dict.get("Column"), + name=json_dict.get("Name"), + type=CdcColumnType(type_raw) if type_raw is not None else None, + ) + + +class CdcSinkPostgresSettings: + def __init__(self, publication_name: Optional[str] = None, slot_name: Optional[str] = None): + self.publication_name = publication_name + self.slot_name = slot_name + + def to_json(self) -> Dict[str, Any]: + return {"PublicationName": self.publication_name, "SlotName": self.slot_name} + + @classmethod + def from_json(cls, json_dict: Dict[str, Any]) -> "CdcSinkPostgresSettings": + return cls( + publication_name=json_dict.get("PublicationName"), + slot_name=json_dict.get("SlotName"), + ) + + +class CdcSinkOnDeleteConfig: + def __init__(self, patch: Optional[str] = None, ignore_deletes: bool = False): + self.patch = patch + self.ignore_deletes = ignore_deletes + + def to_json(self) -> Dict[str, Any]: + return {"Patch": self.patch, "IgnoreDeletes": self.ignore_deletes} + + @classmethod + def from_json(cls, json_dict: Dict[str, Any]) -> "CdcSinkOnDeleteConfig": + return cls( + patch=json_dict.get("Patch"), + ignore_deletes=json_dict.get("IgnoreDeletes", False), + ) + + +class CdcSinkLinkedTableConfig: + def __init__( + self, + source_table_schema: Optional[str] = None, + source_table_name: Optional[str] = None, + property_name: Optional[str] = None, + join_columns: Optional[List[str]] = None, + linked_collection_name: Optional[str] = None, + ): + self.source_table_schema = source_table_schema + self.source_table_name = source_table_name + self.property_name = property_name + self.join_columns = join_columns if join_columns is not None else [] + self.linked_collection_name = linked_collection_name + + def to_json(self) -> Dict[str, Any]: + return { + "SourceTableSchema": self.source_table_schema, + "SourceTableName": self.source_table_name, + "PropertyName": self.property_name, + "JoinColumns": self.join_columns, + "LinkedCollectionName": self.linked_collection_name, + } + + @classmethod + def from_json(cls, json_dict: Dict[str, Any]) -> "CdcSinkLinkedTableConfig": + return cls( + source_table_schema=json_dict.get("SourceTableSchema"), + source_table_name=json_dict.get("SourceTableName"), + property_name=json_dict.get("PropertyName"), + join_columns=json_dict.get("JoinColumns") or [], + linked_collection_name=json_dict.get("LinkedCollectionName"), + ) + + +class CdcSinkEmbeddedTableConfig: + def __init__( + self, + source_table_schema: Optional[str] = None, + source_table_name: Optional[str] = None, + property_name: Optional[str] = None, + columns: Optional[List[CdcColumnMapping]] = None, + primary_key_columns: Optional[List[str]] = None, + join_columns: Optional[List[str]] = None, + type: CdcSinkRelationType = None, + patch: Optional[str] = None, + on_delete: Optional[CdcSinkOnDeleteConfig] = None, + case_sensitive_keys: bool = False, + embedded_tables: Optional[List["CdcSinkEmbeddedTableConfig"]] = None, + linked_tables: Optional[List[CdcSinkLinkedTableConfig]] = None, + ): + self.source_table_schema = source_table_schema + self.source_table_name = source_table_name + self.property_name = property_name + self.columns = columns if columns is not None else [] + self.primary_key_columns = primary_key_columns if primary_key_columns is not None else [] + self.join_columns = join_columns if join_columns is not None else [] + self.type = type if type is not None else CdcSinkRelationType.ARRAY + self.patch = patch + self.on_delete = on_delete + self.case_sensitive_keys = case_sensitive_keys + self.embedded_tables = embedded_tables if embedded_tables is not None else [] + self.linked_tables = linked_tables if linked_tables is not None else [] + + def to_json(self) -> Dict[str, Any]: + return { + "SourceTableSchema": self.source_table_schema, + "SourceTableName": self.source_table_name, + "PropertyName": self.property_name, + "Columns": [column.to_json() for column in self.columns], + "PrimaryKeyColumns": self.primary_key_columns, + "JoinColumns": self.join_columns, + "Type": self.type.value, + "Patch": self.patch, + "OnDelete": self.on_delete.to_json() if self.on_delete else None, + "CaseSensitiveKeys": self.case_sensitive_keys, + "EmbeddedTables": [embedded.to_json() for embedded in self.embedded_tables], + "LinkedTables": [linked.to_json() for linked in self.linked_tables], + } + + @classmethod + def from_json(cls, json_dict: Dict[str, Any]) -> "CdcSinkEmbeddedTableConfig": + type_raw = json_dict.get("Type") + return cls( + source_table_schema=json_dict.get("SourceTableSchema"), + source_table_name=json_dict.get("SourceTableName"), + property_name=json_dict.get("PropertyName"), + columns=[CdcColumnMapping.from_json(c) for c in json_dict.get("Columns") or []], + primary_key_columns=json_dict.get("PrimaryKeyColumns") or [], + join_columns=json_dict.get("JoinColumns") or [], + type=CdcSinkRelationType(type_raw) if type_raw is not None else None, + patch=json_dict.get("Patch"), + on_delete=(CdcSinkOnDeleteConfig.from_json(json_dict["OnDelete"]) if json_dict.get("OnDelete") else None), + case_sensitive_keys=json_dict.get("CaseSensitiveKeys", False), + embedded_tables=[CdcSinkEmbeddedTableConfig.from_json(e) for e in json_dict.get("EmbeddedTables") or []], + linked_tables=[CdcSinkLinkedTableConfig.from_json(l) for l in json_dict.get("LinkedTables") or []], + ) + + +class CdcSinkTableConfig: + def __init__( + self, + collection_name: Optional[str] = None, + source_table_schema: Optional[str] = None, + source_table_name: Optional[str] = None, + columns: Optional[List[CdcColumnMapping]] = None, + primary_key_columns: Optional[List[str]] = None, + patch: Optional[str] = None, + on_delete: Optional[CdcSinkOnDeleteConfig] = None, + disabled: bool = False, + embedded_tables: Optional[List[CdcSinkEmbeddedTableConfig]] = None, + linked_tables: Optional[List[CdcSinkLinkedTableConfig]] = None, + ): + self.collection_name = collection_name + self.source_table_schema = source_table_schema + self.source_table_name = source_table_name + self.columns = columns if columns is not None else [] + self.primary_key_columns = primary_key_columns if primary_key_columns is not None else [] + self.patch = patch + self.on_delete = on_delete + self.disabled = disabled + self.embedded_tables = embedded_tables if embedded_tables is not None else [] + self.linked_tables = linked_tables if linked_tables is not None else [] + + def to_json(self) -> Dict[str, Any]: + return { + "CollectionName": self.collection_name, + "SourceTableSchema": self.source_table_schema, + "SourceTableName": self.source_table_name, + "Columns": [column.to_json() for column in self.columns], + "PrimaryKeyColumns": self.primary_key_columns, + "Patch": self.patch, + "OnDelete": self.on_delete.to_json() if self.on_delete else None, + "Disabled": self.disabled, + "EmbeddedTables": [embedded.to_json() for embedded in self.embedded_tables], + "LinkedTables": [linked.to_json() for linked in self.linked_tables], + } + + @classmethod + def from_json(cls, json_dict: Dict[str, Any]) -> "CdcSinkTableConfig": + return cls( + collection_name=json_dict.get("CollectionName"), + source_table_schema=json_dict.get("SourceTableSchema"), + source_table_name=json_dict.get("SourceTableName"), + columns=[CdcColumnMapping.from_json(c) for c in json_dict.get("Columns") or []], + primary_key_columns=json_dict.get("PrimaryKeyColumns") or [], + patch=json_dict.get("Patch"), + on_delete=(CdcSinkOnDeleteConfig.from_json(json_dict["OnDelete"]) if json_dict.get("OnDelete") else None), + disabled=json_dict.get("Disabled", False), + embedded_tables=[CdcSinkEmbeddedTableConfig.from_json(e) for e in json_dict.get("EmbeddedTables") or []], + linked_tables=[CdcSinkLinkedTableConfig.from_json(l) for l in json_dict.get("LinkedTables") or []], + ) + + +class CdcSinkConfiguration: + def __init__( + self, + task_id: int = 0, + disabled: bool = False, + name: Optional[str] = None, + mentor_node: Optional[str] = None, + pin_to_mentor_node: bool = False, + connection_string_name: Optional[str] = None, + tables: Optional[List[CdcSinkTableConfig]] = None, + postgres: Optional[CdcSinkPostgresSettings] = None, + skip_initial_load: bool = False, + ): + self.task_id = task_id + self.disabled = disabled + self.name = name + self.mentor_node = mentor_node + self.pin_to_mentor_node = pin_to_mentor_node + self.connection_string_name = connection_string_name + self.tables = tables if tables is not None else [] + self.postgres = postgres + self.skip_initial_load = skip_initial_load + + def to_json(self) -> Dict[str, Any]: + return { + "Name": self.name, + "TaskId": self.task_id, + "Disabled": self.disabled, + "ConnectionStringName": self.connection_string_name, + "MentorNode": self.mentor_node, + "PinToMentorNode": self.pin_to_mentor_node, + "Tables": [table.to_json() for table in self.tables], + "Postgres": self.postgres.to_json() if self.postgres else None, + "SkipInitialLoad": self.skip_initial_load, + } + + @classmethod + def from_json(cls, json_dict: Dict[str, Any]) -> "CdcSinkConfiguration": + postgres_dict = json_dict.get("Postgres") + return cls( + task_id=json_dict.get("TaskId", 0), + disabled=json_dict.get("Disabled", False), + name=json_dict.get("Name"), + mentor_node=json_dict.get("MentorNode"), + pin_to_mentor_node=json_dict.get("PinToMentorNode", False), + connection_string_name=json_dict.get("ConnectionStringName"), + tables=[CdcSinkTableConfig.from_json(t) for t in json_dict.get("Tables") or []], + postgres=CdcSinkPostgresSettings.from_json(postgres_dict) if postgres_dict else None, + skip_initial_load=json_dict.get("SkipInitialLoad", False), + ) + + def validate(self) -> List[str]: + """Returns the validation errors of this configuration, mirroring the server side CdcSinkConfiguration.""" + errors: List[str] = [] + + if _is_blank(self.name): + errors.append("Name of CDC Sink configuration cannot be empty") + + if _is_blank(self.connection_string_name): + errors.append("ConnectionStringName cannot be empty") + + # The referenced connection string is validated server side; this client holds only its name. + if len(self.tables) == 0: + errors.append("'Tables' list cannot be empty.") + + unique_names = set() + + for table in self.tables: + if _is_blank(table.collection_name): + errors.append("Table collection name must not be empty") + + if _is_blank(table.source_table_name): + errors.append(f"Table '{table.collection_name}' must have a source table name") + + if not table.primary_key_columns: + errors.append(f"Table '{table.collection_name}' must have at least one primary key column") + + if not table.columns: + errors.append(f"Table '{table.collection_name}' must have at least one column mapping") + + name_key = table.collection_name.lower() if table.collection_name else None + if name_key in unique_names: + errors.append(f"Table name '{table.collection_name}' is already defined. Table names must be unique") + else: + unique_names.add(name_key) + + self._validate_primary_key_columns_exist( + table.collection_name, table.primary_key_columns, table.columns, errors + ) + self._validate_columns_and_property_names( + table.collection_name, table.columns, table.embedded_tables, table.linked_tables, errors + ) + self._validate_embedded_tables(table.embedded_tables, table.collection_name, errors) + self._validate_linked_tables(table.linked_tables, table.collection_name, errors) + + return errors + + @staticmethod + def _validate_primary_key_columns_exist( + table_name: str, + primary_key_columns: Optional[List[str]], + columns: Optional[List[CdcColumnMapping]], + errors: List[str], + ) -> None: + if not primary_key_columns or not columns: + return + + column_names = set() + for column in columns: + if column.column is not None: + column_names.add(column.column.lower()) + + for primary_key in primary_key_columns: + if primary_key is None or primary_key.lower() not in column_names: + errors.append( + f"Table '{table_name}': primary key column '{primary_key}' is not listed in the column " + "mappings. Primary key columns must be included in the column mappings so they are stored in " + "the document \u2014 without them, the system cannot identify which array element to update or " + "delete on subsequent changes. Add a column mapping for this column " + f'(e.g. {{ Column = "{primary_key}", Name = "..." }}) or correct the primary key column name.' + ) + + @staticmethod + def _validate_columns_and_property_names( + table_name: str, + columns: Optional[List[CdcColumnMapping]], + embedded_tables: Optional[List[CdcSinkEmbeddedTableConfig]], + linked_tables: Optional[List[CdcSinkLinkedTableConfig]], + errors: List[str], + ) -> None: + column_names = set() + property_names = set() + + if columns is None: + errors.append(f"Table '{table_name}': Columns list is null") + return + + for column in columns: + if _is_blank(column.column): + name_hint = "" if _is_blank(column.name) else f" (Name: '{column.name}')" + errors.append(f"Table '{table_name}': column mapping has an empty Column name{name_hint}") + continue + + if _is_blank(column.name): + errors.append(f"Table '{table_name}': column '{column.column}' has an empty Name") + continue + + if not _add_case_insensitive(column_names, column.column): + errors.append(f"Table '{table_name}': duplicate column '{column.column}'") + + if not _add_case_insensitive(property_names, column.name): + errors.append(f"Table '{table_name}': duplicate target name '{column.name}' (used by multiple columns)") + + if embedded_tables: + for embedded in embedded_tables: + if embedded.property_name is not None and not _add_case_insensitive( + property_names, embedded.property_name + ): + errors.append( + f"Table '{table_name}': property name '{embedded.property_name}' from embedded table " + f"'{embedded.source_table_name}' conflicts with a column mapping or another " + "embedded/linked table" + ) + + if linked_tables: + for linked in linked_tables: + if linked.property_name is not None and not _add_case_insensitive(property_names, linked.property_name): + errors.append( + f"Table '{table_name}': property name '{linked.property_name}' from linked table " + f"'{linked.source_table_name}' conflicts with a column mapping or another " + "embedded/linked table" + ) + + @classmethod + def _validate_embedded_tables( + cls, + embedded_tables: Optional[List[CdcSinkEmbeddedTableConfig]], + parent_name: str, + errors: List[str], + ) -> None: + if embedded_tables is None: + return + + property_names = set() + + for embedded in embedded_tables: + if _is_blank(embedded.source_table_name): + errors.append(f"Embedded table under '{parent_name}' must have a source table name") + elif parent_name and embedded.source_table_name.lower() == parent_name.lower(): + errors.append( + f"Embedded table '{embedded.source_table_name}' under '{parent_name}' cannot reference " + "its own parent table" + ) + + if _is_blank(embedded.property_name): + errors.append( + f"Embedded table '{embedded.source_table_name}' under '{parent_name}' must have a property name" + ) + elif not _add_case_insensitive(property_names, embedded.property_name): + errors.append( + f"Embedded table property name '{embedded.property_name}' under '{parent_name}' is already " + "defined. Property names must be unique within the same parent" + ) + + if not embedded.join_columns: + errors.append( + f"Embedded table '{embedded.source_table_name}' under '{parent_name}' must have join columns" + ) + + if not embedded.primary_key_columns: + errors.append( + f"Embedded table '{embedded.source_table_name}' under '{parent_name}' must have primary key columns" + ) + + if not embedded.columns: + errors.append( + f"Embedded table '{embedded.source_table_name}' under '{parent_name}' must have at least one column mapping" + ) + + cls._validate_primary_key_columns_exist( + embedded.source_table_name, embedded.primary_key_columns, embedded.columns, errors + ) + cls._validate_columns_and_property_names( + embedded.source_table_name, embedded.columns, embedded.embedded_tables, embedded.linked_tables, errors + ) + cls._validate_embedded_tables(embedded.embedded_tables, embedded.source_table_name, errors) + cls._validate_linked_tables(embedded.linked_tables, embedded.source_table_name, errors) + + @staticmethod + def _validate_linked_tables( + linked_tables: Optional[List[CdcSinkLinkedTableConfig]], + parent_name: str, + errors: List[str], + ) -> None: + if linked_tables is None: + return + + property_names = set() + + for linked in linked_tables: + if _is_blank(linked.source_table_name): + errors.append(f"Linked table under '{parent_name}' must have a source table name") + + if _is_blank(linked.property_name): + errors.append( + f"Linked table '{linked.source_table_name}' under '{parent_name}' must have a property name" + ) + elif not _add_case_insensitive(property_names, linked.property_name): + errors.append( + f"Linked table property name '{linked.property_name}' under '{parent_name}' is already defined. " + "Property names must be unique within the same parent" + ) + + if _is_blank(linked.linked_collection_name): + errors.append( + f"Linked table '{linked.source_table_name}' under '{parent_name}' must have a linked collection name" + ) + + if not linked.join_columns: + errors.append(f"Linked table '{linked.source_table_name}' under '{parent_name}' must have join columns") diff --git a/ravendb/documents/operations/cdc_sink/operations.py b/ravendb/documents/operations/cdc_sink/operations.py new file mode 100644 index 00000000..713cf241 --- /dev/null +++ b/ravendb/documents/operations/cdc_sink/operations.py @@ -0,0 +1,84 @@ +from __future__ import annotations + +import json +from typing import TYPE_CHECKING, Optional + +import requests + +from ravendb.documents.operations.cdc_sink.configuration import CdcSinkConfiguration +from ravendb.documents.operations.definitions import MaintenanceOperation +from ravendb.documents.operations.etl.etl_operation_results import AddEtlOperationResult, UpdateEtlOperationResult +from ravendb.http.raven_command import RavenCommand +from ravendb.http.server_node import ServerNode +from ravendb.http.topology import RaftCommand +from ravendb.util.util import RaftIdGenerator + +if TYPE_CHECKING: + from ravendb.documents.conventions import DocumentConventions + + +class AddCdcSinkOperation(MaintenanceOperation[AddEtlOperationResult]): + def __init__(self, configuration: CdcSinkConfiguration): + if configuration is None: + raise ValueError("configuration cannot be None") + self._configuration = configuration + + def get_command(self, conventions: "DocumentConventions") -> RavenCommand[AddEtlOperationResult]: + return AddCdcSinkOperation._AddCdcSinkCommand(self._configuration) + + class _AddCdcSinkCommand(RavenCommand[AddEtlOperationResult], RaftCommand): + def __init__(self, configuration: CdcSinkConfiguration): + super().__init__(AddEtlOperationResult) + self._configuration = configuration + + def is_read_request(self) -> bool: + return False + + def create_request(self, node: ServerNode) -> requests.Request: + url = f"{node.url}/databases/{node.database}/admin/cdc-sink" + request = requests.Request("PUT", url) + request.data = self._configuration.to_json() + return request + + def set_response(self, response: Optional[str], from_cache: bool) -> None: + if response is None: + self._throw_invalid_response() + self.result = AddEtlOperationResult.from_json(json.loads(response)) + + def get_raft_unique_request_id(self) -> str: + return RaftIdGenerator.new_id() + + +class UpdateCdcSinkOperation(MaintenanceOperation[UpdateEtlOperationResult]): + # The server applies an update as a delete followed by an add, so the task id changes. + def __init__(self, task_id: int, configuration: CdcSinkConfiguration): + if configuration is None: + raise ValueError("configuration cannot be None") + self._task_id = task_id + self._configuration = configuration + + def get_command(self, conventions: "DocumentConventions") -> RavenCommand[UpdateEtlOperationResult]: + return UpdateCdcSinkOperation._UpdateCdcSinkCommand(self._task_id, self._configuration) + + class _UpdateCdcSinkCommand(RavenCommand[UpdateEtlOperationResult], RaftCommand): + def __init__(self, task_id: int, configuration: CdcSinkConfiguration): + super().__init__(UpdateEtlOperationResult) + self._task_id = task_id + self._configuration = configuration + + def is_read_request(self) -> bool: + return False + + def create_request(self, node: ServerNode) -> requests.Request: + url = f"{node.url}/databases/{node.database}/admin/cdc-sink?id={self._task_id}" + request = requests.Request("PUT", url) + request.data = self._configuration.to_json() + return request + + def set_response(self, response: Optional[str], from_cache: bool) -> None: + if response is None: + self._throw_invalid_response() + self.result = UpdateEtlOperationResult.from_json(json.loads(response)) + + def get_raft_unique_request_id(self) -> str: + return RaftIdGenerator.new_id() diff --git a/ravendb/documents/operations/ongoing_tasks.py b/ravendb/documents/operations/ongoing_tasks.py index 027ead01..aa295e3c 100644 --- a/ravendb/documents/operations/ongoing_tasks.py +++ b/ravendb/documents/operations/ongoing_tasks.py @@ -1,5 +1,6 @@ from __future__ import annotations import json +from datetime import datetime from enum import Enum from typing import Optional, TYPE_CHECKING, Union @@ -18,6 +19,7 @@ from ravendb.documents.conventions import DocumentConventions from ravendb.documents.operations.ai.gen_ai_configuration import GenAiConfiguration from ravendb.documents.operations.ai.embeddings_generation_configuration import EmbeddingsGenerationConfiguration + from ravendb.documents.operations.cdc_sink.configuration import CdcSinkConfiguration class OngoingTaskType(Enum): @@ -33,6 +35,7 @@ class OngoingTaskType(Enum): PULL_REPLICATION_AS_HUB = "PullReplicationAsHub" PULL_REPLICATION_AS_SINK = "PullReplicationAsSink" QUEUE_SINK = "QueueSink" + CDC_SINK = "CdcSink" EMBEDDINGS_GENERATION = "EmbeddingsGeneration" GEN_AI = "GenAi" @@ -468,6 +471,97 @@ def from_json(cls, json_dict: dict) -> Optional["OngoingTaskPullReplicationAsSin ) +class OngoingTaskCdcSink(OngoingTask): + def __init__( + self, + task_id: Optional[int] = None, + responsible_node: Optional[NodeId] = None, + task_state: Optional[OngoingTaskState] = None, + task_connection_status: Optional[OngoingTaskConnectionStatus] = None, + task_name: Optional[str] = None, + error: Optional[str] = None, + mentor_node: Optional[str] = None, + pin_to_mentor_node: Optional[bool] = None, + configuration: Optional["CdcSinkConfiguration"] = None, + connection_string_name: Optional[str] = None, + factory_name: Optional[str] = None, + last_batch_time: Optional[datetime] = None, + last_checkpoint: Optional[str] = None, + seconds_since_last_batch: Optional[float] = None, + last_activity_time: Optional[datetime] = None, + seconds_since_last_activity: Optional[float] = None, + health_issue: Optional[str] = None, + ): + super().__init__( + task_id=task_id, + task_type=OngoingTaskType.CDC_SINK, + responsible_node=responsible_node, + task_state=task_state, + task_connection_status=task_connection_status, + task_name=task_name, + error=error, + mentor_node=mentor_node, + pin_to_mentor_node=pin_to_mentor_node, + ) + self.configuration = configuration + self.connection_string_name = connection_string_name + self.factory_name = factory_name + self.last_batch_time = last_batch_time + self.last_checkpoint = last_checkpoint + self.seconds_since_last_batch = seconds_since_last_batch + self.last_activity_time = last_activity_time + self.seconds_since_last_activity = seconds_since_last_activity + self.health_issue = health_issue + + def to_json(self) -> dict: + result = super().to_json() + result["ConnectionStringName"] = self.connection_string_name + result["FactoryName"] = self.factory_name + result["Configuration"] = self.configuration.to_json() if self.configuration else None + result["LastBatchTime"] = ( + Utils.datetime_to_string(self.last_batch_time) if self.last_batch_time is not None else None + ) + result["LastCheckpoint"] = self.last_checkpoint + result["SecondsSinceLastBatch"] = self.seconds_since_last_batch + result["LastActivityTime"] = ( + Utils.datetime_to_string(self.last_activity_time) if self.last_activity_time is not None else None + ) + result["SecondsSinceLastActivity"] = self.seconds_since_last_activity + result["HealthIssue"] = self.health_issue + return result + + @classmethod + def from_json(cls, json_dict: dict) -> Optional["OngoingTaskCdcSink"]: + from ravendb.documents.operations.cdc_sink.configuration import CdcSinkConfiguration + + if json_dict is None: + return None + task_state_str = json_dict.get("TaskState") + task_connection_status_str = json_dict.get("TaskConnectionStatus") + config_dict = json_dict.get("Configuration") + return cls( + task_id=json_dict.get("TaskId"), + responsible_node=NodeId.from_json(json_dict.get("ResponsibleNode")), + task_state=OngoingTaskState(task_state_str) if task_state_str else None, + task_connection_status=( + OngoingTaskConnectionStatus(task_connection_status_str) if task_connection_status_str else None + ), + task_name=json_dict.get("TaskName"), + error=json_dict.get("Error"), + mentor_node=json_dict.get("MentorNode"), + pin_to_mentor_node=json_dict.get("PinToMentorNode"), + configuration=CdcSinkConfiguration.from_json(config_dict) if config_dict else None, + connection_string_name=json_dict.get("ConnectionStringName"), + factory_name=json_dict.get("FactoryName"), + last_batch_time=Utils.string_to_datetime(json_dict.get("LastBatchTime")), + last_checkpoint=json_dict.get("LastCheckpoint"), + seconds_since_last_batch=json_dict.get("SecondsSinceLastBatch"), + last_activity_time=Utils.string_to_datetime(json_dict.get("LastActivityTime")), + seconds_since_last_activity=json_dict.get("SecondsSinceLastActivity"), + health_issue=json_dict.get("HealthIssue"), + ) + + class ToggleOngoingTaskStateOperation(MaintenanceOperation[ModifyOngoingTaskResult]): def __init__( self, task_name_or_id: Union[int, str], type_of_task: Optional[OngoingTaskType], disable: Optional[bool] @@ -637,7 +731,7 @@ def set_response(self, response: Optional[str], from_cache: bool) -> None: def _deserialize_task( self, json_dict: dict - ) -> OngoingTask | OngoingTaskGenAi | OngoingTaskEmbeddingsGeneration: + ) -> OngoingTask | OngoingTaskGenAi | OngoingTaskEmbeddingsGeneration | OngoingTaskCdcSink: """Deserialize the task based on its type.""" if self._task_type == OngoingTaskType.GEN_AI: return OngoingTaskGenAi.from_json(json_dict) @@ -645,6 +739,8 @@ def _deserialize_task( return OngoingTaskEmbeddingsGeneration.from_json(json_dict) elif self._task_type == OngoingTaskType.PULL_REPLICATION_AS_SINK: return OngoingTaskPullReplicationAsSink.from_json(json_dict) + elif self._task_type == OngoingTaskType.CDC_SINK: + return OngoingTaskCdcSink.from_json(json_dict) else: # todo: handle more types of tasks return OngoingTask.from_json(json_dict) diff --git a/ravendb/tests/operations_tests/test_cdc_sink.py b/ravendb/tests/operations_tests/test_cdc_sink.py new file mode 100644 index 00000000..66647bb1 --- /dev/null +++ b/ravendb/tests/operations_tests/test_cdc_sink.py @@ -0,0 +1,575 @@ +"""CDC Sink client API tests, ported from CdcSinkCrudTests. + +The CRUD class needs a licensed server: CDC Sink is a licensed feature. +""" + +import json +import os +import unittest +from datetime import datetime + +from ravendb.exceptions.raven_exceptions import RavenException +from ravendb.documents.operations.etl.sql import SqlConnectionString +from ravendb.documents.operations.connection_string.put_connection_string_operation import ( + PutConnectionStringOperation, +) +from ravendb.documents.operations.ongoing_tasks import ( + DeleteOngoingTaskOperation, + GetOngoingTaskInfoOperation, + OngoingTaskCdcSink, + OngoingTaskType, + ToggleOngoingTaskStateOperation, +) +from ravendb.documents.operations.cdc_sink import ( + AddCdcSinkOperation, + CdcColumnMapping, + CdcColumnType, + CdcSinkConfiguration, + CdcSinkEmbeddedTableConfig, + CdcSinkLinkedTableConfig, + CdcSinkOnDeleteConfig, + CdcSinkPostgresSettings, + CdcSinkRelationType, + CdcSinkTableConfig, + UpdateCdcSinkOperation, +) +from ravendb.http.server_node import ServerNode +from ravendb.tests.test_base import TestBase + + +def _build_config(name, connection_string_name): + return CdcSinkConfiguration( + name=name, + connection_string_name=connection_string_name, + tables=[ + CdcSinkTableConfig( + collection_name="Orders", + source_table_schema="dbo", + source_table_name="orders", + columns=[ + CdcColumnMapping(column="order_id", name="OrderId"), + CdcColumnMapping(column="customer_id", name="CustomerId"), + ], + primary_key_columns=["order_id"], + ) + ], + ) + + +class TestCdcSinkWireShape(unittest.TestCase): + """Serialization shapes, pinned to the reference ToJson payloads.""" + + def test_configuration_to_json_key_set(self): + config = _build_config("test-cdc", "sql-cs") + payload = config.to_json() + self.assertEqual( + set(payload.keys()), + { + "Name", + "TaskId", + "Disabled", + "ConnectionStringName", + "MentorNode", + "PinToMentorNode", + "Tables", + "Postgres", + "SkipInitialLoad", + }, + ) + + def test_table_config_to_json_key_set(self): + table = _build_config("c", "cs").tables[0] + self.assertEqual( + set(table.to_json().keys()), + { + "CollectionName", + "SourceTableSchema", + "SourceTableName", + "Columns", + "PrimaryKeyColumns", + "Patch", + "OnDelete", + "Disabled", + "EmbeddedTables", + "LinkedTables", + }, + ) + + def test_column_mapping_type_omitted_when_default(self): + mapping = CdcColumnMapping(column="order_id", name="OrderId") + self.assertEqual(mapping.to_json(), {"Column": "order_id", "Name": "OrderId"}) + mapping.type = CdcColumnType.JSON + self.assertEqual(mapping.to_json(), {"Column": "order_id", "Name": "OrderId", "Type": "Json"}) + mapping.type = CdcColumnType.ATTACHMENT + self.assertEqual( + mapping.to_json(), + {"Column": "order_id", "Name": "OrderId", "Type": "Attachment"}, + ) + + def test_postgres_settings_to_json(self): + postgres = CdcSinkPostgresSettings(publication_name="pub", slot_name="slot") + self.assertEqual(postgres.to_json(), {"PublicationName": "pub", "SlotName": "slot"}) + + def test_embedded_table_config_to_json_writes_type_always(self): + embedded = CdcSinkEmbeddedTableConfig( + source_table_name="order_items", + property_name="Items", + join_columns=["order_id"], + primary_key_columns=["id"], + columns=[CdcColumnMapping(column="id", name="Id")], + ) + payload = embedded.to_json() + self.assertEqual( + set(payload.keys()), + { + "SourceTableSchema", + "SourceTableName", + "PropertyName", + "Columns", + "PrimaryKeyColumns", + "JoinColumns", + "Type", + "Patch", + "OnDelete", + "CaseSensitiveKeys", + "EmbeddedTables", + "LinkedTables", + }, + ) + # Unlike CdcColumnMapping, Type is always written, and the default + # enum member (Array) is written as its name. + self.assertEqual("Array", payload["Type"]) + embedded.type = CdcSinkRelationType.MAP + self.assertEqual("Map", embedded.to_json()["Type"]) + embedded.type = CdcSinkRelationType.VALUE + self.assertEqual("Value", embedded.to_json()["Type"]) + + def test_linked_table_config_to_json(self): + linked = CdcSinkLinkedTableConfig( + source_table_name="customers", + property_name="Customer", + join_columns=["customer_id"], + linked_collection_name="Customers", + ) + self.assertEqual( + linked.to_json(), + { + "SourceTableSchema": None, + "SourceTableName": "customers", + "PropertyName": "Customer", + "JoinColumns": ["customer_id"], + "LinkedCollectionName": "Customers", + }, + ) + + def test_on_delete_config_to_json(self): + on_delete = CdcSinkOnDeleteConfig(patch="this.Archived = true;", ignore_deletes=True) + self.assertEqual( + on_delete.to_json(), + {"Patch": "this.Archived = true;", "IgnoreDeletes": True}, + ) + + def test_configuration_from_json_round_trip(self): + payload = _build_config("test-cdc", "sql-cs").to_json() + payload["Postgres"] = {"PublicationName": "pub", "SlotName": "slot"} + config = CdcSinkConfiguration.from_json(payload) + self.assertEqual("test-cdc", config.name) + self.assertEqual("sql-cs", config.connection_string_name) + self.assertEqual(1, len(config.tables)) + table = config.tables[0] + self.assertEqual("Orders", table.collection_name) + self.assertEqual("dbo", table.source_table_schema) + self.assertEqual("orders", table.source_table_name) + self.assertEqual(["order_id"], table.primary_key_columns) + self.assertEqual("order_id", table.columns[0].column) + self.assertEqual("OrderId", table.columns[0].name) + self.assertEqual("pub", config.postgres.publication_name) + self.assertEqual("slot", config.postgres.slot_name) + + +class TestCdcSinkOperationWireShape(unittest.TestCase): + """Add/update URLs and response parsing, pinned to the reference operations.""" + + def _body(self, request): + return json.loads(request.data) if isinstance(request.data, str) else request.data + + def test_add_operation_url_and_body(self): + config = _build_config("test-cdc", "sql-cs") + command = AddCdcSinkOperation(config).get_command(None) + request = command.create_request(ServerNode("http://localhost:8080", "db")) + self.assertEqual("PUT", request.method) + self.assertEqual("http://localhost:8080/databases/db/admin/cdc-sink", request.url) + self.assertEqual(config.to_json(), self._body(request)) + + def test_update_operation_url_and_body(self): + config = _build_config("test-cdc", "sql-cs") + command = UpdateCdcSinkOperation(42, config).get_command(None) + request = command.create_request(ServerNode("http://localhost:8080", "db")) + self.assertEqual("PUT", request.method) + self.assertEqual("http://localhost:8080/databases/db/admin/cdc-sink?id=42", request.url) + self.assertEqual(config.to_json(), self._body(request)) + + def test_add_result_parses_raft_command_index_and_task_id(self): + command = AddCdcSinkOperation(_build_config("c", "cs")).get_command(None) + command.set_response('{"RaftCommandIndex": 7, "TaskId": 7}', False) + self.assertEqual(7, command.result.raft_command_index) + self.assertEqual(7, command.result.task_id) + + def test_update_result_parses_raft_command_index_and_task_id(self): + command = UpdateCdcSinkOperation(7, _build_config("c", "cs")).get_command(None) + command.set_response('{"RaftCommandIndex": 9, "TaskId": 9}', False) + self.assertEqual(9, command.result.raft_command_index) + self.assertEqual(9, command.result.task_id) + + +class TestCdcSinkTaskInfoWireShape(unittest.TestCase): + """The /task?type=CdcSink response shape and the OngoingTaskCdcSink parse.""" + + _CONFIGURATION = { + "Name": "test-cdc", + "TaskId": 42, + "Disabled": False, + "ConnectionStringName": "sql-cs", + "MentorNode": None, + "PinToMentorNode": False, + "Tables": [ + { + "CollectionName": "Orders", + "SourceTableSchema": "dbo", + "SourceTableName": "orders", + "Columns": [{"Column": "order_id", "Name": "OrderId"}], + "PrimaryKeyColumns": ["order_id"], + "Patch": None, + "OnDelete": None, + "Disabled": False, + "EmbeddedTables": [], + "LinkedTables": [], + } + ], + "Postgres": None, + "SkipInitialLoad": False, + } + + _TASK_INFO = { + "TaskId": 42, + "TaskName": "test-cdc", + "TaskType": "CdcSink", + "TaskState": "Enabled", + "TaskConnectionStatus": "Active", + "ResponsibleNode": {"NodeTag": "A", "NodeUrl": "http://127.0.0.1:8080"}, + "Error": None, + "MentorNode": None, + "PinToMentorNode": False, + "Configuration": _CONFIGURATION, + "ConnectionStringName": "sql-cs", + "FactoryName": "Microsoft.Data.SqlClient", + "LastBatchTime": "2026-06-01T12:00:00.0000000", + "LastCheckpoint": "0/1A2B3C", + "SecondsSinceLastBatch": 12.5, + "LastActivityTime": "2026-06-01T12:00:30.0000000", + "SecondsSinceLastActivity": 3.25, + "HealthIssue": None, + } + + def _get_task_command(self): + operation = GetOngoingTaskInfoOperation(42, OngoingTaskType.CDC_SINK) + return operation.get_command(None) + + def test_get_task_info_url_uses_type_cdc_sink(self): + request = self._get_task_command().create_request(ServerNode("http://localhost:8080", "db")) + self.assertEqual("http://localhost:8080/databases/db/task?key=42&type=CdcSink", request.url) + + def test_get_task_info_dispatches_to_ongoing_task_cdc_sink(self): + command = self._get_task_command() + command.set_response(json.dumps(self._TASK_INFO), False) + task = command.result + self.assertIsInstance(task, OngoingTaskCdcSink) + self.assertEqual(OngoingTaskType.CDC_SINK, task.task_type) + + def test_task_info_from_json_binds_all_fields(self): + task = OngoingTaskCdcSink.from_json(self._TASK_INFO) + self.assertEqual(42, task.task_id) + self.assertEqual("test-cdc", task.task_name) + self.assertEqual("sql-cs", task.connection_string_name) + self.assertEqual("Microsoft.Data.SqlClient", task.factory_name) + self.assertEqual("0/1A2B3C", task.last_checkpoint) + self.assertEqual(12.5, task.seconds_since_last_batch) + self.assertEqual(3.25, task.seconds_since_last_activity) + self.assertEqual(datetime(2026, 6, 1, 12, 0, 0), task.last_batch_time) + self.assertEqual(datetime(2026, 6, 1, 12, 0, 30), task.last_activity_time) + self.assertIsNone(task.health_issue) + self.assertEqual("test-cdc", task.configuration.name) + self.assertEqual("Orders", task.configuration.tables[0].collection_name) + self.assertEqual("order_id", task.configuration.tables[0].columns[0].column) + + def test_task_info_absent_runtime_fields_are_none(self): + task = OngoingTaskCdcSink.from_json( + {"TaskId": 42, "TaskName": "test-cdc", "Configuration": self._CONFIGURATION} + ) + self.assertIsNone(task.last_checkpoint) + self.assertIsNone(task.seconds_since_last_batch) + self.assertIsNone(task.last_batch_time) + self.assertIsNone(task.health_issue) + self.assertIsNone(task.factory_name) + + +class TestCdcSinkValidate(unittest.TestCase): + """validate() error strings, pinned to CdcSinkConfiguration.Validate.""" + + def _config(self, **kwargs): + config = _build_config("test-cdc", "sql-cs") + for key, value in kwargs.items(): + setattr(config, key, value) + return config + + def test_empty_name(self): + errors = self._config(name="").validate() + self.assertEqual(["Name of CDC Sink configuration cannot be empty"], errors) + + def test_empty_connection_string_name(self): + errors = self._config(connection_string_name="").validate() + self.assertEqual(["ConnectionStringName cannot be empty"], errors) + + def test_empty_tables(self): + errors = self._config(tables=[]).validate() + self.assertEqual(["'Tables' list cannot be empty."], errors) + + def test_table_checks(self): + table = CdcSinkTableConfig( + collection_name="Orders", + source_table_name="orders", + columns=[], + primary_key_columns=[], + ) + errors = self._config(tables=[table]).validate() + self.assertIn("Table 'Orders' must have at least one primary key column", errors) + self.assertIn("Table 'Orders' must have at least one column mapping", errors) + + table.collection_name = "" + table.source_table_name = "" + table.columns = [CdcColumnMapping(column="order_id", name="OrderId")] + table.primary_key_columns = ["order_id"] + errors = self._config(tables=[table]).validate() + self.assertIn("Table collection name must not be empty", errors) + self.assertIn("Table '' must have a source table name", errors) + + def test_duplicate_table_name_case_insensitive(self): + second = _build_config("c", "cs").tables[0] + second.collection_name = "orders" + errors = self._config(tables=[_build_config("c", "cs").tables[0], second]).validate() + self.assertIn("Table name 'orders' is already defined. Table names must be unique", errors) + + def test_primary_key_not_in_column_mappings(self): + table = _build_config("c", "cs").tables[0] + table.primary_key_columns = ["missing_pk"] + errors = self._config(tables=[table]).validate() + self.assertIn( + "Table 'Orders': primary key column 'missing_pk' is not listed in the column mappings. " + "Primary key columns must be included in the column mappings so they are stored in the " + "document — without them, the system cannot identify which array element to update or " + "delete on subsequent changes. Add a column mapping for this column " + '(e.g. { Column = "missing_pk", Name = "..." }) or correct the primary key column name.', + errors, + ) + + def test_column_mapping_errors(self): + table = _build_config("c", "cs").tables[0] + table.columns = [ + CdcColumnMapping(column="", name="OrderId"), + CdcColumnMapping(column="customer_id", name=""), + ] + errors = self._config(tables=[table]).validate() + self.assertIn( + "Table 'Orders': column mapping has an empty Column name (Name: 'OrderId')", + errors, + ) + self.assertIn("Table 'Orders': column 'customer_id' has an empty Name", errors) + + table.columns = [ + CdcColumnMapping(column="order_id", name="OrderId"), + CdcColumnMapping(column="order_id", name="Other"), + ] + errors = self._config(tables=[table]).validate() + self.assertIn("Table 'Orders': duplicate column 'order_id'", errors) + + table.columns = [ + CdcColumnMapping(column="order_id", name="OrderId"), + CdcColumnMapping(column="customer_id", name="OrderId"), + ] + errors = self._config(tables=[table]).validate() + self.assertIn( + "Table 'Orders': duplicate target name 'OrderId' (used by multiple columns)", + errors, + ) + + def test_embedded_table_errors(self): + table = _build_config("c", "cs").tables[0] + table.embedded_tables = [ + CdcSinkEmbeddedTableConfig( + source_table_name="order_items", + property_name="Items", + columns=[], + primary_key_columns=[], + join_columns=[], + ) + ] + errors = self._config(tables=[table]).validate() + self.assertIn("Embedded table 'order_items' under 'Orders' must have join columns", errors) + self.assertIn( + "Embedded table 'order_items' under 'Orders' must have primary key columns", + errors, + ) + self.assertIn( + "Embedded table 'order_items' under 'Orders' must have at least one column mapping", + errors, + ) + + embedded = CdcSinkEmbeddedTableConfig( + source_table_name="orders", + property_name="Items", + columns=[CdcColumnMapping(column="id", name="Id")], + primary_key_columns=["id"], + join_columns=["order_id"], + ) + table.embedded_tables = [embedded] + errors = self._config(tables=[table]).validate() + self.assertIn( + "Embedded table 'orders' under 'Orders' cannot reference its own parent table", + errors, + ) + + embedded = CdcSinkEmbeddedTableConfig( + source_table_name="order_items", + property_name="", + columns=[CdcColumnMapping(column="id", name="Id")], + primary_key_columns=["id"], + join_columns=["order_id"], + ) + table.embedded_tables = [embedded] + errors = self._config(tables=[table]).validate() + self.assertIn( + "Embedded table 'order_items' under 'Orders' must have a property name", + errors, + ) + + def test_linked_table_errors(self): + table = _build_config("c", "cs").tables[0] + table.linked_tables = [ + CdcSinkLinkedTableConfig( + source_table_name="", + property_name="", + linked_collection_name="", + join_columns=[], + ) + ] + errors = self._config(tables=[table]).validate() + self.assertIn("Linked table under 'Orders' must have a source table name", errors) + self.assertIn("Linked table '' under 'Orders' must have a property name", errors) + self.assertIn("Linked table '' under 'Orders' must have a linked collection name", errors) + self.assertIn("Linked table '' under 'Orders' must have join columns", errors) + + def test_property_name_conflict_with_column_mapping(self): + table = _build_config("c", "cs").tables[0] + table.embedded_tables = [ + CdcSinkEmbeddedTableConfig( + source_table_name="order_items", + property_name="OrderId", + columns=[CdcColumnMapping(column="id", name="Id")], + primary_key_columns=["id"], + join_columns=["order_id"], + ) + ] + errors = self._config(tables=[table]).validate() + self.assertIn( + "Table 'Orders': property name 'OrderId' from embedded table 'order_items' conflicts with " + "a column mapping or another embedded/linked table", + errors, + ) + + +@unittest.skipIf( + os.environ.get("RAVENDB_LICENSE") is None and os.environ.get("RAVEN_License") is None, + "Insufficient license permissions. Skipping on CI/CD.", +) +class TestCdcSinkCrud(TestBase): + # Ported from CdcSinkCrudTests.cs: CanAddCdcSinkTask, + # CanUpdateCdcSinkTask, CanDeleteCdcSinkTask, CanGetCdcSinkTaskInfo, + # CanToggleCdcSinkTaskState, CanAddMultipleCdcSinkTasks. + + def setUp(self): + super().setUp() + self.connection_string_name = "sql-cs" + self.store.maintenance.send( + PutConnectionStringOperation( + SqlConnectionString( + name=self.connection_string_name, + factory_name="Microsoft.Data.SqlClient", + connection_string="Server=localhost;Database=test;", + ) + ) + ) + + def _send(self, operation): + try: + return self.store.maintenance.send(operation) + except RavenException as e: + # The CDC sink license feature gates the endpoint; the limit is + # asserted from the first license-status refresh onwards, so a + # freshly started server may accept the command before refusing. + if "CDC sink feature" in str(e): + self.skipTest("License does not support the CDC sink feature") + raise + + def _get_task(self, task_id): + return self._send(GetOngoingTaskInfoOperation(task_id, OngoingTaskType.CDC_SINK)) + + def test_add_cdc_sink_task(self): + config = _build_config("test-cdc", self.connection_string_name) + result = self._send(AddCdcSinkOperation(config)) + self.assertIsNotNone(result) + self.assertGreater(result.task_id, 0) + self.assertIsNotNone(result.raft_command_index) + + task = self._get_task(result.task_id) + self.assertIsNotNone(task) + self.assertEqual(task.task_type, OngoingTaskType.CDC_SINK) + self.assertEqual(task.configuration.name, "test-cdc") + self.assertEqual(task.connection_string_name, self.connection_string_name) + + def test_update_cdc_sink_task(self): + config = _build_config("test-cdc", self.connection_string_name) + add_result = self._send(AddCdcSinkOperation(config)) + + config.tables[0].source_table_name = "updated_orders" + self._send(UpdateCdcSinkOperation(add_result.task_id, config)) + + # The server applies an update as delete + re-add, so the task gets a + # new TaskId; task ids are not stable across updates, names are. + task = self._send(GetOngoingTaskInfoOperation("test-cdc", OngoingTaskType.CDC_SINK)) + self.assertIsNotNone(task) + self.assertEqual(task.configuration.tables[0].source_table_name, "updated_orders") + + def test_delete_cdc_sink_task(self): + config = _build_config("test-cdc", self.connection_string_name) + add_result = self._send(AddCdcSinkOperation(config)) + self._send(DeleteOngoingTaskOperation(add_result.task_id, OngoingTaskType.CDC_SINK)) + task = self._get_task(add_result.task_id) + self.assertIsNone(task) + + def test_toggle_cdc_sink_task_state(self): + config = _build_config("test-cdc", self.connection_string_name) + add_result = self._send(AddCdcSinkOperation(config)) + task_id = add_result.task_id + + self._send(ToggleOngoingTaskStateOperation(task_id, OngoingTaskType.CDC_SINK, disable=True)) + self.assertTrue(self._get_task(task_id).configuration.disabled) + + self._send(ToggleOngoingTaskStateOperation(task_id, OngoingTaskType.CDC_SINK, disable=False)) + self.assertFalse(self._get_task(task_id).configuration.disabled) + + def test_add_multiple_cdc_sink_tasks(self): + for name in ("cdc-sink-1", "cdc-sink-2"): + result = self._send(AddCdcSinkOperation(_build_config(name, self.connection_string_name))) + self.assertGreater(result.task_id, 0) + task = self._get_task(result.task_id) + self.assertEqual(task.configuration.name, name) From 685e5f1eeb10f1cd83d9dcaf2033cd817b2f1345 Mon Sep 17 00:00:00 2001 From: reforge Date: Thu, 27 Aug 2026 23:34:31 -0300 Subject: [PATCH 8/8] Pin the client version to 7.2.5 The client-version header the server reads comes from this constant. The package release number in setup.py is a release chore and stays as it is. Reforge-Run: 20260901T212227Z-1941916-reforge --- ravendb/http/request_executor.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ravendb/http/request_executor.py b/ravendb/http/request_executor.py index a9402629..360da4dc 100644 --- a/ravendb/http/request_executor.py +++ b/ravendb/http/request_executor.py @@ -56,7 +56,7 @@ class RequestExecutor: __INITIAL_TOPOLOGY_ETAG = -2 __GLOBAL_APPLICATION_IDENTIFIER = uuid.uuid4() - CLIENT_VERSION = "7.2.3" + CLIENT_VERSION = "7.2.5" logger = logging.getLogger("request_executor") # todo: initializer should take also cryptography certificates