Commit c9e219c1 authored by mathieui's avatar mathieui

mypy: Reduce errors on muctab.py by a lot

parent 4fae01c0
......@@ -15,7 +15,17 @@ import shutil
import time
import uuid
from collections import defaultdict
from typing import Callable, Dict, List, Optional, Set, Tuple, Type
from typing import (
cast,
Callable,
Dict,
List,
Optional,
Set,
Tuple,
Type,
TypeVar,
)
from xml.etree import ElementTree as ET
from functools import partial
......@@ -65,6 +75,7 @@ from poezio.ui.types import Message, InfoMessage
log = logging.getLogger(__name__)
T = TypeVar('T', bound=tabs.Tab)
class Core:
"""
......@@ -99,8 +110,10 @@ class Core:
# that are displayed in almost all tabs, in an
# information window.
self.information_buffer = TextBuffer()
self.information_win_size = config.get(
'info_win_height', section='var')
self.information_win_size = cast(
int,
config.get('info_win_height', section='var'),
)
self.information_win = windows.TextWin(300)
self.information_buffer.add_window(self.information_win)
self.left_tab_win = None
......@@ -813,7 +826,7 @@ class Core:
####################### XMPP-related actions ##################################
def get_status(self) -> str:
def get_status(self) -> Status:
"""
Get the last status that was previously set
"""
......@@ -1016,7 +1029,7 @@ class Core:
### Tab getters ###
def get_tabs(self, cls: Type[tabs.Tab] = None) -> List[tabs.Tab]:
def get_tabs(self, cls: Type[T] = None) -> List[T]:
"Get all the tabs of a type"
if cls is None:
return self.tabs.get_tabs()
......@@ -1324,7 +1337,7 @@ class Core:
if tab.name.startswith(room_name):
tab.activate(reason=reason)
def on_user_changed_status_in_private(self, jid: JID, status: str) -> None:
def on_user_changed_status_in_private(self, jid: JID, status: Status) -> None:
tab = self.tabs.by_name_and_class(jid, tabs.ChatTab)
if tab is not None: # display the message in private
tab.update_status(status)
......@@ -1652,7 +1665,7 @@ class Core:
return
else:
scr = self.stdscr
tabs.Tab.resize(scr)
tabs.Tab.initial_resize(scr)
self.resize_global_info_bar()
self.resize_global_information_win()
for tab in self.tabs:
......@@ -2105,7 +2118,7 @@ class Core:
self.bookmarks.get_remote(self.xmpp, self.information,
_join_remote_only)
def room_error(self, error, room_name):
def room_error(self, error: IqError, room_name: str) -> None:
"""
Display the error in the tab
"""
......
"""
Module defining structures useful to the core class and related methods
"""
from dataclasses import dataclass
from typing import Any, Callable, List, Dict
__all__ = [
'ERROR_AND_STATUS_CODES', 'DEPRECATED_ERRORS', 'POSSIBLE_SHOW', 'Status',
......@@ -51,23 +53,11 @@ POSSIBLE_SHOW = {
}
@dataclass
class Status:
__slots__ = ('show', 'message')
def __init__(self, show, message):
self.show = show
self.message = message
class Command:
__slots__ = ('func', 'desc', 'comp', 'short_desc', 'usage')
def __init__(self, func, desc, comp, short_desc, usage):
self.func = func
self.desc = desc
self.comp = comp
self.short_desc = short_desc
self.usage = usage
show: str
message: str
class Completion:
......@@ -75,8 +65,13 @@ class Completion:
A completion result essentially currying the input completion call.
"""
__slots__ = ['func', 'args', 'kwargs', 'comp_list']
def __init__(self, func, comp_list, *args, **kwargs):
def __init__(
self,
func: Callable[..., Any],
comp_list: List[str],
*args: Any,
**kwargs: Any
) -> None:
self.func = func
self.comp_list = comp_list
self.args = args
......@@ -84,3 +79,12 @@ class Completion:
def run(self):
return self.func(self.comp_list, *self.args, **self.kwargs)
@dataclass
class Command:
__slots__ = ('func', 'desc', 'comp', 'short_desc', 'usage')
func: Callable[..., Any]
desc: str
comp: Callable[['windows.Input'], Completion]
short_desc: str
usage: str
......@@ -347,16 +347,16 @@ class Tabs:
if new_pos < len(self._tabs):
old_tab = self._tabs[old_pos]
self._tabs[new_pos], self._tabs[
old_pos] = old_tab, tabs.GapTab(self)
old_pos] = old_tab, tabs.GapTab(None)
else:
self._tabs.append(self._tabs[old_pos])
self._tabs[old_pos] = tabs.GapTab(self)
self._tabs[old_pos] = tabs.GapTab(None)
else:
if new_pos > old_pos:
self._tabs.insert(new_pos, tab)
self._tabs[old_pos] = tabs.GapTab(self)
self._tabs[old_pos] = tabs.GapTab(None)
elif new_pos < old_pos:
self._tabs[old_pos] = tabs.GapTab(self)
self._tabs[old_pos] = tabs.GapTab(None)
self._tabs.insert(new_pos, tab)
else:
return False
......
......@@ -5,7 +5,8 @@ upstream.
TODO: Check that they are fixed and remove those hacks
"""
from slixmpp.stanza import Message
from typing import Callable, Any
from slixmpp import Message, Iq, ClientXMPP
from slixmpp.xmlstream import ET
import logging
......@@ -25,7 +26,7 @@ def has_identity(xmpp, jid, identity, on_true=None, on_false=None):
xmpp.plugin['xep_0030'].get_info(jid=jid, callback=_cb)
def get_room_form(xmpp, room, callback):
def get_room_form(xmpp: ClientXMPP, room: str, callback: Callable[[Iq], Any]):
def _cb(result):
if result["type"] == "error":
return callback(None)
......
......@@ -104,7 +104,7 @@ def main():
logger.create_logger()
from poezio import roster
roster.create_roster()
roster.roster.reset()
from poezio.core.core import Core
......
......@@ -10,6 +10,8 @@ Defines the Roster and RosterGroup classes
import logging
log = logging.getLogger(__name__)
from typing import List
from poezio.config import config
from poezio.contact import Contact
from poezio.roster_sorting import SORTING_METHODS, GROUP_SORTING_METHODS
......@@ -18,6 +20,7 @@ from os import path as p
from datetime import datetime
from poezio.common import safeJID
from slixmpp.exceptions import IqError, IqTimeout
from slixmpp import JID
class Roster:
......@@ -29,6 +32,22 @@ class Roster:
DEFAULT_FILTER = (lambda x, y: None, None)
def __init__(self):
self.__node = None
# A tuple(function, *args) function to filter contacts
# on search, for example
self.contact_filter = self.DEFAULT_FILTER
self.groups = {}
self.contacts = {}
self.length = 0
self.connected = 0
self.folded_groups = []
# Used for caching roster infos
self.last_built = datetime.now()
self.last_modified = datetime.now()
def reset(self):
"""
node: the RosterSingle from slixmpp
"""
......@@ -143,7 +162,7 @@ class Roster:
"""Subscribe to a jid"""
self.__node.subscribe(jid)
def jids(self):
def jids(self) -> List[JID]:
"""List of the contact JIDS"""
l = []
for key in self.__node.keys():
......@@ -335,11 +354,6 @@ class RosterGroup:
return len([1 for contact in self.contacts if len(contact)])
def create_roster():
"Create the global roster object"
global roster
roster = Roster()
# Shared roster object
roster = None
roster = Roster()
......@@ -28,6 +28,7 @@ from typing import (
List,
Optional,
Union,
Tuple,
TYPE_CHECKING,
)
......@@ -52,6 +53,8 @@ from slixmpp import JID, InvalidJID, Message as SMessage
if TYPE_CHECKING:
from _curses import _CursesWindow # pylint: disable=E0611
from poezio.size_manager import SizeManager
from poezio.core.core import Core
log = logging.getLogger(__name__)
......@@ -117,7 +120,7 @@ class Tab:
height = 1
width = 1
def __init__(self, core):
def __init__(self, core: 'Core'):
self.core = core
self.nb = 0
if not hasattr(self, 'name'):
......@@ -133,7 +136,7 @@ class Tab:
self.commands = {} # and their own commands
@property
def size(self) -> int:
def size(self) -> 'SizeManager':
return self.core.size
@staticmethod
......@@ -196,7 +199,7 @@ class Tab:
self._state = 'normal'
@staticmethod
def resize(scr: '_CursesWindow'):
def initial_resize(scr: '_CursesWindow'):
Tab.height, Tab.width = scr.getmaxyx()
windows.base_wins.TAB_WIN = scr
......@@ -327,7 +330,7 @@ class Tab:
else:
return False
def refresh_tab_win(self):
def refresh_tab_win(self) -> None:
if config.get('enable_vertical_tab_list'):
left_tab_win = self.core.left_tab_win
if left_tab_win and not self.size.core_degrade_x:
......@@ -371,12 +374,12 @@ class Tab:
"""
pass
def update_commands(self):
def update_commands(self) -> None:
for c in self.plugin_commands:
if c not in self.commands:
self.commands[c] = self.plugin_commands[c]
def update_keys(self):
def update_keys(self) -> None:
for k in self.plugin_keys:
if k not in self.key_func:
self.key_func[k] = self.plugin_keys[k]
......@@ -435,7 +438,7 @@ class Tab:
"""
pass
def on_close(self):
def on_close(self) -> None:
"""
Called when the tab is to be closed
"""
......@@ -443,7 +446,7 @@ class Tab:
self.input.on_delete()
self.closed = True
def matching_names(self) -> List[str]:
def matching_names(self) -> List[Tuple[int, str]]:
"""
Returns a list of strings that are used to name a tab with the /win
command. For example you could switch to a tab that returns
......@@ -532,7 +535,7 @@ class ChatTab(Tab):
desc='Fix the last message with whatever you want.',
shortdesc='Correct the last message.',
completion=self.completion_correct)
self.chat_state = None
self.chat_state = None # type: Optional[str]
self.update_commands()
self.update_keys()
......@@ -667,11 +670,11 @@ class ChatTab(Tab):
self._text_buffer.messages = []
self.text_win.rebuild_everything(self._text_buffer)
def check_send_chat_state(self):
def check_send_chat_state(self) -> bool:
"If we should send a chat state"
return True
def send_chat_state(self, state, always_send=False):
def send_chat_state(self, state: str, always_send: bool = False) -> None:
"""
Send an empty chatstate message
"""
......@@ -691,9 +694,8 @@ class ChatTab(Tab):
x = ET.Element('{%s}x' % NS_MUC_USER)
msg.append(x)
msg.send()
return True
def send_composing_chat_state(self, empty_after):
def send_composing_chat_state(self, empty_after: bool) -> None:
"""
Send the "active" or "composing" chatstate, depending
on the the current status of the input
......@@ -729,7 +731,7 @@ class ChatTab(Tab):
self.core.add_timed_event(new_event)
self.timed_event_not_paused = new_event
def cancel_paused_delay(self):
def cancel_paused_delay(self) -> None:
"""
Remove that event from the list and set it to None.
Called for example when the input is emptied, or when the message
......@@ -741,7 +743,7 @@ class ChatTab(Tab):
self.core.remove_timed_event(self.timed_event_not_paused)
self.timed_event_not_paused = None
def set_last_sent_message(self, msg, correct=False):
def set_last_sent_message(self, msg: SMessage, correct: bool = False) -> None:
"""Ensure last_sent_message is set with the correct attributes"""
if correct:
# XXX: Is the copy needed. Is the object passed here reused
......@@ -751,7 +753,7 @@ class ChatTab(Tab):
self.last_sent_message = msg
@command_args_parser.raw
def command_correct(self, line):
def command_correct(self, line: str) -> None:
"""
/correct <fixed message>
"""
......@@ -777,7 +779,7 @@ class ChatTab(Tab):
return self.core.status.show in ('xa', 'away') or\
(hasattr(self, 'directed_presence') and not self.directed_presence)
def move_separator(self):
def move_separator(self) -> None:
self.text_win.remove_line_separator()
self.text_win.add_line_separator(self._text_buffer)
self.text_win.refresh()
......@@ -786,7 +788,7 @@ class ChatTab(Tab):
def get_conversation_messages(self):
return self._text_buffer.messages
def check_scrolled(self):
def check_scrolled(self) -> None:
if self.text_win.pos != 0:
self.state = 'scrolled'
......
......@@ -18,9 +18,21 @@ import re
import functools
from copy import copy
from datetime import datetime
from typing import Dict, Callable, List, Optional, Tuple, Union, Set
from typing import (
cast,
Any,
Dict,
Callable,
List,
Optional,
Tuple,
Union,
Set,
Pattern,
TYPE_CHECKING,
)
from slixmpp import InvalidJID, JID, Presence
from slixmpp import InvalidJID, JID, Presence, Iq
from slixmpp.exceptions import IqError, IqTimeout
from poezio.tabs import ChatTab, Tab, SHOW_NAME
......@@ -49,6 +61,10 @@ from poezio.ui.types import (
StatusMessage,
)
if TYPE_CHECKING:
from poezio.core.core import Core
from slixmpp.plugins.xep_0004 import Form
log = logging.getLogger(__name__)
NS_MUC_USER = 'http://jabber.org/protocol/muc#user'
......@@ -64,11 +80,11 @@ class MucTab(ChatTab):
"""
message_type = 'groupchat'
plugin_commands = {} # type: Dict[str, Command]
plugin_keys = {} # type: Dict[str, Callable]
plugin_keys = {} # type: Dict[str, Callable[..., Any]]
additional_information = {} # type: Dict[str, Callable[[str], str]]
lagged = False
def __init__(self, core, jid, nick, password=None):
def __init__(self, core: 'Core', jid: JID, nick: str, password: Optional[str] = None) -> None:
ChatTab.__init__(self, core, jid)
self.joined = False
self._state = 'disconnected'
......@@ -78,7 +94,7 @@ class MucTab(ChatTab):
self.own_user = None # type: Optional[User]
self.password = password
# buffered presences
self.presence_buffer = []
self.presence_buffer = [] # type: List[Presence]
# userlist
self.users = [] # type: List[User]
# private conversations
......@@ -88,13 +104,13 @@ class MucTab(ChatTab):
self.topic = ''
self.topic_from = ''
# Self ping event, so we can cancel it when we leave the room
self.self_ping_event = None
self.self_ping_event = None # type: Optional[timed_events.DelayedEvent]
# UI stuff
self.topic_win = windows.Topic()
self.v_separator = windows.VerticalSeparator()
self.user_win = windows.UserList()
self.info_header = windows.MucInfoWin()
self.input = windows.MessageInput()
self.input = windows.MessageInput() # type: windows.MessageInput
# List of ignored users
self.ignores = [] # type: List[User]
# keys
......@@ -106,7 +122,7 @@ class MucTab(ChatTab):
self.resize()
@property
def general_jid(self):
def general_jid(self) -> JID:
return self.jid
def check_send_chat_state(self) -> bool:
......@@ -136,21 +152,21 @@ class MucTab(ChatTab):
"""
del MucTab.additional_information[plugin_name]
def cancel_config(self, form):
def cancel_config(self, form: 'Form') -> None:
"""
The user do not want to send their config, send an iq cancel
"""
muc.cancel_config(self.core.xmpp, self.jid.bare)
self.core.close_tab()
def send_config(self, form):
def send_config(self, form: 'Form') -> None:
"""
The user sends their config to the server
"""
muc.configure_room(self.core.xmpp, self.jid.bare, form)
self.core.close_tab()
def join(self):
def join(self) -> None:
"""
Join the room
"""
......@@ -167,12 +183,12 @@ class MucTab(ChatTab):
self.core,
self.jid.bare,
self.own_nick,
self.password,
self.password or '',
status=status.message,
show=status.show,
seconds=seconds)
def leave_room(self, message: str):
def leave_room(self, message: str) -> None:
if self.joined:
theme = get_theme()
info_col = dump_tuple(theme.COLOR_INFORMATION_TEXT)
......@@ -216,15 +232,17 @@ class MucTab(ChatTab):
muc.leave_groupchat(self.core.xmpp, self.jid.bare, self.own_nick,
message)
def change_affiliation(self,
def change_affiliation(
self,
nick_or_jid: Union[str, JID],
affiliation: str,
reason=''):
reason: str = ''
) -> None:
"""
Change the affiliation of a nick or JID
"""
def callback(iq):
def callback(iq: Iq) -> None:
if iq['type'] == 'error':
self.core.information(
"Could not set affiliation '%s' for '%s'." %
......@@ -235,9 +253,10 @@ class MucTab(ChatTab):
valid_affiliations = ('outcast', 'none', 'member', 'admin', 'owner')
if affiliation not in valid_affiliations:
return self.core.information(
self.core.information(
'The affiliation must be one of ' +
', '.join(valid_affiliations), 'Error')
return
if nick_or_jid in [user.nick for user in self.users]:
muc.set_user_affiliation(
self.core.xmpp,
......@@ -255,12 +274,12 @@ class MucTab(ChatTab):
callback=callback,
reason=reason)
def change_role(self, nick: str, role: str, reason=''):
def change_role(self, nick: str, role: str, reason: str = '') -> None:
"""
Change the role of a nick
"""
def callback(iq):
def callback(iq: Iq) -> None:
if iq['type'] == 'error':
self.core.information(
"Could not set role '%s' for '%s'." % (role, nick),
......@@ -269,14 +288,16 @@ class MucTab(ChatTab):
valid_roles = ('none', 'visitor', 'participant', 'moderator')
if not self.joined or role not in valid_roles:
return self.core.information(
self.core.information(
'The role must be one of ' + ', '.join(valid_roles), 'Error')
return
try:
target_jid = copy(self.jid)
target_jid.resource = nick
except InvalidJID:
return self.core.information('Invalid nick', 'Info')
self.core.information('Invalid nick', 'Info')
return
muc.set_user_role(
self.core.xmpp, self.jid.bare, nick, reason, role, callback=callback)
......@@ -313,12 +334,12 @@ class MucTab(ChatTab):
self.add_message(InfoMessage(info), typ=0)
return True
def change_topic(self, topic: str):
def change_topic(self, topic: str) -> None:
"""Change the current topic"""
muc.change_subject(self.core.xmpp, self.jid.bare, topic)
@refresh_wrapper.always
def show_topic(self):
def show_topic(self) -> None:
"""
Print the current topic
"""
......@@ -345,7 +366,7 @@ class MucTab(ChatTab):
)
@refresh_wrapper.always
def recolor(self, random_colors=False):
def recolor(self, random_colors: bool = False) -> None:
"""Recolor the current MUC users"""
deterministic = config.get_by_tabname('deterministic_nick_colors',
self.jid.bare)
......@@ -410,7 +431,7 @@ class MucTab(ChatTab):
self.text_win.rebuild_everything(self._text_buffer)
return True
def on_input(self, key, raw):
def on_input(self, key: str, raw: bool) -> bool:
if not raw and key in self.key_func:
self.key_func[key]()
return False
......@@ -424,17 +445,17 @@ class MucTab(ChatTab):
def get_nick(self) -> str:
if config.get('show_muc_jid'):
return self.jid.bare
return cast(str, self.jid.bare)
bookmark = self.core.bookmarks[self.jid.bare]
if bookmark is not None and bookmark.name:
return bookmark.name
# TODO: send the disco#info identity name here, if it exists.
return self.jid.user
def get_text_window(self):
def get_text_window(self) -> windows.TextWin:
return self.text_win
def on_lose_focus(self):<