plugin XEP-0163: allow unset event_type in addPEPEvent, check conflict + type hints

sat/plugins/plugin_xep_0163.py

@@ -17,18 +17,21 @@
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+from typing import Optional, Callable
 from sat.core.i18n import _
 from sat.core import exceptions
 from sat.core.constants import Const as C
 from sat.core.log import getLogger
 
-log = getLogger(__name__)
 from twisted.words.xish import domish
 
 from wokkel import disco, pubsub
 from wokkel.formats import Mood
 from sat.tools.common import data_format
 
+
+log = getLogger(__name__)
+
 NS_USER_MOOD = "http://jabber.org/protocol/mood"
 
 PLUGIN_INFO = {
@@ -71,21 +74,34 @@ class XEP_0163(object):
         disco_info.extend(list(map(disco.DiscoFeature, self.pep_events)))
         return True
 
-    def addPEPEvent(self, event_type, node, in_callback, out_callback=None, notify=True):
+    def addPEPEvent(
+        self,
+        event_type: Optional[str],
+        node: str,
+        in_callback: Callable,
+        out_callback: Optional[Callable] = None,
+        notify: bool = True
+    ) -> None:
         """Add a Personal Eventing Protocol event manager
 
-        @param event_type(unicode): type of the event (always uppercase),
-            can be MOOD, TUNE, etc
-        @param node(unicode): namespace of the node (e.g. http://jabber.org/protocol/mood
+        @param event_type: type of the event (stored uppercase),
+            only used when out_callback is set.
+            Can be MOOD, TUNE, etc.
+        @param node: namespace of the node (e.g. http://jabber.org/protocol/mood
             for User Mood)
-        @param in_callback(callable): method to call when this event occur
+        @param in_callback: method to call when this event occur
             the callable will be called with (itemsEvent, profile) as arguments
-        @param out_callback(callable,None): method to call when we want to publish this
+        @param out_callback: method to call when we want to publish this
             event (must return a deferred)
             the callable will be called when sendPEPEvent is called
-        @param notify(bool): add autosubscribe (+notify) if True
+        @param notify: add autosubscribe (+notify) if True
         """
-        if out_callback:
+        if event_type and out_callback:
+            event_type = event_type.upper()
+            if event_type in self.pep_out_cb:
+                raise exceptions.ConflictError(
+                    f"event_type {event_type!r} already exists"
+                )
             self.pep_out_cb[event_type] = out_callback
         self.pep_events.add(node)
         if notify: