✨ 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__

CHANGES.rst

@@ -5,6 +5,9 @@ pytest-mqtt changelog
 
 in progress
 ===========
+- Accept command line options ``--mqtt-host`` and ``--mqtt-port``,
+  in order to connect to an MQTT broker on a different endpoint
+  than ``localhost:1883``. Thanks, @zedfmario.
 
 
 2023-08-03 0.3.1

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``. Default values are ``localhost``/``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.
 

pytest_mqtt/capmqtt.py

@@ -20,17 +20,19 @@
 import paho.mqtt.client as mqtt
 import pytest
 
-from pytest_mqtt.model import MqttMessage
+from pytest_mqtt.model import MqttMessage, MqttSettings
 from pytest_mqtt.util import delay
 
 logger = logging.getLogger(__name__)
 
 
 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, mqtt_settings: MqttSettings):
     """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 = mqtt_settings.host, mqtt_settings.port
+
     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/model.py

@@ -11,3 +11,9 @@ class MqttMessage:
     topic: str
     payload: t.Union[str, bytes]
     userdata: t.Optional[t.Union[t.Dict, None]]
+
+
+@dataclasses.dataclass
+class MqttSettings:
+    host: str
+    port: int

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]>
 #
@@ -20,6 +21,7 @@
 from pytest_docker_fixtures import images
 from pytest_docker_fixtures.containers._base import BaseImage
 
+from pytest_mqtt.model import MqttSettings
 from pytest_mqtt.util import probe_tcp_connect
 
 images.settings["mosquitto"] = {
@@ -36,7 +38,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 +76,18 @@
 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(mqtt_settings: MqttSettings):
+
+    host, port = mqtt_settings.host, mqtt_settings.port
 
     # 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.

testing/conftest.py

@@ -0,0 +1,16 @@
+import pytest
+
+from pytest_mqtt.model import MqttSettings
+
+
+def pytest_addoption(parser) -> None:
+    parser.addoption("--mqtt-host", action="store", type=str, default="localhost", help="MQTT host name")
+    parser.addoption("--mqtt-port", action="store", type=int, default=1883, help="MQTT port number")
+
+
+@pytest.fixture(scope="session")
+def mqtt_settings(pytestconfig) -> MqttSettings:
+    return MqttSettings(
+        host=pytestconfig.getoption("--mqtt-host"),
+        port=pytestconfig.getoption("--mqtt-port"),
+    )

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)