Test using consul container instead of plain binaries stored in the repo

conftest.py

@@ -0,0 +1,166 @@
+import collections
+import json
+import os
+import socket
+import time
+import uuid
+
+import docker
+import pytest
+import requests
+from requests import RequestException
+
+CONSUL_VERSIONS = ["1.16.1", "1.17.3"]
+
+ConsulInstance = collections.namedtuple("ConsulInstance", ["container", "port", "version"])
+
+# Create a logs directory if it doesn't exist
+LOGS_DIR = os.path.join(os.path.dirname(__file__), "logs")
+os.makedirs(LOGS_DIR, exist_ok=True)
+
+
+def get_free_ports(num, host=None):
+    if not host:
+        host = "127.0.0.1"
+    sockets = []
+    ret = []
+    for _ in range(num):
+        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        s.bind((host, 0))
+        ret.append(s.getsockname()[1])
+        sockets.append(s)
+    for s in sockets:
+        s.close()
+    return ret
+
+
+@pytest.fixture(scope="session", autouse=True)
+def _unset_consul_token():
+    if "CONSUL_HTTP_TOKEN" in os.environ:
+        del os.environ["CONSUL_HTTP_TOKEN"]
+
+
+def start_consul_container(version, acl_master_token=None):
+    """
+    Starts a Consul container. If acl_master_token is None, ACL will be disabled
+    for this server, otherwise it will be enabled and the master token will be
+    set to the supplied token.
+
+    Returns: a tuple of the container object and the HTTP port the instance is listening on
+    """
+    client = docker.from_env()
+    allocated_ports = get_free_ports(5)
+    ports = {
+        "http": allocated_ports[0],
+        "server": allocated_ports[1],
+        "grpc": allocated_ports[2],
+        "serf_lan": allocated_ports[3],
+        "serf_wan": allocated_ports[4],
+    }
+
+    base_config = {
+        "ports": {
+            "https": -1,
+            "dns": -1,
+            "grpc_tls": -1,
+        },
+        "performance": {"raft_multiplier": 1},
+        "enable_script_checks": True,
+    }
+    docker_config = {
+        "ports": {
+            8500: ports["http"],
+            8300: ports["server"],
+            8502: ports["grpc"],
+            8301: ports["serf_lan"],
+            8302: ports["serf_wan"],
+        },
+        "environment": {"CONSUL_LOCAL_CONFIG": json.dumps(base_config)},
+        "detach": True,
+        "name": f"consul_test_{uuid.uuid4().hex[:8]}",  # Add a unique name
+    }
+
+    # Extend the base config with required ACL fields if needed
+    if acl_master_token:
+        acl_config = {
+            "primary_datacenter": "dc1",
+            "acl": {"enabled": True, "tokens": {"initial_management": acl_master_token}},
+        }
+        merged_config = {**base_config, **acl_config}
+        docker_config["environment"]["CONSUL_LOCAL_CONFIG"] = json.dumps(merged_config)
+
+    container = client.containers.run(
+        f"hashicorp/consul:{version}", command="agent -dev -client=0.0.0.0 -log-level trace", **docker_config
+    )
+
+    # Wait for Consul to be ready
+    base_uri = f"http://127.0.0.1:{ports['http']}/v1/"
+    start_time = time.time()
+    global_timeout = 10
+
+    while True:
+        if time.time() - start_time > global_timeout:
+            container.stop()
+            container.remove()
+            raise TimeoutError("Global timeout reached")
+        time.sleep(0.1)
+        try:
+            response = requests.get(base_uri + "status/leader", timeout=2)
+            if response.status_code == 200 and response.json():
+                break
+        except RequestException:
+            continue
+
+    # Additional check to ensure Consul is fully ready
+    for _ in range(10):
+        try:
+            requests.put(base_uri + "agent/service/register", json={"name": "test-service"}, timeout=2)
+            response = requests.get(base_uri + "health/service/test-service", timeout=2)
+            if response.status_code == 200 and response.json():
+                requests.put(base_uri + "agent/service/deregister/test-service", timeout=2)
+                return container, ports["http"]
+        except RequestException:
+            time.sleep(0.5)
+
+    container.stop()
+    container.remove()
+    raise Exception("Failed to verify Consul startup")  # pylint: disable=broad-exception-raised
+
+
+def get_consul_version(port):
+    base_uri = f"http://127.0.0.1:{port}/v1/"
+    response = requests.get(base_uri + "agent/self", timeout=10)
+    return response.json()["Config"]["Version"].strip()
+
+
+def setup_and_teardown_consul(request, version, acl_master_token=None):
+    # Start the container, yield, get container logs, store them in logs/<test_name>.log, stop the container
+    container, port = start_consul_container(version=version, acl_master_token=acl_master_token)
+    version = get_consul_version(port)
+    instance = ConsulInstance(container, port, version)
+
+    yield instance if acl_master_token is None else (instance, acl_master_token)
+
+    logs = container.logs().decode("utf-8")
+    log_file = os.path.join(LOGS_DIR, f"{request.node.name}.log")
+    with open(log_file, "w", encoding="utf-8") as f:
+        f.write(logs)
+
+    container.stop()
+    container.remove()
+
+
+@pytest.fixture(params=CONSUL_VERSIONS)
+def consul_instance(request):
+    yield from setup_and_teardown_consul(request, version=request.param)
+
+
+@pytest.fixture(params=CONSUL_VERSIONS)
+def acl_consul_instance(request):
+    acl_master_token = uuid.uuid4().hex
+    yield from setup_and_teardown_consul(request, version=request.param, acl_master_token=acl_master_token)
+
+
+@pytest.fixture
+def consul_port(consul_instance):
+    return consul_instance.port, consul_instance.version

