Implement Multipeer Replicator Support in iOS (API 1.2.1)

.gitignore

@@ -18,3 +18,4 @@ session.log
 http_log
 .terraform
 servers/downloaded
+.version

client/smoke_tests/.gitignore

@@ -1,2 +1,4 @@
 http_log
-local_config.json
+local_config.json
+config.json
+topology.json

client/smoke_tests/README.md

@@ -0,0 +1,10 @@
+Running Smoke Tests
+===================
+
+You can write a config file by copying config_in.json to config.json and adding in a "test-servers" array entry yourself, or alternatively if you want to streamline multiple devices for some reason you can copy topology.example.json to topology.json, edit the various fields and add more entries as you please and then run the following:
+
+`python ../../environment/aws/start_backend.py --topology ./topology.json --tdk-config-out ./config.json --tdk-config-in ./config_in.json`
+
+This will read from config_in.json, build and run the services defined in topology.json and write the resulting config to config.json.
+
+After that it's just standard pytest stuff (e.g. `pytest -v --no-header --config config.json`)

client/smoke_tests/config_in.json

@@ -0,0 +1,4 @@
+{
+    "$schema": "https://packages.couchbase.com/couchbase-lite/testserver.schema.json",
+    "api-version": 1
+}

client/smoke_tests/test_multipeer.py

@@ -0,0 +1,28 @@
+import asyncio
+
+import pytest
+from cbltest import CBLPyTest
+from cbltest.api.multipeer_replicator import MultipeerReplicator
+from cbltest.api.replicator_types import ReplicatorCollectionEntry
+from cbltest.globals import CBLPyTestGlobal
+
+
+class TestMultipeerReplicator:
+    def setup_method(self, method):
+        # If writing a new test do not forget this step or the test server
+        # will not be informed about the currently running test
+        CBLPyTestGlobal.running_test_name = method.__name__
+
+    @pytest.mark.asyncio(loop_scope="session")
+    async def test_start_stop_multipeer(self, cblpytest: CBLPyTest) -> None:
+        dbs = await cblpytest.test_servers[0].create_and_reset_db(["db1"])
+        db = dbs[0]
+        multipeer = MultipeerReplicator(
+            "couchtest", db, [ReplicatorCollectionEntry(["_default._default"])]
+        )
+        await multipeer.start()
+        await asyncio.sleep(2)
+        status = await multipeer.get_status()
+        assert status is not None, "A started multipeer replicator should have a status"
+        assert len(status.replicators) == 0, "Nothing should be found"
+        await multipeer.stop()

client/smoke_tests/topology.example.json

@@ -0,0 +1,11 @@
+{
+    "$schema": "../../environment/aws/topology_setup/topology_schema.json",
+    "test_servers": [
+        {
+            "platform": "dotnet_ios",
+            "cbl_version": "<replace-me>",
+            "dataset_version": "<replace-me>",
+            "location": "<replace-me>"
+        }
+    ]
+}

client/src/cbltest/api/multipeer_replicator.py

@@ -1,10 +1,15 @@
+import asyncio
+from datetime import timedelta
+from time import time
 from typing import cast
 
 from opentelemetry.trace import get_tracer
 
 from cbltest.api.database import Database
+from cbltest.api.error import CblTestError, CblTimeoutError
 from cbltest.api.multipeer_replicator_types import MultipeerReplicatorAuthenticator
 from cbltest.api.replicator import ReplicatorCollectionEntry
+from cbltest.api.replicator_types import ReplicatorActivityLevel
 from cbltest.api.x509_certificate import CertKeyPair, create_leaf_certificate
 from cbltest.logging import cbl_error, cbl_trace
 from cbltest.requests import TestServerRequestType
@@ -135,28 +140,56 @@ async def stop(self) -> None:
 
             self.__id = ""
 
-    async def get_status(self) -> MultipeerReplicatorStatus | None:
+    async def get_status(self) -> MultipeerReplicatorStatus:
         """
         Gets the status of the multipeer replicator
         """
         with self.__tracer.start_as_current_span("get_multipeer_replicator_status"):
             if not self.__id:
-                cbl_error(
-                    "Cannot get status of multipeer replicator, it has not been started"
-                )
-                return None
+                raise CblTestError("MultipeerReplicator start call has not completed!")
 
             req = self.__request_factory.create_request(
                 TestServerRequestType.MULTIPEER_REPLICATOR_STATUS,
                 PostGetMultipeerReplicatorStatusRequestBody(self.__id),
             )
             resp = await self.__request_factory.send_request(self.__index, req)
-            if resp.error is not None:
-                cbl_error(
-                    "Failed to get multipeer replicator status (see trace log for details)"
-                )
-                cbl_trace(resp.error.message)
-                return None
-
             cast_resp = cast(PostGetMultipeerReplicatorStatusResponse, resp)
             return MultipeerReplicatorStatus(cast_resp.replicators)
