Add support for connecting as named user

README.rst

@@ -58,6 +58,9 @@ Capture MQTT messages, using the `Paho MQTT Python Client`_, in the spirit of
 MQTT server host and port are configurable via pytest cli arguments:
 ``--mqtt-host`` and ``--mqtt-port``. Default values are ``localhost``/``1883``.
 
+You may additionally specify a username and password for connecting to the broker with:
+``--mqtt-username`` and ``--mqtt-password``. Default values are ``guest``/``guest``.
+
 ``mosquitto`` fixture
 =====================
 

pytest_mqtt/capmqtt.py

@@ -28,7 +28,14 @@
 
 
 class MqttClientAdapter(threading.Thread):
-    def __init__(self, on_message_callback: t.Optional[t.Callable] = None, host: str = "localhost", port: int = 1883):
+    def __init__(
+        self,
+        on_message_callback: t.Optional[t.Callable] = None,
+        host: str = "localhost",
+        port: int = 1883,
+        username: str = "guest",
+        password: str = "guest",
+    ):
         super().__init__()
         self.client: mqtt.Client
         if not hasattr(mqtt, "CallbackAPIVersion"):
@@ -42,6 +49,9 @@ def __init__(self, on_message_callback: t.Optional[t.Callable] = None, host: str
         self.on_message_callback = on_message_callback
         self.host = host
         self.port = int(port)
+        self.username = username
+        self.password = password
+
         self.setup()
 
     def setup(self):
@@ -52,6 +62,7 @@ def setup(self):
         client.on_message = self.on_message
         if self.on_message_callback:
             client.on_message = self.on_message_callback
+        client.username_pw_set(self.username, self.password)
 
         logger.debug("[PYTEST] Connecting to MQTT broker")
         client.connect(host=self.host, port=self.port)
@@ -92,12 +103,21 @@ 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], host: str = "localhost", port: int = 1883) -> None:
+    def __init__(
+        self,
+        decode_utf8: t.Optional[bool],
+        host: str = "localhost",
+        port: int = 1883,
+        username: str = "guest",
+        password: str = "guest",
+    ) -> None:
         """Creates a new funcarg."""
         self._buffer: t.List[MqttMessage] = []
         self._decode_utf8: bool = decode_utf8 or False
 
-        self.mqtt_client = MqttClientAdapter(on_message_callback=self.on_message, host=host, port=port)
+        self.mqtt_client = MqttClientAdapter(
+            on_message_callback=self.on_message, host=host, port=port, username=username, password=password
+        )
         self.mqtt_client.start()
         # time.sleep(0.1)
 
@@ -145,13 +165,16 @@ def capmqtt(request, mqtt_settings: MqttSettings):
     # 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
+    username, password = mqtt_settings.username, mqtt_settings.password
 
     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, host=host, port=port)
+    result = MqttCaptureFixture(
+        decode_utf8=capmqtt_decode_utf8, host=host, port=port, username=username, password=password
+    )
     delay()
     yield result
     result.finalize()

pytest_mqtt/model.py

@@ -17,3 +17,5 @@ class MqttMessage:
 class MqttSettings:
     host: str
     port: int
+    username: str
+    password: str

pytest_mqtt/mosquitto.py

@@ -84,13 +84,17 @@ def is_mosquitto_running(host: str, port: int) -> bool:
 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")
+    parser.addoption("--mqtt-username", action="store", type=str, default="guest", help="Username for connection")
+    parser.addoption("--mqtt-password", action="store", type=str, default="guest", help="Password for connection")
 
 
 @pytest.fixture(scope="session")
 def mqtt_settings(pytestconfig) -> MqttSettings:
     return MqttSettings(
         host=pytestconfig.getoption("--mqtt-host"),
         port=pytestconfig.getoption("--mqtt-port"),
+        username=pytestconfig.getoption("--mqtt-username"),
+        password=pytestconfig.getoption("--mqtt-password"),
     )