From e7c4db56ed174872aaa22edb38eecf8e4641ad20 Mon Sep 17 00:00:00 2001 From: TiM Date: Tue, 30 Jun 2026 09:06:27 +1200 Subject: [PATCH] Tests: Run Couchbase container on Docker-assigned ports The Couchbase test container bound host ports 1:1 (8091, 11210, ...), which clashes with any Couchbase Server already running on the workstation and could fail test_couchbase_source_local outright. The pending testcontainers-python Couchbase support binds the same ports 1:1, so it would not remove the conflict either. Switch CouchbaseContainer to Docker-assigned host ports via with_exposed_ports() and advertise those mapped ports to SDK clients through Couchbase alternate addresses (couchbase-cli setting-alternate-address for mgmt/kv/n1ql/capi). Connection strings now use the mapped KV port with network=external, so the SDK bootstraps through the alternate addresses instead of the container-internal ones. To make network=external reach the SDK, CouchbaseSource.dlt_source() now preserves non-ssl URI query parameters when building the connection string, while still consuming ssl=true internally to upgrade couchbase:// to couchbases://. Adds Docker-free unit tests for the URI query-parameter handling and updates the integration-test docstring, which no longer requires stopping a local Couchbase Server. --- omniload/source/couchbase/api.py | 8 +++- tests/main/test_sources.py | 51 ++++++++++++++++++++++++ tests/util/container/impl/couchbase.py | 55 +++++++++++--------------- tests/warehouse/db/test_couchbase.py | 4 +- 4 files changed, 84 insertions(+), 34 deletions(-) diff --git a/omniload/source/couchbase/api.py b/omniload/source/couchbase/api.py index f144ffc77..f1bd1a235 100644 --- a/omniload/source/couchbase/api.py +++ b/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 diff --git a/tests/main/test_sources.py b/tests/main/test_sources.py index ce4d7ea68..450062ce7 100644 --- a/tests/main/test_sources.py +++ b/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): """ diff --git a/tests/util/container/impl/couchbase.py b/tests/util/container/impl/couchbase.py index 8a3f3f346..8bf312796 100644 --- a/tests/util/container/impl/couchbase.py +++ b/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.""" diff --git a/tests/warehouse/db/test_couchbase.py b/tests/warehouse/db/test_couchbase.py index 8f6f46588..39aacfadb 100644 --- a/tests/warehouse/db/test_couchbase.py +++ b/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)