Tests: Run Couchbase container on Docker-assigned ports

omniload/source/couchbase/api.py

@@ -90,7 +90,7 @@ def dlt_source(self, uri: str, table: str, **kwargs):
             netloc = netloc.split("@", 1)[1]
 
         # Parse query parameters from URI
-        from urllib.parse import parse_qs
+        from urllib.parse import parse_qs, urlencode
 
         query_params = parse_qs(parsed.query)
 
@@ -104,6 +104,12 @@ def dlt_source(self, uri: str, table: str, **kwargs):
                 scheme = "couchbases"
 
         connection_string = f"{scheme}://{netloc}"
+        connection_query = urlencode(
+            {key: value for key, value in query_params.items() if key != "ssl"},
+            doseq=True,
+        )
+        if connection_query:
+            connection_string = f"{connection_string}?{connection_query}"
 
         # Extract bucket from URI path if present (e.g., couchbase://host/bucket)
         bucket_from_uri = None

tests/main/test_sources.py

@@ -10,6 +10,7 @@
 import omniload.source.adjust.adapter
 from omniload.core.router import SqlSourceRouter
 from omniload.source.adjust.api import AdjustSource
+from omniload.source.couchbase.api import CouchbaseSource
 from omniload.source.fluxx.api import FluxxSource
 from omniload.source.mongodb.api import MongoDbSource
 
@@ -183,6 +184,56 @@ def mongo(connection_url, database, collection, incremental, parallel):
         self.assertIsNotNone(res)
 
 
+class CouchbaseSourceTest(unittest.TestCase):
+    def test_preserves_sdk_query_params_and_strips_ssl_flag(self):
+        captured = {}
+
+        class DummyResource:
+            max_table_nesting = None
+
+        def couchbase_collection(**kwargs):
+            captured.update(kwargs)
+            return DummyResource()
+
+        source = CouchbaseSource(table_builder=couchbase_collection)
+        res = source.dlt_source(
+            "couchbase://Administrator:password@localhost:33083?network=external&ssl=false",
+            "test_bucket._default._default",
+        )
+
+        self.assertIsNotNone(res)
+        self.assertEqual(
+            captured["connection_string"],
+            "couchbase://localhost:33083?network=external",
+        )
+        self.assertEqual(captured["username"], "Administrator")
+        self.assertEqual(captured["password"], "password")
+        self.assertEqual(captured["bucket"], "test_bucket")
+        self.assertEqual(captured["scope"], "_default")
+        self.assertEqual(captured["collection"], "_default")
+
+    def test_ssl_true_upgrades_scheme_without_leaking_ssl_query_param(self):
+        captured = {}
+
+        class DummyResource:
+            max_table_nesting = None
+
+        def couchbase_collection(**kwargs):
+            captured.update(kwargs)
+            return DummyResource()
+
+        source = CouchbaseSource(table_builder=couchbase_collection)
+        source.dlt_source(
+            "couchbase://Administrator:password@localhost:33083?ssl=true&network=external",
+            "test_bucket._default._default",
+        )
+
+        self.assertEqual(
+            captured["connection_string"],
+            "couchbases://localhost:33083?network=external",
+        )
+
+
 @pytest.fixture(scope="function")
 def monkeypatch_provider(request, monkeypatch):
     """

tests/util/container/impl/couchbase.py

@@ -8,15 +8,7 @@ class CouchbaseContainer(DockerContainer):
 
     def __init__(self, image: str = "couchbase:community", **kwargs):
         super().__init__(image, **kwargs)
-        # Use 1:1 port mapping (requires local Couchbase to be stopped)
-        # This allows SDK to connect without alternate addresses
-        self.with_bind_ports(8091, 8091)
-        self.with_bind_ports(8092, 8092)
-        self.with_bind_ports(8093, 8093)
-        self.with_bind_ports(8094, 8094)
-        self.with_bind_ports(8095, 8095)
-        self.with_bind_ports(8096, 8096)
-        self.with_bind_ports(11210, 11210)
+        self.with_exposed_ports(8091, 8092, 8093, 11210)
         self.username = "Administrator"
         self.password = "password"
         self.bucket_name = "test_bucket"
@@ -36,6 +28,9 @@ def start(self):
         # Create bucket
         self._create_bucket()
 
+        # Advertise Docker's mapped host ports to Couchbase SDK clients.
+        self._setup_alternate_addresses()
+
         # Wait for bucket to be ready
         time.sleep(10)
 
@@ -88,25 +83,21 @@ def _initialize_cluster(self):
 
     def _setup_alternate_addresses(self):
         """Setup alternate addresses for SDK bootstrap."""
-        import requests
-
         host = self.get_container_host_ip()
-        port = self.get_exposed_port(8091)
-
-        # Configure alternate addresses so SDK can connect from outside container
-        requests.post(
-            f"http://{host}:{port}/node/controller/setupAlternateAddresses/external",
-            auth=(self.username, self.password),
-            json={
-                "hostname": host,
-                "mgmt": int(self.get_exposed_port(8091)),
-                "kv": int(self.get_exposed_port(11210)),
-                "n1ql": int(self.get_exposed_port(8093)),
-                "capi": int(self.get_exposed_port(8092)),
-                "fts": int(self.get_exposed_port(8094)),
-                "cbas": int(self.get_exposed_port(8095)),
-                "eventingAdminPort": int(self.get_exposed_port(8096)),
-            },
+        port_spec = ",".join(
+            [
+                f"mgmt={self.get_exposed_port(8091)}",
+                f"kv={self.get_exposed_port(11210)}",
+                f"n1ql={self.get_exposed_port(8093)}",
+                f"capi={self.get_exposed_port(8092)}",
+            ]
+        )
+        self._exec_checked(
+            f"couchbase-cli setting-alternate-address -c 127.0.0.1 "
+            f"-u {self.username} -p {self.password} "
+            f"--set --node 127.0.0.1 "
+            f"--hostname {host} --ports {port_spec}",
+            "alternate address setup",
         )
         time.sleep(2)
 
@@ -174,13 +165,15 @@ def _create_primary_index(self):
 
     def get_connection_string(self) -> str:
         """Get Couchbase connection string."""
-        # With 1:1 port mapping, use localhost
-        return "couchbase://localhost"
+        host = self.get_container_host_ip()
+        port = self.get_exposed_port(11210)
+        return f"couchbase://{host}:{port}?network=external"
 
     def get_connection_url(self) -> str:
         """Get connection URL with credentials."""
-        # With 1:1 port mapping, use localhost
-        return f"couchbase://{self.username}:{self.password}@localhost"
+        host = self.get_container_host_ip()
+        port = self.get_exposed_port(11210)
+        return f"couchbase://{self.username}:{self.password}@{host}:{port}?network=external"
 
     def insert_documents(self, documents: list):
         """Insert documents using Couchbase Python SDK from test machine."""

tests/warehouse/db/test_couchbase.py

@@ -19,8 +19,8 @@ def test_couchbase_source_local(dest):
     """
     Test Couchbase source with local containerized Couchbase instance.
 
-    NOTE: This test requires local Couchbase Server to be stopped first,
-    as it uses 1:1 port mapping (8091, 11210, etc.) to avoid SDK connection issues.
+    The container advertises Docker's mapped host ports through Couchbase alternate
+    addresses, so it can run alongside a local Couchbase Server.
     """
 
     couchbase = CouchbaseContainer(COUCHBASE_IMAGE)