Mercurial > libervia-backend
diff sat/plugins/plugin_xep_0384.py @ 4037:524856bd7b19
massive refactoring to switch from camelCase to snake_case:
historically, Libervia (SàT before) was using camelCase as allowed by PEP8 when using a
pre-PEP8 code, to use the same coding style as in Twisted.
However, snake_case is more readable and it's better to follow PEP8 best practices, so it
has been decided to move on full snake_case. Because Libervia has a huge codebase, this
ended with a ugly mix of camelCase and snake_case.
To fix that, this patch does a big refactoring by renaming every function and method
(including bridge) that are not coming from Twisted or Wokkel, to use fully snake_case.
This is a massive change, and may result in some bugs.
author | Goffi <goffi@goffi.org> |
---|---|
date | Sat, 08 Apr 2023 13:54:42 +0200 |
parents | 1d5a81e3c9e8 |
children | c23cad65ae99 |
line wrap: on
line diff
--- a/sat/plugins/plugin_xep_0384.py Fri Apr 07 15:18:39 2023 +0200 +++ b/sat/plugins/plugin_xep_0384.py Sat Apr 08 13:54:42 2023 +0200 @@ -205,7 +205,7 @@ def __init__(self, profile: str, own_bare_jid: str) -> None: """ @param profile: The profile this OMEMO data belongs to. - @param own_bare_jid: The own bare JID, to return by the :meth:`loadOwnData` call. + @param own_bare_jid: The own bare JID, to return by the :meth:`load_own_data` call. """ self.__storage = persistent.LazyPersistentBinaryDict("XEP-0384", profile) @@ -346,7 +346,7 @@ node = f"eu.siacs.conversations.axolotl.bundles:{device_id}" try: - items, __ = await xep_0060.getItems(client, jid.JID(bare_jid), node, max_items=1) + items, __ = await xep_0060.get_items(client, jid.JID(bare_jid), node, max_items=1) except Exception as e: raise omemo.BundleDownloadFailed( f"Bundle download failed for {bare_jid}: {device_id} under namespace" @@ -726,7 +726,7 @@ ) # Finally, encrypt and send the trust message! - message_data = client.generateMessageXML(MessageData({ + message_data = client.generate_message_xml(MessageData({ "from": own_jid, "to": recipient_jid, "uid": str(uuid.uuid4()), @@ -811,7 +811,7 @@ with XMPP interactions and trust handled via the SAT instance. """ - client = sat.getClient(profile) + client = sat.get_client(profile) xep_0060 = cast(XEP_0060, sat.plugins["XEP-0060"]) class SessionManagerImpl(omemo.SessionManager): @@ -827,7 +827,7 @@ node = "urn:xmpp:omemo:2:bundles" try: - await xep_0060.sendItem( + await xep_0060.send_item( client, client.jid.userhostJID(), node, @@ -867,7 +867,7 @@ node = f"eu.siacs.conversations.axolotl.bundles:{bundle.device_id}" try: - await xep_0060.sendItem( + await xep_0060.send_item( client, client.jid.userhostJID(), node, @@ -897,7 +897,7 @@ node = "urn:xmpp:omemo:2:bundles" try: - items, __ = await xep_0060.getItems( + items, __ = await xep_0060.get_items( client, jid.JID(bare_jid), node, @@ -951,7 +951,7 @@ node = "urn:xmpp:omemo:2:bundles" try: - await xep_0060.retractItems( + await xep_0060.retract_items( client, client.jid.userhostJID(), node, @@ -1000,7 +1000,7 @@ raise omemo.UnknownNamespace(f"Unknown namespace: {namespace}") try: - await xep_0060.sendItem( + await xep_0060.send_item( client, client.jid.userhostJID(), node, @@ -1050,7 +1050,7 @@ raise omemo.UnknownNamespace(f"Unknown namespace: {namespace}") try: - items, __ = await xep_0060.getItems(client, jid.JID(bare_jid), node) + items, __ = await xep_0060.get_items(client, jid.JID(bare_jid), node) except exceptions.NotFound: return {} except Exception as e: @@ -1117,7 +1117,7 @@ # on the trust system and phase if trust_level is TrustLevel.BLINDLY_TRUSTED: # Get the name of the active trust system - trust_system = cast(str, sat.memory.getParamA( + trust_system = cast(str, sat.memory.param_get_a( PARAM_NAME, PARAM_CATEGORY, profile_key=profile @@ -1247,7 +1247,7 @@ if element is None: raise omemo.UnknownNamespace(f"Unknown namespace: {message.namespace}") - message_data = client.generateMessageXML(MessageData({ + message_data = client.generate_message_xml(MessageData({ "from": client.jid, "to": jid.JID(bare_jid), "uid": str(uuid.uuid4()), @@ -1282,7 +1282,7 @@ """ # This session manager handles encryption with both twomemo and oldmemo, but - # both are currently registered as different plugins and the `deferXMLUI` + # both are currently registered as different plugins and the `defer_xmlui` # below requires a single namespace identifying the encryption plugin. Thus, # get the namespace of the requested encryption method from the encryption # session using the feedback JID. @@ -1313,7 +1313,7 @@ own_device, __ = await self.get_own_device_information() - trust_ui.changeContainer("label") + trust_ui.change_container("label") trust_ui.addLabel(D_("This device ID")) trust_ui.addText(str(own_device.device_id)) trust_ui.addLabel(D_("This device's fingerprint")) @@ -1333,11 +1333,11 @@ trust_ui.addLabel(D_("Fingerprint")) trust_ui.addText(" ".join(self.format_identity_key(device.identity_key))) trust_ui.addLabel(D_("Trust this device?")) - trust_ui.addBool(f"trust_{index}", value=C.boolConst(False)) + trust_ui.addBool(f"trust_{index}", value=C.bool_const(False)) trust_ui.addEmpty() trust_ui.addEmpty() - trust_ui_result = await xml_tools.deferXMLUI( + trust_ui_result = await xml_tools.defer_xmlui( sat, trust_ui, action_extra={ "meta_encryption_trust": namespace }, @@ -1347,7 +1347,7 @@ if C.bool(trust_ui_result.get("cancelled", "false")): raise omemo.TrustDecisionFailed("Trust UI cancelled.") - data_form_result = cast(Dict[str, str], xml_tools.XMLUIResult2DataFormResult( + data_form_result = cast(Dict[str, str], xml_tools.xmlui_result_2_data_form_result( trust_ui_result )) @@ -1375,7 +1375,7 @@ )) # Check whether ATM is enabled and handle everything in case it is - trust_system = cast(str, sat.memory.getParamA( + trust_system = cast(str, sat.memory.param_get_a( PARAM_NAME, PARAM_CATEGORY, profile_key=profile @@ -1434,7 +1434,7 @@ :meth:`~omemo.session_manager.SessionManager.create`. """ - client = sat.getClient(profile) + client = sat.get_client(profile) xep_0060 = cast(XEP_0060, sat.plugins["XEP-0060"]) storage = StorageImpl(profile) @@ -1549,7 +1549,7 @@ # Add configuration option to choose between manual trust and BTBV as the trust # model - sat.memory.updateParams(DEFAULT_TRUST_MODEL_PARAM) + sat.memory.update_params(DEFAULT_TRUST_MODEL_PARAM) # Plugins self.__xep_0045 = cast(Optional[XEP_0045], sat.plugins.get("XEP-0045")) @@ -1585,7 +1585,7 @@ priority=100050 ) sat.trigger.add( - "sendMessageData", + "send_message_data", self.__send_message_data_trigger, priority=100050 ) @@ -1596,18 +1596,18 @@ # including IQs. # Give twomemo a (slightly) higher priority than oldmemo - sat.registerEncryptionPlugin(self, "TWOMEMO", twomemo.twomemo.NAMESPACE, 101) - sat.registerEncryptionPlugin(self, "OLDMEMO", oldmemo.oldmemo.NAMESPACE, 100) + sat.register_encryption_plugin(self, "TWOMEMO", twomemo.twomemo.NAMESPACE, 101) + sat.register_encryption_plugin(self, "OLDMEMO", oldmemo.oldmemo.NAMESPACE, 100) xep_0163 = cast(XEP_0163, sat.plugins["XEP-0163"]) - xep_0163.addPEPEvent( + xep_0163.add_pep_event( "TWOMEMO_DEVICES", TWOMEMO_DEVICE_LIST_NODE, lambda items_event, profile: defer.ensureDeferred( self.__on_device_list_update(items_event, profile) ) ) - xep_0163.addPEPEvent( + xep_0163.add_pep_event( "OLDMEMO_DEVICES", OLDMEMO_DEVICE_LIST_NODE, lambda items_event, profile: defer.ensureDeferred( @@ -1620,9 +1620,9 @@ except KeyError: log.info(_("Text commands not available")) else: - self.__text_commands.registerTextCommands(self) - - def profileConnected( # pylint: disable=invalid-name + self.__text_commands.register_text_commands(self) + + def profile_connected( # pylint: disable=invalid-name self, client: SatXMPPClient ) -> None: @@ -1653,12 +1653,12 @@ """ twomemo_requested = \ - client.encryption.isEncryptionRequested(mess_data, twomemo.twomemo.NAMESPACE) + client.encryption.is_encryption_requested(mess_data, twomemo.twomemo.NAMESPACE) oldmemo_requested = \ - client.encryption.isEncryptionRequested(mess_data, oldmemo.oldmemo.NAMESPACE) + client.encryption.is_encryption_requested(mess_data, oldmemo.oldmemo.NAMESPACE) if not (twomemo_requested or oldmemo_requested): - self.__text_commands.feedBack( + self.__text_commands.feed_back( client, _("You need to have OMEMO encryption activated to reset the session"), mess_data @@ -1674,7 +1674,7 @@ log.debug(f"Replacing sessions with device {device}") await session_manager.replace_sessions(device) - self.__text_commands.feedBack( + self.__text_commands.feed_back( client, _("OMEMO session has been reset"), mess_data @@ -1682,7 +1682,7 @@ return False - async def getTrustUI( # pylint: disable=invalid-name + async def get_trust_ui( # pylint: disable=invalid-name self, client: SatXMPPClient, entity: jid.JID @@ -1698,7 +1698,7 @@ raise ValueError("A bare JID is expected.") bare_jids: Set[str] - if self.__xep_0045 is not None and self.__xep_0045.isJoinedRoom(client, entity): + if self.__xep_0045 is not None and self.__xep_0045.is_joined_room(client, entity): bare_jids = self.__get_joined_muc_users(client, self.__xep_0045, entity) else: bare_jids = { entity.userhost() } @@ -1729,7 +1729,7 @@ data_form_result = cast( Dict[str, str], - xml_tools.XMLUIResult2DataFormResult(data) + xml_tools.xmlui_result_2_data_form_result(data) ) trust_updates: Set[TrustUpdate] = set() @@ -1763,7 +1763,7 @@ )) # Check whether ATM is enabled and handle everything in case it is - trust_system = cast(str, self.__sat.memory.getParamA( + trust_system = cast(str, self.__sat.memory.param_get_a( PARAM_NAME, PARAM_CATEGORY, profile_key=profile @@ -1785,7 +1785,7 @@ return {} - submit_id = self.__sat.registerCallback(callback, with_data=True, one_shot=True) + submit_id = self.__sat.register_callback(callback, with_data=True, one_shot=True) result = xml_tools.XMLUI( panel_type=C.XMLUI_FORM, @@ -1809,7 +1809,7 @@ own_device, __ = await session_manager.get_own_device_information() - trust_ui.changeContainer("label") + trust_ui.change_container("label") trust_ui.addLabel(D_("This device ID")) trust_ui.addText(str(own_device.device_id)) trust_ui.addLabel(D_("This device's fingerprint")) @@ -1880,7 +1880,7 @@ bare_jids: Set[str] = set() try: - room = cast(muc.Room, xep_0045.getRoom(client, room_jid)) + room = cast(muc.Room, xep_0045.get_room(client, room_jid)) except exceptions.NotFound as e: raise exceptions.InternalError( "Participant list of unjoined MUC requested." @@ -2134,7 +2134,7 @@ room_jid = feedback_jid = sender_jid.userhostJID() try: - room = cast(muc.Room, self.__xep_0045.getRoom(client, room_jid)) + room = cast(muc.Room, self.__xep_0045.get_room(client, room_jid)) except exceptions.NotFound: log.warning( f"Ignoring MUC message from a room that has not been joined:" @@ -2165,7 +2165,7 @@ message_uid: Optional[str] = None if self.__xep_0359 is not None: - message_uid = self.__xep_0359.getOriginId(message_elt) + message_uid = self.__xep_0359.get_origin_id(message_elt) if message_uid is None: message_uid = message_elt.getAttribute("id") if message_uid is not None: @@ -2358,13 +2358,13 @@ await session_manager._evaluate_custom_trust_level(device_information) if trust_level is omemo.TrustLevel.TRUSTED: - post_treat.addCallback(client.encryption.markAsTrusted) + post_treat.addCallback(client.encryption.mark_as_trusted) else: - post_treat.addCallback(client.encryption.markAsUntrusted) + post_treat.addCallback(client.encryption.mark_as_untrusted) # Mark the message as originally encrypted post_treat.addCallback( - client.encryption.markAsEncrypted, + client.encryption.mark_as_encrypted, namespace=message.namespace ) @@ -2432,7 +2432,7 @@ # Add a store hint if this is a message stanza if stanza.name == "message": - self.__xep_0334.addHintElements(stanza, [ "store" ]) + self.__xep_0334.add_hint_elements(stanza, [ "store" ]) # Let the flow continue. return True @@ -2473,7 +2473,7 @@ ) # Add a store hint - self.__xep_0334.addHintElements(stanza, [ "store" ]) + self.__xep_0334.add_hint_elements(stanza, [ "store" ]) async def encrypt( self,