pep/poezio
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

mypy: Make multiuserchat.py pass --strict

Browse source
This commit is contained in:
mathieui 2020-05-28 18:49:22 +02:00
parent 6425daae1e
commit 927f3843ad
1 changed files with 82 additions and 63 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

143
poezio/multiuserchat.py
View file

@ -11,18 +11,39 @@ slix plugin
"""
from xml.etree import ElementTree as ET
from typing import (
Callable,
Optional,
TYPE_CHECKING,
)
from poezio.common import safeJID
from slixmpp import JID
from slixmpp.exceptions import IqError, IqTimeout
from slixmpp import (
JID,
ClientXMPP,
Iq,
)
import logging
log = logging.getLogger(__name__)
if TYPE_CHECKING:
from poezio.core import Core
from poezio.tabs import Tab
from slixmpp.plugins.xep_0004 import Form
NS_MUC_ADMIN = 'http://jabber.org/protocol/muc#admin'
NS_MUC_OWNER = 'http://jabber.org/protocol/muc#owner'
def destroy_room(xmpp, room, reason='', altroom=''):
def destroy_room(
xmpp: ClientXMPP,
room: str,
reason: str = '',
altroom: str = ''
) -> bool:
"""
destroy a room
"""
@ -42,7 +63,7 @@ def destroy_room(xmpp, room, reason='', altroom=''):
query.append(destroy)
iq.append(query)
def callback(iq):
def callback(iq: Iq) -> None:
if not iq or iq['type'] == 'error':
xmpp.core.information('Unable to destroy room %s' % room, 'Info')
else:
@ -52,23 +73,13 @@ def destroy_room(xmpp, room, reason='', altroom=''):
return True
def send_private_message(xmpp, jid, line):
"""
Send a private message
"""
jid = safeJID(jid)
xmpp.send_message(mto=jid, mbody=line, mtype='chat')
def send_groupchat_message(xmpp, jid, line):
"""
Send a message to the groupchat
"""
jid = safeJID(jid)
xmpp.send_message(mto=jid, mbody=line, mtype='groupchat')
def change_show(xmpp, jid: JID, own_nick: str, show, status):
def change_show(
xmpp: ClientXMPP,
jid: JID,
own_nick: str,
show: str,
status: Optional[str]
) -> None:
"""
Change our 'Show'
"""
@ -81,7 +92,7 @@ def change_show(xmpp, jid: JID, own_nick: str, show, status):
pres.send()
def change_subject(xmpp, jid, subject):
def change_subject(xmpp: ClientXMPP, jid: JID, subject: str) -> None:
"""
Change the room subject
"""
@ -92,7 +103,13 @@ def change_subject(xmpp, jid, subject):
msg.send()
def change_nick(core, jid, nick, status=None, show=None):
def change_nick(
core: 'Core',
jid: JID,
nick: str,
status: Optional[str] = None,
show: Optional[str] = None
) -> None:
"""
Change our own nick in a room
"""
@ -103,14 +120,16 @@ def change_nick(core, jid, nick, status=None, show=None):
presence.send()
def join_groupchat(core,
jid,
nick,
passwd='',
status=None,
show=None,
seconds=None,
tab=None):
def join_groupchat(
core: 'Core',
jid: JID,
nick: str,
passwd: str = '',
status: Optional[str] = None,
show: Optional[str] = None,
seconds: Optional[int] = None,
tab: Optional['Tab'] = None
) -> None:
xmpp = core.xmpp
stanza = xmpp.make_presence(
pto='%s/%s' % (jid, nick), pstatus=status, pshow=show)
@ -119,8 +138,10 @@ def join_groupchat(core,
passelement = ET.Element('password')
passelement.text = passwd
x.append(passelement)
def on_disco(iq):
if 'urn:xmpp:mam:2' in iq['disco_info'].get_features() or (tab and tab._text_buffer.last_message):
def on_disco(iq: Iq) -> None:
if ('urn:xmpp:mam:2' in iq['disco_info'].get_features()
or (tab and tab._text_buffer.last_message)):
history = ET.Element('{http://jabber.org/protocol/muc}history')
history.attrib['seconds'] = str(0)
x.append(history)
@ -136,13 +157,15 @@ def join_groupchat(core,
xmpp.plugin['xep_0045'].rooms[jid] = {}
xmpp.plugin['xep_0045'].our_nicks[jid] = to.resource
try:
xmpp.plugin['xep_0030'].get_info(jid=jid, callback=on_disco)
except (IqError, IqTimeout):
return core.information('Failed to retrieve messages', 'Error')
def leave_groupchat(xmpp, jid, own_nick, msg):
def leave_groupchat(
xmpp: ClientXMPP,
jid: JID,
own_nick: str,
msg: str
) -> None:
"""
Leave the groupchat
"""
@ -156,7 +179,14 @@ def leave_groupchat(xmpp, jid, own_nick, msg):
exc_info=True)
def set_user_role(xmpp, jid, nick, reason, role, callback=None):
def set_user_role(
xmpp: ClientXMPP,
jid: JID,
nick: str,
reason: str,
role: str,
callback: Callable[[Iq], None]
) -> None:
"""
(try to) Set the role of a MUC user
(role = 'none': eject user)
@ -172,21 +202,18 @@ def set_user_role(xmpp, jid, nick, reason, role, callback=None):
query.append(item)
iq.append(query)
iq['to'] = jid
if callback:
return iq.send(callback=callback)
try:
return iq.send()
except (IqError, IqTimeout) as e:
return e.iq
iq.send(callback=callback)
def set_user_affiliation(xmpp,
muc_jid,
affiliation,
nick=None,
jid=None,
reason=None,
callback=None):
def set_user_affiliation(
xmpp: ClientXMPP,
muc_jid: JID,
affiliation: str,
callback: Callable[[Iq], None],
nick: Optional[str] = None,
jid: Optional[JID] = None,
reason: Optional[str] = None
) -> None:
"""
(try to) Set the affiliation of a MUC user
"""
@ -212,18 +239,10 @@ def set_user_affiliation(xmpp,
query.append(item)
iq = xmpp.make_iq_set(query)
iq['to'] = muc_jid
if callback:
return iq.send(callback=callback)
try:
return xmpp.plugin['xep_0045'].set_affiliation(
str(muc_jid),
str(jid) if jid else None, nick, affiliation)
except:
log.debug('Error setting the affiliation: %s', exc_info=True)
return False
iq.send(callback=callback)
def cancel_config(xmpp, room):
def cancel_config(xmpp: ClientXMPP, room: str) -> None:
query = ET.Element('{http://jabber.org/protocol/muc#owner}query')
x = ET.Element('{jabber:x:data}x', type='cancel')
query.append(x)
@ -232,7 +251,7 @@ def cancel_config(xmpp, room):
iq.send()
def configure_room(xmpp, room, form):
def configure_room(xmpp: ClientXMPP, room: str, form: 'Form') -> None:
if form is None:
return
iq = xmpp.make_iq_set()