pyproject.toml

@@ -1,6 +1,7 @@
 [tool.pytest.ini_options]
 addopts = "--cov=. --cov-context=test --durations=0 --durations-min=1.0"
 asyncio_mode = "auto"
+asyncio_default_fixture_loop_scope = "function"
 
 
 [tool.coverage.report]

setup.py

@@ -1,11 +1,9 @@
 import glob
 import os
 import re
-import sys
 
 from setuptools import find_packages, setup
 from setuptools.command.install import install
-from setuptools.command.test import test as TestCommand
 
 with open("consul/__init__.py", encoding="utf-8") as f:
     metadata = dict(re.findall('__([a-z]+)__ = "([^"]+)"', f.read()))
@@ -28,20 +26,6 @@ def run(self):
         install.run(self)
 
 
-class PyTest(TestCommand):
-    # pylint: disable=attribute-defined-outside-init
-    def finalize_options(self):
-        TestCommand.finalize_options(self)
-        self.test_args = []
-        self.test_suite = True
-
-    def run_tests(self):
-        import pytest  # pylint: disable=import-outside-toplevel
-
-        errno = pytest.main(self.test_args)
-        sys.exit(errno)
-
-
 with open("README.md", encoding="utf-8") as f1, open("CHANGELOG.md", encoding="utf-8") as f2:
     long_description = f"{f1.read()}\n\n{f2.read()}"
 
@@ -64,7 +48,7 @@ def run_tests(self):
     data_files=[(".", ["requirements.txt", "tests-requirements.txt"])],
     packages=find_packages(exclude=["tests*"]),
     tests_require=_read_reqs("tests-requirements.txt"),
-    cmdclass={"test": PyTest, "install": Install},
+    cmdclass={"install": Install},
     classifiers=[
         "Development Status :: 3 - Alpha",
         "Intended Audience :: Developers",

tests-requirements.txt

@@ -1,5 +1,6 @@
 aiohttp
 asynctest
+docker
 pre-commit
 pyOpenSSL
 pylint

tests/api/test_agent.py

@@ -37,9 +37,10 @@ def verify_check_status(check_id, status, notes=None):
         assert c.agent.check.register("check name", Check.script("/bin/true", 10, "10m"), check_id="check_id") is True
         verify_and_dereg_check("check_id")
 
-        http_addr = f"http://127.0.0.1:{consul_port}"
-        assert c.agent.check.register("http_check", Check.http(http_addr, "10ms")) is True
-        time.sleep(1)
+        # 1s is the minimal interval for HTTP checks
+        http_addr = "http://localhost:8500"
+        assert c.agent.check.register("http_check", Check.http(http_addr, "1s")) is True
+        time.sleep(1.5)
         verify_check_status("http_check", "passing")
         verify_and_dereg_check("http_check")
 
@@ -70,13 +71,13 @@ def verify_check_status(check_id, status, notes=None):
     def test_service_multi_check(self, consul_port):
         consul_port, _consul_version = consul_port
         c = consul.Consul(port=consul_port)
-        http_addr = f"http://127.0.0.1:{consul_port}"
+        http_addr = "http://127.0.0.1:8500"
         c.agent.service.register(
             "foo1",
-            check=Check.http(http_addr, "10ms"),
+            check=Check.http(http_addr, "1s"),
             extra_checks=[
-                Check.http(http_addr, "20ms"),
-                Check.http(http_addr, "30ms"),
+                Check.http(http_addr, "2s"),
+                Check.http(http_addr, "3s"),
             ],
         )
 
@@ -91,7 +92,7 @@ def test_service_multi_check(self, consul_port):
             "service:foo1:3",
             "serfHealth",
         }
