MQTT Monitor: comparison MqttMonitor/MqttConnectionProfilesDialog.py

-:70b8858199f5
+:5fe4f179975f
 """
 Module implementing a dialog to edit the MQTT connection profiles.
 """
 import collections
+import copy
 from PyQt6.QtCore import pyqtSlot, Qt, QUuid
 from PyQt6.QtWidgets import (
 QDialog, QDialogButtonBox, QAbstractButton, QListWidgetItem, QInputDialog,
 QLineEdit
 @param profiles dictionary containing dictionaries containing the
 connection parameters. Each entry must have the keys
 "BrokerAddress", "BrokerPort", "ClientId", "Protocol",
 "ConnectionTimeout", "Keepalive", "CleanSession", "Username",
 "Password", "WillTopic", "WillMessage", "WillQos", "WillRetain",
-"TlsEnable", "TlsCaCert", "TlsClientCert", "TlsClientKey".
+"TlsEnable", "TlsCaCert", "TlsClientCert", "TlsClientKey",
+"UserProperties".
 @type dict
 @param parent reference to the parent widget
 @type QWidget
 """
 super().__init__(parent)
 EricPathPickerModes.OPEN_FILE_MODE)
 self.tlsSelfSignedClientKeyFilePicker.setFilters(
 self.tr("Key Files (*.key *.pem);;All Files (*)"))
 self.profileTabWidget.setCurrentIndex(0)
+self.connectPropertiesButton.clicked[bool].connect(
+self.__propertiesTypeSelected)
+self.disconnectPropertiesButton.clicked[bool].connect(
+self.__propertiesTypeSelected)
 if len(self.__profiles) == 0:
 self.minusButton.setEnabled(False)
 self.copyButton.setEnabled(False)
 def getProfiles(self):
 """
 Public method to return a dictionary of profiles.
 @return dictionary containing dictionaries containing the defined
-connection profiles. Each entry have the keys "BrokerAddress",
+connection profiles. Each entry has the keys "BrokerAddress",
 "BrokerPort", "ClientId", "Protocol", "ConnectionTimeout",
 "Keepalive", "CleanSession", "Username", "Password", "WillTopic",
 "WillMessage", "WillQos", "WillRetain", "TlsEnable", "TlsCaCert",
-"TlsClientCert", "TlsClientKey".
+"TlsClientCert", "TlsClientKey", "UserProperties".
 @rtype dict
 """
 profilesDict = {}
 profilesDict.update(self.__profiles)
 return profilesDict
 protocol = MqttProtocols.MQTTv311
 elif self.mqttv5Button.isChecked():
 protocol = MqttProtocols.MQTTv5
 else:
 protocol = MqttProtocols.MQTTv311
+if protocol == MqttProtocols.MQTTv5:
+if self.connectPropertiesButton.isChecked():
+self.__userProperties["connect"] = (
+self.propertiesWidget.getProperties())
+else:
+self.__userProperties["disconnect"] = (
+self.propertiesWidget.getProperties())
+self.__userProperties["use_connect"] = (
+self.samePropertiesCheckBox.isChecked())
+else:
+self.__userProperties = {}
 profileName = self.profileEdit.text()
 connectionProfile = {
 "BrokerAddress": self.brokerAddressEdit.text(),
 "BrokerPort": self.brokerPortSpinBox.value(),
 "WillRetain": self.willRetainCheckBox.isChecked(),
 "TlsEnable": self.tlsGroupBox.isChecked(),
 "TlsCaCert": "",
 "TlsClientCert": "",
 "TlsClientKey": "",
+"UserProperties": copy.deepcopy(self.__userProperties),
 }
 if connectionProfile["TlsEnable"]:
 if self.tlsCertsFileButton.isChecked():
 connectionProfile["TlsCaCert"] = self.tlsCertsFilePicker.text()
 elif self.tlsSelfSignedCertsButton.isChecked():
 if profileName is not None:
 self.profileEdit.setText(profileName)
 self.brokerAddressEdit.setText(connectionProfile["BrokerAddress"])
 self.brokerPortSpinBox.setValue(connectionProfile["BrokerPort"])
 self.clientIdEdit.setText(connectionProfile["ClientId"])
+# general tab
 self.mqttv31Button.setChecked(
 connectionProfile["Protocol"] == MqttProtocols.MQTTv31)
 self.mqttv311Button.setChecked(
 connectionProfile["Protocol"] == MqttProtocols.MQTTv311)
 self.mqttv5Button.setChecked(
 connectionProfile["Protocol"] == MqttProtocols.MQTTv5)
+self.on_mqttv5Button_toggled(self.mqttv5Button.isChecked())
 self.connectionTimeoutSpinBox.setValue(
 connectionProfile["ConnectionTimeout"])
 self.keepaliveSpinBox.setValue(connectionProfile["Keepalive"])
 self.cleanSessionCheckBox.setChecked(connectionProfile["CleanSession"])
+# user credentials tab
 self.usernameEdit.setText(connectionProfile["Username"])
 self.passwordEdit.setText(
 pwConvert(connectionProfile["Password"], encode=False))
+# will tab
 self.willTopicEdit.setText(connectionProfile["WillTopic"])
 self.willMessageEdit.setPlainText(connectionProfile["WillMessage"])
 self.willQosSpinBox.setValue(connectionProfile["WillQos"])
 self.willRetainCheckBox.setChecked(connectionProfile["WillRetain"])
+# SSL/TLS tab
 self.tlsGroupBox.setChecked(connectionProfile["TlsEnable"])
 if (
 connectionProfile["TlsCaCert"] and
 connectionProfile["TlsClientCert"]
 ):
 elif connectionProfile["TlsCaCert"]:
 self.tlsCertsFileButton.setChecked(True)
 self.tlsCertsFilePicker.setText(connectionProfile["TlsCaCert"])
 else:
 self.tlsDefaultCertsButton.setChecked(True)
+# user properties tab
+self.__userProperties = copy.deepcopy(
+connectionProfile.get("UserProperties", {}))
+if not self.__userProperties:
+self.__userProperties = {
+"connect": [],
+"disconnect": [],
+"use_connect": True,
+}
+if connectionProfile["Protocol"] == MqttProtocols.MQTTv5:
+self.connectPropertiesButton.setChecked(True)
+self.propertiesWidget.setProperties(
+self.__userProperties["connect"])
+self.samePropertiesCheckBox.setChecked(
+self.__userProperties["use_connect"])
+self.disconnectPropertiesButton.setEnabled(
+not self.__userProperties["use_connect"])
+else:
+self.propertiesWidget.clear()
 self.__populatingProfile = False
 self.showPasswordButton.setChecked(False)
 self.profileFrame.setEnabled(True)
 self.__updateApplyButton()
+self.profileTabWidget.setCurrentIndex(0)
 def __clearProfile(self):
 """
 Private method to clear the profile data entry fields.
 """
 self.tlsCertsFilePicker.setText("")
 self.tlsSelfSignedCertsButton.setChecked(False)
 self.tlsSelfSignedCertsFilePicker.setText("")
 self.tlsSelfSignedClientCertFilePicker.setText("")
 self.tlsSelfSignedClientKeyFilePicker.setText("")
+self.__userProperties = {
+"connect": [],
+"disconnect": [],
+"use_connect": True,
+}
+self.propertiesWidget.clear()
+self.samePropertiesCheckBox.setChecked(True)
+self.connectPropertiesButton.setChecked(True)
+self.disconnectPropertiesButton.setEnabled(False)
 self.__populatingProfile = False
 self.showPasswordButton.setChecked(False)
 self.profileFrame.setEnabled(False)
 self.__updateApplyButton()
 self.tlsSelfSignedClientCertFilePicker.text() !=
 connectionProfile["TlsClientCert"] or
 self.tlsSelfSignedClientKeyFilePicker.text() !=
 connectionProfile["TlsClientKey"]
 )
+# check user properties only, if not yet changed
+if not changed and protocol == MqttProtocols.MQTTv5:
+properties = {
+"connect": self.propertiesWidget.getProperties(),
+"disconnect": self.__userProperties["disconnect"],
+} if self.connectPropertiesButton.isChecked() else {
+"connect": self.__userProperties["connect"],
+"disconnect": self.propertiesWidget.getProperties(),
+}
+changed |= (
+self.samePropertiesCheckBox.isChecked() !=
+connectionProfile["UserProperties"]["use_connect"] or
+sorted(properties["connect"]) !=
+sorted(connectionProfile["UserProperties"]["connect"]) or
+sorted(properties["disconnect"]) !=
+sorted(connectionProfile["UserProperties"]["disconnect"])
+)
 return changed
 else:
 return True
 @param checked current state of the clean session selection
 @type bool
 """
 self.__updateApplyButton()
+@pyqtSlot(bool)
+def on_mqttv5Button_toggled(self, checked):
+"""
+Private slot to handle the selection of the MQTT protocol.
+@param checked state of the button
+@type bool
+"""
+self.profileTabWidget.setTabEnabled(
+self.profileTabWidget.indexOf(self.propertiesTab),
+checked
+)
+# TODO: add code to enable the WILL properties button
 @pyqtSlot(bool)
 def on_showPasswordButton_toggled(self, checked):
 """
 Private slot to show or hide the password.
 @param checked state of the selection
 @type bool
 """
 self.__updateApplyButton()
+@pyqtSlot(bool)
+def on_samePropertiesCheckBox_toggled(self, checked):
+"""
+Private slot to handle a change of the properties usage.
+@param checked flag indicating to use the same user properties for
+CONNECT and DISCONNECT
+@type bool
+"""
+if checked and not self.connectPropertiesButton.isChecked():
+self.connectPropertiesButton.click()
+self.disconnectPropertiesButton.setEnabled(not checked)
+@pyqtSlot(bool)
+def __propertiesTypeSelected(self, checked):
+"""
+Private slot to handle the switching of the user properties type.
+@param checked state of the buttons
+@type bool
+"""
+if checked:
+# handle the selection only
+if self.connectPropertiesButton.isChecked():
+self.__userProperties["disconnect"] = (
+self.propertiesWidget.getProperties())
+self.propertiesWidget.setProperties(
+self.__userProperties["connect"])
+else:
+self.__userProperties["connect"] = (
+self.propertiesWidget.getProperties())
+self.propertiesWidget.setProperties(
+self.__userProperties["disconnect"])
 @pyqtSlot()
 def reject(self):
 """
 Public slot to reject the dialog changes.
 """

Mercurial Repositories > MQTT Monitor / file comparison

comparison: MqttMonitor/MqttConnectionProfilesDialog.py

MqttMonitor/MqttConnectionProfilesDialog.py