From c115896e34790870f292bcc1bb3c8ec9d295870f Mon Sep 17 00:00:00 2001 From: Emma May Date: Mon, 25 Nov 2024 14:34:28 -0800 Subject: [PATCH] Add support for honoring additional arguments in consul.Consul() (#90) * Add support for honoring the host and port arguments in consul.Consul() * honor the scheme and verify arguments as well * fix lint * forgot to actually run tox after pulling in the linter suggestions --------- Co-authored-by: Emma May --- consul/base.py | 27 ++++++---- tests/test_base.py | 123 ++++++++++++++++++++++++++++++++++++++++++++- 2 files changed, 138 insertions(+), 12 deletions(-) diff --git a/consul/base.py b/consul/base.py index a00aa72..4939e85 100644 --- a/consul/base.py +++ b/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 :" ) 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) diff --git a/tests/test_base.py b/tests/test_base.py index c2e2ab6..3dce651 100644 --- a/tests/test_base.py +++ b/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