Cover remaining client-surface variations (revisions + pull replication)

Pull replication (secured store, using the workflow TLS setup):
- hub with filtering + SinkToHub/combined mode
- full register -> get -> unregister certificate-based hub-access flow

Pull replication (client-side):
- ReplicationType "Type" serialization
- use_server_certificate drops the certificate private key

Revisions:
- enforce with max_ops_per_second (+ its validation)
- delete by date range, delete for multiple documents
- revert multiple documents by id, bin-cleaner disabled

Author: poissoncorp

ravendb/tests/jvm_migrated_tests/client_tests/replication_tests/test_pull_replication.py

@@ -2,6 +2,7 @@
 import os
 import unittest
 from datetime import timedelta
+from types import SimpleNamespace
 
 from ravendb import (
     ExternalReplication,
@@ -179,3 +180,23 @@ def test_pull_replication_as_sink_json_round_trip(self):
         self.assertEqual("s", parsed.name)
         self.assertEqual(PullReplicationMode.HUB_TO_SINK, parsed.mode)
         self.assertEqual(["users/*"], parsed.allowed_hub_to_sink_paths)
+
+    def test_replication_node_to_json_includes_type(self):
+        sink_json = PullReplicationAsSink(database="db", connection_string_name="cs", hub_name="h").to_json()
+        self.assertEqual("PullAsSink", sink_json["Type"])
+        external_json = ExternalReplication(database="db", connection_string_name="cs").to_json()
+        self.assertEqual("External", external_json["Type"])
+
+    def test_update_sink_with_server_certificate_omits_private_key(self):
+        node = SimpleNamespace(url="http://localhost:8080", database="db")
+        sink = PullReplicationAsSink(
+            database="db", connection_string_name="cs", hub_name="h", name="s", certificate_with_private_key="SECRET"
+        )
+
+        with_server_cert = UpdatePullReplicationAsSinkOperation(sink, use_server_certificate=True).get_command(None)
+        sent = with_server_cert.create_request(node).data["PullReplicationAsSink"]
+        self.assertIsNone(sent["CertificateWithPrivateKey"])
+
+        without = UpdatePullReplicationAsSinkOperation(sink, use_server_certificate=False).get_command(None)
+        sent = without.create_request(node).data["PullReplicationAsSink"]
+        self.assertEqual("SECRET", sent["CertificateWithPrivateKey"])

ravendb/tests/jvm_migrated_tests/client_tests/replication_tests/test_pull_replication_secured.py

@@ -0,0 +1,70 @@
+import base64
+import hashlib
+import os
+import re
+import unittest
+
+from ravendb import (
+    GetPullReplicationTasksInfoOperation,
+    GetReplicationHubAccessOperation,
+    PullReplicationDefinition,
+    PullReplicationMode,
+    PutPullReplicationAsHubOperation,
+    RegisterReplicationHubAccessOperation,
+    ReplicationHubAccess,
+    UnregisterReplicationHubAccessOperation,
+)
+from ravendb.tests.test_base import TestBase
+
+
+def _certificate_base64_and_thumbprint(pem_path):
+    """Extract the certificate (base64 DER) and its SHA-1 thumbprint from a PEM file."""
+    with open(pem_path, "r") as pem_file:
+        pem = pem_file.read()
+    match = re.search(r"-----BEGIN CERTIFICATE-----(.+?)-----END CERTIFICATE-----", pem, re.DOTALL)
+    certificate_base64 = "".join(match.group(1).split())
+    der = base64.b64decode(certificate_base64)
+    return certificate_base64, hashlib.sha1(der).hexdigest().upper()
+
+
+@unittest.skipIf(os.environ.get("RAVENDB_LICENSE") is None, "Insufficient license permissions. Skipping on CI/CD.")
+class TestPullReplicationSecured(TestBase):
+    """Pull-replication surface that the server only permits over SSL (filtering, the
+    SinkToHub direction, and certificate-based hub access) - exercised against a secured store."""
+
+    def setUp(self):
+        super().setUp()
+
+    def test_can_define_hub_with_filtering_and_combined_mode(self):
+        store = self.secured_document_store
+        definition = PullReplicationDefinition(
+            "filtered-hub", mode=PullReplicationMode.HUB_TO_SINK_AND_SINK_TO_HUB, with_filtering=True
+        )
+
+        result = store.maintenance.send(PutPullReplicationAsHubOperation(definition))
+        self.assertIsNotNone(result.task_id)
+
+        info = store.maintenance.send(GetPullReplicationTasksInfoOperation(result.task_id))
+        self.assertEqual("filtered-hub", info.definition.name)
+        self.assertEqual(PullReplicationMode.HUB_TO_SINK_AND_SINK_TO_HUB, info.definition.mode)
+        self.assertTrue(info.definition.with_filtering)
+
+    def test_register_get_and_unregister_replication_hub_access(self):
+        store = self.secured_document_store
+        store.maintenance.send(
+            PutPullReplicationAsHubOperation(PullReplicationDefinition("access-hub", with_filtering=True))
+        )
+
+        certificate_base64, thumbprint = _certificate_base64_and_thumbprint(self.test_client_certificate_url)
+        access = ReplicationHubAccess(
+            name="sink-access", certificate_base64=certificate_base64, allowed_hub_to_sink_paths=["users/*"]
+        )
+        store.maintenance.send(RegisterReplicationHubAccessOperation("access-hub", access))
+
+        registered = store.maintenance.send(GetReplicationHubAccessOperation("access-hub"))
+        self.assertEqual(1, len(registered))
+        self.assertEqual(thumbprint.lower(), registered[0].thumbprint.lower())
+        self.assertEqual(1, len(registered[0].allowed_hub_to_sink_paths))
+
+        store.maintenance.send(UnregisterReplicationHubAccessOperation("access-hub", thumbprint))
+        self.assertEqual([], store.maintenance.send(GetReplicationHubAccessOperation("access-hub")))

ravendb/tests/jvm_migrated_tests/client_tests/revisions_tests/test_revisions_operations.py

@@ -182,3 +182,76 @@ def test_adopt_orphaned_revisions_completes(self):
         # wait_for_completion raises if the operation faults.
         operation = self.store.operations.send_async(AdoptOrphanedRevisionsOperation())
         operation.wait_for_completion()
+
+    def test_enforce_revisions_configuration_with_max_ops_per_second(self):
+        self.setup_revisions(self.store, False, 100)
+        company = self._create_company_with_revisions(4)
+
+        configuration = RevisionsConfiguration()
+        default_config = RevisionsCollectionConfiguration()
+        default_config.minimum_revisions_to_keep = 2
+        configuration.default_config = default_config
+        self.store.maintenance.send(ConfigureRevisionsOperation(configuration))
+
+        parameters = EnforceRevisionsConfigurationOperation.Parameters(max_ops_per_second=10000)
+        operation = self.store.operations.send_async(EnforceRevisionsConfigurationOperation(parameters))
+        operation.wait_for_completion()
+
+        with self.store.open_session() as session:
+            self.assertEqual(2, session.advanced.revisions.get_count_for(company.Id))
+
+    def test_enforce_parameters_reject_non_positive_max_ops_per_second(self):
+        self.assertRaises(ValueError, lambda: EnforceRevisionsConfigurationOperation.Parameters(max_ops_per_second=0))
+
+    def test_delete_revisions_by_date_range(self):
+        self.setup_revisions(self.store, False, 100)
+        company = self._create_company_with_revisions(4)
+
+        result = self.store.maintenance.send(
+            DeleteRevisionsOperation(
+                document_id=company.Id, from_date=datetime(2000, 1, 1), to_date=datetime(2100, 1, 1)
+            )
+        )
+        self.assertEqual(5, result.total_deletes)
+        with self.store.open_session() as session:
+            self.assertEqual(0, session.advanced.revisions.get_count_for(company.Id))
+
+    def test_delete_revisions_for_multiple_documents(self):
+        self.setup_revisions(self.store, False, 100)
+        first = self._create_company_with_revisions(4)
+        second = self._create_company_with_revisions(4)
+
+        result = self.store.maintenance.send(DeleteRevisionsOperation(document_ids=[first.Id, second.Id]))
+        self.assertEqual(10, result.total_deletes)
+        with self.store.open_session() as session:
+            self.assertEqual(0, session.advanced.revisions.get_count_for(first.Id))
+            self.assertEqual(0, session.advanced.revisions.get_count_for(second.Id))
+
+    def test_revert_multiple_revisions_by_id(self):
+        self.setup_revisions(self.store, False, 100)
+
+        id_to_old_change_vector = {}
+        for _ in range(2):
+            company = Company(name="Old Name")
+            with self.store.open_session() as session:
+                session.store(company)
+                session.save_changes()
+            with self.store.open_session() as session:
+                session.load(company.Id, Company).name = "New Name"
+                session.save_changes()
+            with self.store.open_session() as session:
+                metadata = session.advanced.revisions.get_metadata_for(company.Id)
+                id_to_old_change_vector[company.Id] = metadata[1][constants.Documents.Metadata.CHANGE_VECTOR]
+
+        self.store.operations.send(RevertRevisionsByIdOperation(id_to_change_vector=id_to_old_change_vector))
+
+        with self.store.open_session() as session:
+            for company_id in id_to_old_change_vector:
+                self.assertEqual("Old Name", session.load(company_id, Company).name)
+
+    def test_configure_revisions_bin_cleaner_disabled(self):
+        result = self.store.maintenance.send(
+            ConfigureRevisionsBinCleanerOperation(RevisionsBinConfiguration(disabled=True))
+        )
+        self.assertIsNotNone(result.raft_command_index)
+        self.assertGreater(result.raft_command_index, 0)