diff --git a/ravendb/__init__.py b/ravendb/__init__.py index 172d57b8..fb88cd28 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, @@ -188,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, @@ -356,6 +377,9 @@ GetCertificatesResponse, PutClientCertificateOperation, SecurityClearance, + SsoProvider, + SsoIdentifier, + CertificateUsage, ) from ravendb.serverwide.operations.common import ( BuildNumber, @@ -365,6 +389,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/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/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/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/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/documents/operations/ongoing_tasks.py b/ravendb/documents/operations/ongoing_tasks.py index 62b869c6..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" @@ -392,6 +395,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 +419,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 +434,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 +466,99 @@ 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"), + ) + + +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"), ) @@ -629,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) @@ -637,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/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 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/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/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) 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, + ) 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) 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..a7f67abe --- /dev/null +++ b/ravendb/tests/operations_tests/test_s3_checksum_and_pull_replication.py @@ -0,0 +1,91 @@ +"""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): + 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) + + +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) 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)) 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)