-        time.sleep(1)
+        time.sleep(3.5)
 
         _index, checks = c.health.checks(service="foo1")
         assert [check["CheckID"] for check in checks] == ["service:foo1:1", "service:foo1:2", "service:foo1:3"]

tests/api/test_health.py

@@ -15,9 +15,9 @@ def test_health_service(self, consul_obj):
 
         # register two nodes, one with a long ttl, the other shorter
         c.agent.service.register("foo", service_id="foo:1", check=Check.ttl("10s"), tags=["tag:foo:1"])
-        c.agent.service.register("foo", service_id="foo:2", check=Check.ttl("100ms"))
+        c.agent.service.register("foo", service_id="foo:2", check=Check.ttl("1s"))
 
-        time.sleep(40 / 1000.0)
+        time.sleep(0.2)
 
         # check the nodes show for the /health/service endpoint
         _index, nodes = c.health.service("foo")
@@ -31,14 +31,14 @@ def test_health_service(self, consul_obj):
         c.agent.check.ttl_pass("service:foo:1")
         c.agent.check.ttl_pass("service:foo:2")
 
-        time.sleep(40 / 1000.0)
+        time.sleep(0.2)
 
         # both nodes are now available
         _index, nodes = c.health.service("foo", passing=True)
         assert [node["Service"]["ID"] for node in nodes] == ["foo:1", "foo:2"]
 
         # wait until the short ttl node fails
-        time.sleep(120 / 1000.0)
+        time.sleep(3)
 
         # only one node available
         _index, nodes = c.health.service("foo", passing=True)
@@ -47,7 +47,7 @@ def test_health_service(self, consul_obj):
         # ping the failed node's health check
         c.agent.check.ttl_pass("service:foo:2")
 
-        time.sleep(40 / 1000.0)
+        time.sleep(0.2)
 
         # check both nodes are available
         _index, nodes = c.health.service("foo", passing=True)
@@ -61,7 +61,7 @@ def test_health_service(self, consul_obj):
         c.agent.service.deregister("foo:1")
         c.agent.service.deregister("foo:2")
 
-        time.sleep(40 / 1000.0)
+        time.sleep(0.2)
 
         _index, nodes = c.health.service("foo")
         assert nodes == []
@@ -76,9 +76,9 @@ def test_health_state(self, consul_obj):
 
         # register two nodes, one with a long ttl, the other shorter
         c.agent.service.register("foo", service_id="foo:1", check=Check.ttl("10s"))
-        c.agent.service.register("foo", service_id="foo:2", check=Check.ttl("100ms"))
+        c.agent.service.register("foo", service_id="foo:2", check=Check.ttl("1s"))
 
-        time.sleep(40 / 1000.0)
+        time.sleep(0.2)
 
         # check the nodes show for the /health/state/any endpoint
         _index, nodes = c.health.state("any")
@@ -92,14 +92,14 @@ def test_health_state(self, consul_obj):
         c.agent.check.ttl_pass("service:foo:1")
         c.agent.check.ttl_pass("service:foo:2")
 
-        time.sleep(40 / 1000.0)
+        time.sleep(0.2)
 
         # both nodes are now available
         _index, nodes = c.health.state("passing")
         assert {node["ServiceID"] for node in nodes} == {"", "foo:1", "foo:2"}
 
         # wait until the short ttl node fails
-        time.sleep(2200 / 1000.0)
+        time.sleep(3)
 
         # only one node available
         _index, nodes = c.health.state("passing")
@@ -108,7 +108,7 @@ def test_health_state(self, consul_obj):
         # ping the failed node's health check
         c.agent.check.ttl_pass("service:foo:2")
 
-        time.sleep(40 / 1000.0)
+        time.sleep(0.2)
 
         # check both nodes are available
         _index, nodes = c.health.state("passing")
@@ -118,7 +118,7 @@ def test_health_state(self, consul_obj):
         c.agent.service.deregister("foo:1")
         c.agent.service.deregister("foo:2")
 
-        time.sleep(40 / 1000.0)
+        time.sleep(0.2)
 
         _index, nodes = c.health.state("any")
         assert [node["ServiceID"] for node in nodes] == [""]