✨ Allowing pytest cli arguments for host and port

.gitignore

@@ -4,11 +4,13 @@
 /.ruff_cache
 /.vagrant
 /.venv*
+/.github
+/build
 
 /dist
 
 .coverage*
 coverage.xml
 *.egg-info
 *.pyc
-__pycache__
+__pycache__

README.rst

@@ -47,6 +47,9 @@ fixtures for ``pytest``. It has been conceived for the fine
 Capture MQTT messages, using the `Paho MQTT Python Client`_, in the spirit of
 ``caplog`` and ``capsys``. It can also be used to publish MQTT messages.
 
+MQTT server host and port are configurable via pytest cli arguments:
+``mqtt_host`` and ``mqtt_port`` (defaults to ``localhost`` and ``1883``)
+
 ``mosquitto`` fixture
 =====================
 
@@ -105,9 +108,6 @@ The ``capmqtt_decode_utf8`` setting can be enabled in three ways.
 Issues
 ******
 
-- Both fixtures currently do not support changing the MQTT broker hostname and
-  port number differently than ``localhost:1883``.
-
 - The ``mosquitto`` fixture currently does not support either authentication or
   encryption.
 

pyproject.toml

@@ -61,6 +61,7 @@ dependencies = [
 ]
 
 [project.entry-points.pytest11]
+mqttcliargs = "pytest_mqtt.mqttcliargs"
 capmqtt = "pytest_mqtt.capmqtt"
 mosquitto = "pytest_mqtt.mosquitto"
 

pytest_mqtt/capmqtt.py

@@ -27,10 +27,12 @@
 
 
 class MqttClientAdapter(threading.Thread):
-    def __init__(self, on_message_callback: t.Optional[t.Callable] = None):
+    def __init__(self, on_message_callback: t.Optional[t.Callable] = None, host: str = "localhost", port: int = 1883):
         super().__init__()
         self.client: mqtt.Client = mqtt.Client()
         self.on_message_callback = on_message_callback
+        self.host = host
+        self.port = int(port)
         self.setup()
 
     def setup(self):
@@ -43,7 +45,7 @@ def setup(self):
             client.on_message = self.on_message_callback
 
         logger.debug("[PYTEST] Connecting to MQTT broker")
-        client.connect("localhost", port=1883)
+        client.connect(host=self.host, port=self.port)
         client.subscribe("#")
 
     def run(self):
@@ -75,12 +77,12 @@ def publish(self, topic: str, payload: str, **kwargs) -> mqtt.MQTTMessageInfo:
 class MqttCaptureFixture:
     """Provides access and control of log capturing."""
 
-    def __init__(self, decode_utf8: t.Optional[bool]) -> None:
+    def __init__(self, decode_utf8: t.Optional[bool], host: str = "localhost", port: int = 1883) -> None:
         """Creates a new funcarg."""
         self._buffer: t.List[MqttMessage] = []
         self._decode_utf8: bool = decode_utf8
 
-        self.mqtt_client = MqttClientAdapter(on_message_callback=self.on_message)
+        self.mqtt_client = MqttClientAdapter(on_message_callback=self.on_message, host=host, port=port)
         self.mqtt_client.start()
         # time.sleep(0.1)
 
@@ -119,20 +121,22 @@ def publish(self, topic: str, payload: str, **kwargs) -> mqtt.MQTTMessageInfo:
 
 
 @pytest.fixture(scope="function")
-def capmqtt(request):
+def capmqtt(request, mqttcliargs):
     """Access and control MQTT messages."""
 
     # Configure `capmqtt` fixture, obtaining the `capmqtt_decode_utf8` setting from
     # either a global or module-wide setting, or from a test case marker.
     # https://docs.pytest.org/en/7.1.x/how-to/fixtures.html#fixtures-can-introspect-the-requesting-test-context
     # https://docs.pytest.org/en/7.1.x/how-to/fixtures.html#using-markers-to-pass-data-to-fixtures
+
+    host, port = mqttcliargs
+
     capmqtt_decode_utf8 = (
         getattr(request.config.option, "capmqtt_decode_utf8", False)
         or getattr(request.module, "capmqtt_decode_utf8", False)
         or request.node.get_closest_marker("capmqtt_decode_utf8") is not None
     )
-
-    result = MqttCaptureFixture(decode_utf8=capmqtt_decode_utf8)
+    result = MqttCaptureFixture(decode_utf8=capmqtt_decode_utf8, host=host, port=port)
     delay()
     yield result
     result.finalize()

pytest_mqtt/mosquitto.py

@@ -1,4 +1,5 @@
 # -*- coding: utf-8 -*-
+
 # Copyright (c) 2020-2022 Andreas Motl <[email protected]>
 # Copyright (c) 2020-2022 Richard Pobering <[email protected]>
 #
@@ -36,7 +37,10 @@
 class Mosquitto(BaseImage):
 
     name = "mosquitto"
-    port = 1883
+
+    def __init__(self, host: str = "localhost", port: int = 1883) -> None:
+        self.host = host
+        self.port = port
 
     def check(self):
         # TODO: Add real implementation.
@@ -71,16 +75,18 @@ def run(self):
 mosquitto_image = Mosquitto()
 
 
-def is_mosquitto_running() -> bool:
-    return probe_tcp_connect("localhost", 1883)
+def is_mosquitto_running(host: str, port: int) -> bool:
+    return probe_tcp_connect(host, port)
 
 
 @pytest.fixture(scope="session")
-def mosquitto():
+def mosquitto(request, mqttcliargs):
+
+    host, port = mqttcliargs
 
     # Gracefully skip spinning up the Docker container if Mosquitto is already running.
-    if is_mosquitto_running():
-        yield "localhost", 1883
+    if is_mosquitto_running(host, port):
+        yield host, port
         return
 
     # Spin up Mosquitto container.

pytest_mqtt/mqttcliargs.py

@@ -0,0 +1,10 @@
+import typing as t
+
+import pytest
+
+
+@pytest.fixture(scope="session")
+def mqttcliargs(request) -> t.Tuple[str, int]:
+    host = request.config.getoption("--mqtt_host", "localhost")
+    port = int(request.config.getoption("--mqtt_port", 1883))
+    yield host, port

testing/conftest.py

@@ -0,0 +1,3 @@
+def pytest_addoption(parser) -> None:
+    parser.addoption("--mqtt_host", action="store", default="localhost", help="mqtt host to be connected through")
+    parser.addoption("--mqtt_port", action="store", default=1883, help="mqtt port to be connected through")

testing/test_capmqtt.py

@@ -3,9 +3,13 @@
 
 
 def test_mqtt_client_adapter(mosquitto):
-    mqtt_client = MqttClientAdapter()
+    host, port = mosquitto
+    mqtt_client = MqttClientAdapter(host=host, port=port)
     mqtt_client.start()
 
+    assert mqtt_client.client._host == host
+    assert mqtt_client.client._port == int(port)
+
     # Submit MQTT message.
     message_info = mqtt_client.publish("foo", "bar")
     message_info.wait_for_publish(timeout=0.5)