+
+    async def wait_for_idle(
+        self,
+        interval: timedelta = timedelta(seconds=1),
+        timeout: timedelta = timedelta(seconds=30),
+    ) -> MultipeerReplicatorStatus:
+        """
+        Waits for a given timeout, polling at a set interval, until the Replicator changes to a desired state
+
+        :param activity: The activity level to wait for
+        :param interval: The polling interval (default 1s)
+        :param timeout: The time limit to wait for the state change (default 30s)
+        """
+        with self.__tracer.start_as_current_span("wait_for"):
+            assert interval.total_seconds() > 0.0, (
+                "Zero interval makes no sense, try again"
+            )
+            assert timeout.total_seconds() >= 1.0, (
+                "Timeout too short, must be at least 1 second"
+            )
+
+            all_idle = False
+            start = time()
+            next_status: MultipeerReplicatorStatus = MultipeerReplicatorStatus([])
+            while not all_idle:
+                elapsed = time() - start
+                if elapsed > timeout.total_seconds():
+                    raise CblTimeoutError("Timeout waiting for replicator status")
+
+                next_status = await self.get_status()
+                all_idle = len(next_status.replicators) > 0 and all(
+                    r.status.activity == ReplicatorActivityLevel.IDLE
+                    for r in next_status.replicators
+                )
+                if not all_idle:
+                    await asyncio.sleep(interval.total_seconds())
+
+            return next_status

client/src/cbltest/api/test_functions.py

