Add support for honoring additional arguments in consul.Consul()

consul/base.py

@@ -76,13 +76,13 @@ def close(self):
 class Consul:
     def __init__(
         self,
-        host: str = "127.0.0.1",
-        port: int = 8500,
+        host: str | None = None,
+        port: int | None = None,
         token: str | None = None,
-        scheme: str = "http",
+        scheme: str | None = None,
         consistency: str = "default",
         dc=None,
-        verify: bool = True,
+        verify: bool | None = None,
         cert=None,
     ) -> None:
         """
@@ -106,18 +106,25 @@ def __init__(
 
         # TODO: Status
 
-        if os.getenv("CONSUL_HTTP_ADDR"):
+        if os.getenv("CONSUL_HTTP_ADDR") and not (host or port):
             try:
                 host, port = os.getenv("CONSUL_HTTP_ADDR").split(":")  # type: ignore
             except ValueError as err:
                 raise ConsulException(
                     f"CONSUL_HTTP_ADDR ({os.getenv('CONSUL_HTTP_ADDR')}) invalid, does not match <host>:<port>"
                 ) from err
-        use_ssl = os.getenv("CONSUL_HTTP_SSL")
-        if use_ssl is not None:
-            scheme = "https" if use_ssl == "true" else "http"
-        if os.getenv("CONSUL_HTTP_SSL_VERIFY") is not None:
-            verify = os.getenv("CONSUL_HTTP_SSL_VERIFY") == "true"
+        if not host:
+            host = "127.0.0.1"
+        if not port:
+            port = 8500
+
+        if scheme is None:
+            use_ssl = os.getenv("CONSUL_HTTP_SSL")
+            scheme = ("https" if use_ssl.lower() == "true" else "http") if use_ssl else "http"
+
+        if verify is None:
+            ssl_verify = os.getenv("CONSUL_HTTP_SSL_VERIFY")
+            verify = ssl_verify.lower() == "true" if ssl_verify else True
 
         self.http = self.http_connect(host, port, scheme, verify, cert)
         self.token = os.getenv("CONSUL_HTTP_TOKEN", token)

tests/test_base.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import collections
+import contextlib
 import json
 from typing import Any, Callable, Optional
 
@@ -15,7 +16,11 @@ class HTTPClient:
     def __init__(
         self, host: Optional[str] = None, port: Optional[int] = None, scheme=None, verify: bool = True, cert=None
     ) -> None:
-        pass
+        self.host = host
+        self.port = int(port) if port else None
+        self.scheme = scheme
+        self.verify = verify
+        self.cert = cert
 
     def get(self, callback, path, params=None, headers=None):  # pylint: disable=unused-argument
         return Request("get", path, params, headers, None)
@@ -29,7 +34,7 @@ def delete(self, callback, path, params=None, headers=None):  # pylint: disable=
 
 class Consul(consul.base.Consul):
     def http_connect(self, host: str, port: int, scheme, verify: bool = True, cert=None):
-        return HTTPClient(host, port, scheme, verify=verify, cert=None)
+        return HTTPClient(host=host, port=port, scheme=scheme, verify=verify, cert=None)
 
 
 def _should_support(c: Consul) -> tuple[Callable[..., Any], ...]:
@@ -70,6 +75,120 @@ def _should_support_meta(c: Consul) -> tuple[Callable[..., Any], ...]:
     )
 
 
+class TestBaseInit:
+    """
+    Tests that connection arguments are handled
+    """
+
+    @pytest.mark.parametrize(
+        ("env", "host", "port", "want"),
+        [
+            (
+                None,
+                None,
+                None,
+                {
+                    "host": "127.0.0.1",
+                    "port": 8500,
+                },
+            ),
+            (
+                "127.0.0.1:443",
+                None,
+                None,
+                {
+                    "host": "127.0.0.1",
+                    "port": 443,
+                },
+            ),
+            (
+                None,
+                "consul.domain.tld",
+                None,
+                {
+                    "host": "consul.domain.tld",
+                    "port": 8500,
+                },
+            ),
+            (
+                "consul.domain.tld:443",
+                "127.0.0.1",
+                None,
+                {
+                    "host": "127.0.0.1",
+                    "port": 8500,
+                },
+            ),
+            (
+                "consul.domain.tld:443",
+                "127.0.0.1",
+                8080,
+                {
+                    "host": "127.0.0.1",
+                    "port": 8080,
+                },
+            ),
+            (
+                "bad",
+                "127.0.0.1",
+                8080,
+                {
+                    "host": "127.0.0.1",
+                    "port": 8080,
+                },
+            ),
+        ],
+    )
+    def test_base_init_addr(self, monkeypatch, env, host, port, want) -> None:
+        if env:
+            monkeypatch.setenv("CONSUL_HTTP_ADDR", env)
+        else:
+            with contextlib.suppress(KeyError):
+                monkeypatch.delenv("CONSUL_HTTP_ADDR")
+
+        c = Consul(host=host, port=port)
+        assert c.http.host == want["host"]
+        assert c.http.port == want["port"]
+
+    @pytest.mark.parametrize(
+        ("env", "scheme", "want"),
+        [
+            ("true", None, "https"),
+            ("false", None, "http"),
+            (None, "https", "https"),
+            (None, "http", "http"),
+            ("http", "https", "https"),
+        ],
+    )
+    def test_base_init_scheme(self, monkeypatch, env, scheme, want) -> None:
+        if env:
+            monkeypatch.setenv("CONSUL_HTTP_SSL", env)
+        else:
+            with contextlib.suppress(KeyError):
+                monkeypatch.delenv("CONSUL_HTTP_SSL")
+
+        c = Consul(scheme=scheme)
+        assert c.http.scheme == want
+
+    @pytest.mark.parametrize(
+        ("env", "verify", "want"),
+        [
+            ("true", None, True),
+            (None, True, True),
+            ("false", True, True),
+        ],
+    )
+    def test_base_init_verify(self, monkeypatch, env, verify, want) -> None:
+        if env:
+            monkeypatch.setenv("CONSUL_HTTP_SSL_VERIFY", env)
+        else:
+            with contextlib.suppress(KeyError):
+                monkeypatch.delenv("CONSUL_HTTP_SSL_VERIFY")
+
+        c = Consul(verify=verify)
+        assert c.http.verify == want
+
+
 class TestIndex:
     """
     Tests read requests that should support blocking on an index