RSCBC-147: Fix various issues with running unit tests against a cluster_run-based cluster

sdk/couchbase-connstr/src/lib.rs

@@ -198,26 +198,40 @@ pub async fn resolve(
     conn_spec: ConnSpec,
     dns_config: impl Into<Option<DnsConfig>>,
 ) -> error::Result<ResolvedConnSpec> {
-    let (default_port, has_explicit_scheme, use_ssl) = if let Some(scheme) = &conn_spec.scheme {
+    if let Some(scheme) = &conn_spec.scheme {
         match scheme.as_str() {
-            "couchbase" => (DEFAULT_MEMD_PORT, true, false),
-            "couchbases" => (DEFAULT_SSL_MEMD_PORT, true, true),
-            "couchbase2" => {
-                return handle_couchbase2_scheme(conn_spec);
+            "couchbase" => {
+                handle_couchbase_scheme(conn_spec, dns_config, DEFAULT_MEMD_PORT, true, false).await
             }
-            "" => (DEFAULT_MEMD_PORT, false, false),
-            _ => {
-                return Err(ErrorKind::InvalidArgument {
-                    msg: "unrecognized scheme".to_string(),
-                    arg: "scheme".to_string(),
-                }
-                .into());
+            "couchbases" => {
+                handle_couchbase_scheme(conn_spec, dns_config, DEFAULT_SSL_MEMD_PORT, true, true)
+                    .await
             }
+            "couchbase2" => handle_couchbase2_scheme(conn_spec),
+            "http" => handle_http_scheme(conn_spec, DEFAULT_LEGACY_HTTP_PORT, true, false).await,
+            "https" => handle_http_scheme(conn_spec, DEFAULT_LEGACY_HTTPS_PORT, true, true).await,
+            "" => {
+                handle_couchbase_scheme(conn_spec, dns_config, DEFAULT_MEMD_PORT, false, false)
+                    .await
+            }
+            _ => Err(ErrorKind::InvalidArgument {
+                msg: "unrecognized scheme".to_string(),
+                arg: "scheme".to_string(),
+            }
+            .into()),
         }
     } else {
-        (DEFAULT_MEMD_PORT, false, false)
-    };
+        handle_couchbase_scheme(conn_spec, dns_config, DEFAULT_MEMD_PORT, false, false).await
+    }
+}
 
+async fn handle_couchbase_scheme(
+    conn_spec: ConnSpec,
+    dns_config: impl Into<Option<DnsConfig>>,
+    default_port: u16,
+    has_explicit_scheme: bool,
+    use_ssl: bool,
+) -> error::Result<ResolvedConnSpec> {
     if let Some(srv_record) = conn_spec.srv_record() {
         match lookup_srv(
             &srv_record.scheme,
@@ -358,6 +372,80 @@ fn handle_couchbase2_scheme(conn_spec: ConnSpec) -> error::Result<ResolvedConnSp
     })
 }
 
+async fn handle_http_scheme(
+    conn_spec: ConnSpec,
+    default_port: u16,
+    has_explicit_scheme: bool,
+    use_ssl: bool,
+) -> error::Result<ResolvedConnSpec> {
+    if conn_spec.hosts.is_empty() {
+        let (memd_port, http_port) = if use_ssl {
+            (DEFAULT_SSL_MEMD_PORT, DEFAULT_LEGACY_HTTPS_PORT)
+        } else {
+            (DEFAULT_MEMD_PORT, DEFAULT_LEGACY_HTTP_PORT)
+        };
+
+        return Ok(ResolvedConnSpec {
+            use_ssl,
+            memd_hosts: vec![Address {
+                host: "127.0.0.1".to_string(),
+                port: memd_port,
+            }],
+            http_hosts: vec![Address {
+                host: "127.0.0.1".to_string(),
+                port: http_port,
+            }],
+            couchbase2_host: None,
+            srv_record: None,
+            options: conn_spec.options,
+        });
+    }
+
+    let mut memd_hosts = vec![];
+    let mut http_hosts = vec![];
+    for address in conn_spec.hosts {
+        if let Some(port) = &address.port {
+            if !has_explicit_scheme && address.port != Some(default_port) {
+                return Err(ErrorKind::InvalidArgument {
+                    msg: "ambiguous port without scheme".to_string(),
+                    arg: "port".to_string(),
+                }
+                .into());
+            }
+
+            http_hosts.push(Address {
+                host: address.host,
+                port: *port,
+            });
+        } else {
+            let (memd_port, http_port) = if use_ssl {
+                (DEFAULT_SSL_MEMD_PORT, DEFAULT_LEGACY_HTTPS_PORT)
+            } else {
+                (DEFAULT_MEMD_PORT, DEFAULT_LEGACY_HTTP_PORT)
+            };
+
+            memd_hosts.push(Address {
+                host: address.host.clone(),
+                port: memd_port,
+            });
+
+            http_hosts.push(Address {
+                host: address.host,
+                port: http_port,
+            });
+        }
+    }
+
+    Ok(ResolvedConnSpec {
+        use_ssl,
+        memd_hosts,
+        http_hosts,
+        couchbase2_host: None,
+        srv_record: None,
+        options: conn_spec.options,
+    })
+}
+
 async fn lookup_srv(
     scheme: &str,
     proto: &str,

sdk/couchbase-core/Cargo.toml

@@ -42,14 +42,14 @@ tokio-util = { version = "0.7", features = ["codec"] }
 
 tokio-rustls = { version = "0.26.0", optional = true }
 
+couchbase-connstr = { path = "../couchbase-connstr" }
+
 [dev-dependencies]
 chrono = "0.4.38"
 env_logger = "0.11"
 envconfig = "0.10"
 serial_test = "3.2.0"
 
-couchbase-connstr = { path = "../couchbase-connstr" }
-
 [features]
 dhat-heap = ["dhat"]
 

sdk/couchbase-core/src/agent.rs

@@ -1,4 +1,3 @@
-use crate::address::Address;
 use crate::auth_mechanism::AuthMechanism;
 use crate::authenticator::Authenticator;
 use crate::cbconfig::TerseConfig;
@@ -48,6 +47,7 @@ use crate::vbucketrouter::{
 use crate::{httpx, mgmtx};
 
 use byteorder::BigEndian;
+use couchbase_connstr::Address;
 use futures::executor::block_on;
 use log::{debug, error, info, warn};
 use uuid::Uuid;

sdk/couchbase-core/src/kvclient.rs

@@ -1,4 +1,3 @@
-use crate::address::Address;
 use crate::auth_mechanism::AuthMechanism;
 use crate::authenticator::{Authenticator, UserPassPair};
 use crate::error::Error;
@@ -17,6 +16,7 @@ use crate::service_type::ServiceType;
 use crate::tls_config::TlsConfig;
 use crate::util::hostname_from_addr_str;
 use chrono::{DateTime, FixedOffset, Local, NaiveDateTime, Utc};
+use couchbase_connstr::Address;
 use futures::future::BoxFuture;
 use log::{debug, warn};
 use std::future::Future;

sdk/couchbase-core/src/lib.rs

@@ -2,7 +2,6 @@ extern crate core;
 #[macro_use]
 extern crate lazy_static;
 
-pub mod address;
 pub mod agent;
 pub mod agent_ops;
 pub mod auth_mechanism;
@@ -50,7 +49,7 @@ mod scram;
 pub mod searchcomponent;
 pub mod searchx;
 pub mod service_type;
-mod tls_config;
+pub mod tls_config;
 mod util;
 mod vbucketmap;
 mod vbucketrouter;

sdk/couchbase-core/src/memdx/connection.rs

@@ -1,6 +1,7 @@
 use crate::memdx::error::Error;
 use crate::memdx::error::Result;
 use crate::tls_config::TlsConfig;
+use couchbase_connstr::Address;
 use socket2::TcpKeepalive;
 use std::fmt::Debug;
 use std::io;
@@ -10,7 +11,6 @@ use tokio::io::{AsyncRead, AsyncWrite};
 use tokio::net::TcpStream;
 use tokio::time::{timeout_at, Instant};
 
-use crate::address::Address;
 #[cfg(all(feature = "rustls-tls", not(feature = "native-tls")))]
 use {
     tokio_rustls::rustls::pki_types::DnsName, tokio_rustls::rustls::pki_types::ServerName,

sdk/couchbase-core/src/mgmtx/mgmt.rs

@@ -6,6 +6,7 @@ use crate::mgmtx::error;
 use crate::mgmtx::options::{GetTerseBucketConfigOptions, GetTerseClusterConfigOptions};
 use bytes::Bytes;
 use http::Method;
+use log::debug;
 use serde::de::DeserializeOwned;
 use serde::Deserialize;
 use std::collections::HashMap;
@@ -196,7 +197,7 @@ impl<C: Client> Management<C> {
         opts: &GetTerseClusterConfigOptions<'_>,
     ) -> error::Result<TerseConfig> {
         let method = Method::GET;
-        let path = "/pools/default/nodeServicesStreaming".to_string();
+        let path = "/pools/default/nodeServices".to_string();
 
         let resp = self
             .execute(

sdk/couchbase-core/src/options/agent.rs

@@ -1,4 +1,5 @@
-use crate::address::Address;
+use couchbase_connstr::Address;
+
 use crate::auth_mechanism::AuthMechanism;
 use crate::authenticator::Authenticator;
 use crate::tls_config::TlsConfig;

sdk/couchbase-core/tests/common/default_agent_options.rs

@@ -1,6 +1,7 @@
 use crate::common::test_config::TestSetupConfig;
 use couchbase_core::authenticator::{Authenticator, PasswordAuthenticator};
 use couchbase_core::options::agent::{AgentOptions, SeedConfig};
+use couchbase_core::tls_config::TlsConfig;
 #[cfg(feature = "rustls-tls")]
 use {
     couchbase_core::insecure_certverfier::InsecureCertVerifier, std::sync::Arc,
@@ -9,7 +10,23 @@ use {
 };
 
 pub async fn create_default_options(config: TestSetupConfig) -> AgentOptions {
-    let tls_config = if config.use_ssl {
+    let tls_config = create_tls_config(&config);
+
+    AgentOptions::new(
+        SeedConfig::new()
+            .memd_addrs(config.memd_addrs.clone())
+            .http_addrs(config.http_addrs.clone()),
+        Authenticator::PasswordAuthenticator(PasswordAuthenticator {
+            username: config.username.clone(),
+            password: config.password.clone(),
+        }),
+    )
+    .tls_config(tls_config)
+    .bucket_name(config.bucket.clone())
+}
+
+pub fn create_tls_config(config: &TestSetupConfig) -> Option<TlsConfig> {
+    if config.use_ssl {
         #[cfg(feature = "native-tls")]
         {
             let mut builder = tokio_native_tls::native_tls::TlsConnector::builder();
@@ -20,17 +37,7 @@ pub async fn create_default_options(config: TestSetupConfig) -> AgentOptions {
         Some(get_rustls_config())
     } else {
         None
-    };
-
-    AgentOptions::new(
-        SeedConfig::new().memd_addrs(config.memd_addrs.clone()),
-        Authenticator::PasswordAuthenticator(PasswordAuthenticator {
-            username: config.username.clone(),
-            password: config.password.clone(),
-        }),
-    )
-    .tls_config(tls_config)
-    .bucket_name(config.bucket.clone())
+    }
 }
 
 pub async fn create_options_without_bucket(config: TestSetupConfig) -> AgentOptions {
@@ -48,7 +55,9 @@ pub async fn create_options_without_bucket(config: TestSetupConfig) -> AgentOpti
     };
 
     AgentOptions::new(
-        SeedConfig::new().memd_addrs(config.memd_addrs.clone()),
+        SeedConfig::new()
+            .memd_addrs(config.memd_addrs.clone())
+            .http_addrs(config.http_addrs.clone()),
         Authenticator::PasswordAuthenticator(PasswordAuthenticator {
             username: config.username.clone(),
             password: config.password.clone(),

sdk/couchbase-core/tests/common/test_config.rs

@@ -1,8 +1,7 @@
 use crate::common::default_agent_options;
 use crate::common::node_version::NodeVersion;
 use crate::common::test_agent::TestAgent;
-use couchbase_connstr::ResolvedConnSpec;
-use couchbase_core::address::Address;
+use couchbase_connstr::{Address, ResolvedConnSpec};
 use couchbase_core::agent::Agent;
 use envconfig::Envconfig;
 use lazy_static::lazy_static;
@@ -49,6 +48,7 @@ pub struct TestSetupConfig {
     pub username: String,
     pub password: String,
     pub memd_addrs: Vec<Address>,
+    pub http_addrs: Vec<Address>,
     pub data_timeout: String,
     pub use_ssl: bool,
     pub bucket: String,
@@ -166,14 +166,8 @@ pub async fn create_test_config(test_config: &EnvTestConfig) -> TestSetupConfig
     TestSetupConfig {
         username: test_config.username.clone(),
         password: test_config.password.clone(),
-        memd_addrs: resolved_conn_spec
-            .memd_hosts
-            .iter()
-            .map(|h| Address {
-                host: h.host.clone(),
-                port: h.port,
-            })
-            .collect(),
+        memd_addrs: resolved_conn_spec.memd_hosts.clone(),
+        http_addrs: resolved_conn_spec.http_hosts.clone(),
         data_timeout: test_config.data_timeout.clone(),
         use_ssl: resolved_conn_spec.use_ssl,
         bucket: test_config.default_bucket.clone(),