@@ -118,6 +118,27 @@ def compare_doc_results(
         return DocsCompareResult(True)
 
 
+def compare_doc_results_p2p(
+    local: list[AllDocumentsEntry], remote: list[AllDocumentsEntry]
+) -> DocsCompareResult:
+    local_dict: dict[str, str] = {entry.id: entry.rev for entry in local}
+    remote_dict: dict[str, str] = {entry.id: entry.rev for entry in remote}
+
+    for id in local_dict:
+        if id not in remote_dict:
+            return DocsCompareResult(
+                False, f"Doc '{id}' present in {local_dict} but not {remote_dict}"
+            )
+
+        if not _compare_revisions(local_dict[id], [remote_dict[id], None]):
+            return DocsCompareResult(
+                False,
+                f"Doc '{id}' mismatched revid (local: {local_dict[id]}, remote: {remote_dict[id]})",
+            )
+
+    return DocsCompareResult(True)
+
+
 async def compare_local_and_remote(
     local: Database,
     remote: SyncGateway,

client/src/cbltest/api/x509_certificate.py

@@ -1,8 +1,8 @@
 from datetime import datetime, timedelta, timezone
 
 from cryptography.hazmat.primitives import hashes
-from cryptography.hazmat.primitives.asymmetric import ec
-from cryptography.hazmat.primitives.serialization import Encoding, NoEncryption, pkcs12
+from cryptography.hazmat.primitives.asymmetric import rsa
+from cryptography.hazmat.primitives.serialization import Encoding, PrivateFormat, pkcs12
 from cryptography.x509 import (
     BasicConstraints,
     Certificate,
@@ -22,21 +22,36 @@ class CertKeyPair:
     """
 
     def __init__(
-        self, certificate: Certificate, private_key: ec.EllipticCurvePrivateKey
+        self,
+        certificate: Certificate,
+        private_key: pkcs12.PKCS12PrivateKeyTypes,
+        *,
+        password: str = "couchbase",
     ):
         self.certificate = certificate
         self.private_key = private_key
+        self.password = password
 
     def pfx_bytes(self) -> bytes:
         """
         Returns the certificate and private key in PFX format.
         """
+        # At least at iOS 16, AES is not supported for PFX encryption,
+        # so we have to fallback to this.  Furthermore SHA256 is not
+        # supported either so we have to use SHA1.
+        enc = (
+            PrivateFormat.PKCS12.encryption_builder()
+            .key_cert_algorithm(pkcs12.PBES.PBESv1SHA1And3KeyTripleDESCBC)
+            .hmac_hash(hashes.SHA1())
+            .build(self.password.encode())
+        )
+
         ret_val = pkcs12.serialize_key_and_certificates(
             name=b"cbltest",
             key=self.private_key,
             cert=self.certificate,
             cas=None,
-            encryption_algorithm=NoEncryption(),
+            encryption_algorithm=enc,
         )
 
         return ret_val
@@ -49,7 +64,10 @@ def pem_bytes(self) -> bytes:
 
 
 def create_ca_certificate(CN: str) -> CertKeyPair:
-    private_key = ec.generate_private_key(ec.SECP256R1())
+    private_key = rsa.generate_private_key(
+        public_exponent=65537,
+        key_size=2048,
+    )
     cn_attribute = Name([NameAttribute(NameOID.COMMON_NAME, CN)])
     not_valid_before = datetime.now(timezone.utc)
     not_valid_after = not_valid_before + timedelta(days=1)
@@ -72,7 +90,10 @@ def create_ca_certificate(CN: str) -> CertKeyPair:
 def create_leaf_certificate(
     CN: str, *, issuer_data: CertKeyPair | None = None
 ) -> CertKeyPair:
-    private_key = ec.generate_private_key(ec.SECP256R1())
+    private_key = rsa.generate_private_key(
+        public_exponent=65537,
+        key_size=2048,
+    )
     cn_attribute = Name([NameAttribute(NameOID.COMMON_NAME, CN)])
     not_valid_before = datetime.now(timezone.utc)
     not_valid_after = not_valid_before + timedelta(days=1)

client/src/cbltest/httplog.py

@@ -9,27 +9,40 @@
 class _HttpLogWriter:
     __record_path: Path = Path("http_log")
     __fname_prefix: str
+    __folder_name: str
 
     def __init__(self, num: int):
         test_name = CBLPyTestGlobal.running_test_name
         if test_name.startswith("test_"):
             test_name = test_name[5:]
-        self.__fname_prefix = f"{num:05d}_{test_name}"
+
+        mod_num = num % 100
+        self.__fname_prefix = f"{mod_num:02d}_{test_name}"
+        self.__folder_name = f"{(num // 10000000) * 10000000:08d}"
+
+    def __get_path(self, suffix: str) -> Path:
+        (self.__record_path / self.__folder_name).mkdir(parents=True, exist_ok=True)
+        return (
+            self.__record_path
+            / self.__folder_name
+            / f"{self.__fname_prefix}_{suffix}.txt"
+        )
 
     def write_begin(self, header: str, payload: str) -> None:
-        send_log_path = self.__record_path / f"{self.__fname_prefix}_begin.txt"
+        (self.__record_path / self.__folder_name).mkdir(parents=True, exist_ok=True)
+        send_log_path = self.__get_path("begin")
         with open(send_log_path, "x") as fout:
             fout.write(header)
             fout.write("\n\n")
             fout.write(payload)
 
     def write_error(self, msg: str) -> None:
-        recv_log_path = self.__record_path / f"{self.__fname_prefix}_error.txt"
+        recv_log_path = self.__get_path("error")
         with open(recv_log_path, "x") as fout:
             fout.write(msg)
 
     def write_end(self, header: str, payload: str) -> None:
-        send_log_path = self.__record_path / f"{self.__fname_prefix}_end.txt"
+        send_log_path = self.__get_path("end")
         with open(send_log_path, "x") as fout:
             fout.write(header)
             fout.write("\n\n")

client/src/cbltest/requests.py

@@ -290,8 +290,9 @@ async def send_request(
         the JSON configuration file)"""
         writer = get_next_writer()
         url = self.__server_urls[index]
+        header = f"{r} @ TS-{index}"
         writer.write_begin(
-            str(r), r.payload.serialize() if r.payload is not None else ""
+            header, r.payload.serialize() if r.payload is not None else ""
         )
 
         try:

client/src/cbltest/v1/requests.py

@@ -847,10 +847,11 @@ def to_json(self) -> Any:
         json = {
             "peerGroupID": self.__peerGroupID,
             "database": self.__database,
-            "collections": self.__collections,
+            "collections": [c.to_json() for c in self.collections],
             "identity": {
                 "encoding": "PKCS12",
                 "data": base64.b64encode(self.__identity.pfx_bytes()).decode("utf-8"),
+                "password": self.__identity.password,
             },
         }
 
@@ -1084,3 +1085,53 @@ def __init__(self, uuid: UUID, payload: PostStopListenerRequestBody):
         super().__init__(
             1, uuid, "stopListener", PostStopListenerRequestBody, payload=payload
         )
+
+
+class PostStartMultipeerReplicatorRequest(TestServerRequest):
+    """
+    A POST /startMultipeerReplicator request as specified in version 1 of the
+    `spec <https://github.com/couchbaselabs/couchbase-lite-tests/blob/main/spec/api/api.yaml>`_
+    """
+
+    def __init__(self, uuid: UUID, payload: PostStartMultipeerReplicatorRequestBody):
+        super().__init__(
+            1,
+            uuid,
+            "startMultipeerReplicator",
+            PostStartMultipeerReplicatorRequestBody,
+            payload=payload,
+        )
+
+
+class PostStopMultipeerReplicatorRequest(TestServerRequest):
+    """
+    A POST /startMultipeerReplicator request as specified in version 1 of the
+    `spec <https://github.com/couchbaselabs/couchbase-lite-tests/blob/main/spec/api/api.yaml>`_
+    """
+
+    def __init__(self, uuid: UUID, payload: PostStopMultipeerReplicatorRequestBody):
+        super().__init__(
+            1,
+            uuid,
+            "stopMultipeerReplicator",
+            PostStopMultipeerReplicatorRequestBody,
+            payload=payload,
+        )
+
+
+class PostGetMultipeerReplicatorStatusRequest(TestServerRequest):
+    """
+    A POST /getMultipeerReplicatorStatus request as specified in version 1 of the
+    `spec <https://github.com/couchbaselabs/couchbase-lite-tests/blob/main/spec/api/api.yaml>`_
+    """
+
+    def __init__(
+        self, uuid: UUID, payload: PostGetMultipeerReplicatorStatusRequestBody
+    ):
+        super().__init__(
+            1,
+            uuid,
+            "getMultipeerReplicatorStatus",
+            PostGetMultipeerReplicatorStatusRequestBody,
+            payload=payload,
+        )