diff options
author | mathieui <mathieui@mathieui.net> | 2020-06-01 09:09:36 +0200 |
---|---|---|
committer | mathieui <mathieui@mathieui.net> | 2020-06-01 09:09:36 +0200 |
commit | 291233bbbd81fcb1fddf6e4d1cd93478f73c1ade (patch) | |
tree | 70a184a22724be16ce7a7ef930d07e9eda203ade | |
parent | e662bdee055d25cb609b5ed229e077e3d557bbe0 (diff) | |
parent | c9e219c1405604f2ef81b3e446cc33a0492af42d (diff) | |
download | poezio-291233bbbd81fcb1fddf6e4d1cd93478f73c1ade.tar.gz poezio-291233bbbd81fcb1fddf6e4d1cd93478f73c1ade.tar.bz2 poezio-291233bbbd81fcb1fddf6e4d1cd93478f73c1ade.tar.xz poezio-291233bbbd81fcb1fddf6e4d1cd93478f73c1ade.zip |
Merge branch 'more-typing' into 'master'
More typing
See merge request poezio/poezio!135
-rw-r--r-- | poezio/common.py | 22 | ||||
-rw-r--r-- | poezio/core/core.py | 29 | ||||
-rw-r--r-- | poezio/core/structs.py | 38 | ||||
-rw-r--r-- | poezio/core/tabs.py | 8 | ||||
-rw-r--r-- | poezio/decorators.py | 63 | ||||
-rw-r--r-- | poezio/fixes.py | 5 | ||||
-rw-r--r-- | poezio/multiuserchat.py | 149 | ||||
-rw-r--r-- | poezio/poezio.py | 2 | ||||
-rw-r--r-- | poezio/poezio_shlex.pyi | 45 | ||||
-rw-r--r-- | poezio/roster.py | 28 | ||||
-rw-r--r-- | poezio/tabs/basetabs.py | 38 | ||||
-rw-r--r-- | poezio/tabs/muctab.py | 358 | ||||
-rw-r--r-- | poezio/tabs/privatetab.py | 2 | ||||
-rw-r--r-- | poezio/text_buffer.py | 3 | ||||
-rw-r--r-- | poezio/user.py | 2 | ||||
-rw-r--r-- | poezio/windows/info_wins.py | 17 |
16 files changed, 491 insertions, 318 deletions
diff --git a/poezio/common.py b/poezio/common.py index 98870dda..315d5b9e 100644 --- a/poezio/common.py +++ b/poezio/common.py @@ -14,7 +14,7 @@ from datetime import ( timezone, ) from pathlib import Path -from typing import Dict, List, Optional, Tuple, Union +from typing import Dict, List, Optional, Tuple, Union, Any import os import subprocess @@ -44,7 +44,7 @@ def _get_output_of_command(command: str) -> Optional[List[str]]: return None -def _is_in_path(command: str, return_abs_path=False) -> Union[bool, str]: +def _is_in_path(command: str, return_abs_path: bool = False) -> Union[bool, str]: """ Check if *command* is in the $PATH or not. @@ -111,10 +111,12 @@ def get_os_info() -> str: stdout=subprocess.PIPE, close_fds=True) process.wait() - output = process.stdout.readline().decode('utf-8').strip() - # some distros put n/a in places, so remove those - output = output.replace('n/a', '').replace('N/A', '') - return output + if process.stdout is not None: + out = process.stdout.readline().decode('utf-8').strip() + # some distros put n/a in places, so remove those + out = out.replace('n/a', '').replace('N/A', '') + return out + return '' # lsb_release executable not available, so parse files for distro_name in DISTRO_INFO: @@ -287,7 +289,7 @@ def shell_split(st: str) -> List[str]: return ret -def find_argument(pos: int, text: str, quoted=True) -> int: +def find_argument(pos: int, text: str, quoted: bool = True) -> int: """ Split an input into a list of arguments, return the number of the argument selected by pos. @@ -342,7 +344,7 @@ def _find_argument_unquoted(pos: int, text: str) -> int: return argnum + 1 -def parse_str_to_secs(duration='') -> int: +def parse_str_to_secs(duration: str = '') -> int: """ Parse a string of with a number of d, h, m, s. @@ -370,7 +372,7 @@ def parse_str_to_secs(duration='') -> int: return result -def parse_secs_to_str(duration=0) -> str: +def parse_secs_to_str(duration: int = 0) -> str: """ Do the reverse operation of :py:func:`parse_str_to_secs`. @@ -457,7 +459,7 @@ def format_gaming_string(infos: Dict[str, str]) -> str: return name -def safeJID(*args, **kwargs) -> JID: +def safeJID(*args: Any, **kwargs: Any) -> JID: """ Construct a :py:class:`slixmpp.JID` object from a string. diff --git a/poezio/core/core.py b/poezio/core/core.py index 3ad15719..3a13d4c3 100644 --- a/poezio/core/core.py +++ b/poezio/core/core.py @@ -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 """ diff --git a/poezio/core/structs.py b/poezio/core/structs.py index 72c9628a..a75f1e94 100644 --- a/poezio/core/structs.py +++ b/poezio/core/structs.py @@ -1,6 +1,8 @@ """ 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 diff --git a/poezio/core/tabs.py b/poezio/core/tabs.py index d5909d39..61bad6f2 100644 --- a/poezio/core/tabs.py +++ b/poezio/core/tabs.py @@ -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 diff --git a/poezio/decorators.py b/poezio/decorators.py index 51abf32c..62724ecd 100644 --- a/poezio/decorators.py +++ b/poezio/decorators.py @@ -1,54 +1,68 @@ """ Module containing various decorators """ -from typing import Any, Callable, List, Optional +from typing import ( + cast, + Any, + Callable, + List, + Optional, + TypeVar, + TYPE_CHECKING, +) from poezio import common +if TYPE_CHECKING: + from poezio.tabs import RosterInfoTab + +T = TypeVar('T', bound=Callable[..., Any]) + + class RefreshWrapper: - def __init__(self): + def __init__(self) -> None: self.core = None - def conditional(self, func: Callable) -> Callable: + def conditional(self, func: T) -> T: """ Decorator to refresh the UI if the wrapped function returns True """ - def wrap(*args, **kwargs): + def wrap(*args: Any, **kwargs: Any) -> Any: ret = func(*args, **kwargs) if self.core and ret: self.core.refresh_window() return ret - return wrap + return cast(T, wrap) - def always(self, func: Callable) -> Callable: + def always(self, func: T) -> T: """ Decorator that refreshs the UI no matter what after the function """ - def wrap(*args, **kwargs): + def wrap(*args: Any, **kwargs: Any) -> Any: ret = func(*args, **kwargs) if self.core: self.core.refresh_window() return ret - return wrap + return cast(T, wrap) - def update(self, func: Callable) -> Callable: + def update(self, func: T) -> T: """ Decorator that only updates the screen """ - def wrap(*args, **kwargs): + def wrap(*args: Any, **kwargs: Any) -> Any: ret = func(*args, **kwargs) if self.core: self.core.doupdate() return ret - return wrap + return cast(T, wrap) refresh_wrapper = RefreshWrapper() @@ -61,32 +75,32 @@ class CommandArgParser: """ @staticmethod - def raw(func: Callable) -> Callable: + def raw(func: T) -> T: """Just call the function with a single string, which is the original string untouched """ - def wrap(self, args, *a, **kw): + def wrap(self: Any, args: Any, *a: Any, **kw: Any) -> Any: return func(self, args, *a, **kw) - return wrap + return cast(T, wrap) @staticmethod - def ignored(func: Callable) -> Callable: + def ignored(func: T) -> T: """ Call the function without any argument """ - def wrap(self, args=None, *a, **kw): + def wrap(self: Any, args: Any = None, *a: Any, **kw: Any) -> Any: return func(self, *a, **kw) - return wrap + return cast(T, wrap) @staticmethod def quoted(mandatory: int, - optional=0, + optional: int = 0, defaults: Optional[List[Any]] = None, - ignore_trailing_arguments=False): + ignore_trailing_arguments: bool = False) -> Callable[[T], T]: """The function receives a list with a number of arguments that is between the numbers `mandatory` and `optional`. @@ -131,8 +145,8 @@ class CommandArgParser: """ default_args_outer = defaults or [] - def first(func: Callable): - def second(self, args: str, *a, **kw): + def first(func: T) -> T: + def second(self: Any, args: str, *a: Any, **kw: Any) -> Any: default_args = default_args_outer if args and args.strip(): split_args = common.shell_split(args) @@ -156,8 +170,7 @@ class CommandArgParser: res[-1] += " " + " ".join(split_args) return func(self, res, *a, **kw) - return second - + return cast(T, second) return first @@ -166,11 +179,11 @@ command_args_parser = CommandArgParser() def deny_anonymous(func: Callable) -> Callable: """Decorator to disable commands when using an anonymous account.""" - def wrap(self: 'RosterInfoTab', *args, **kwargs): + def wrap(self: 'RosterInfoTab', *args: Any, **kwargs: Any) -> Any: if self.core.xmpp.anon: return self.core.information( 'This command is not available for anonymous accounts.', 'Info' ) return func(self, *args, **kwargs) - return wrap + return cast(T, wrap) diff --git a/poezio/fixes.py b/poezio/fixes.py index f8de7b14..a9e15dee 100644 --- a/poezio/fixes.py +++ b/poezio/fixes.py @@ -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) diff --git a/poezio/multiuserchat.py b/poezio/multiuserchat.py index 30c36a77..12f97661 100644 --- a/poezio/multiuserchat.py +++ b/poezio/multiuserchat.py @@ -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') + xmpp.plugin['xep_0030'].get_info(jid=jid, callback=on_disco) -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 - - -def set_user_affiliation(xmpp, - muc_jid, - affiliation, - nick=None, - jid=None, - reason=None, - callback=None): + iq.send(callback=callback) + + +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() diff --git a/poezio/poezio.py b/poezio/poezio.py index e38871c6..da1bc3e7 100644 --- a/poezio/poezio.py +++ b/poezio/poezio.py @@ -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 diff --git a/poezio/poezio_shlex.pyi b/poezio/poezio_shlex.pyi new file mode 100644 index 00000000..affbe12b --- /dev/null +++ b/poezio/poezio_shlex.pyi @@ -0,0 +1,45 @@ +from typing import List, Tuple, Any, TextIO, Union, Optional, Iterable, TypeVar +import sys + +def split(s: str, comments: bool = ..., posix: bool = ...) -> List[str]: ... +if sys.version_info >= (3, 8): + def join(split_command: Iterable[str]) -> str: ... +def quote(s: str) -> str: ... + +_SLT = TypeVar('_SLT', bound=shlex) + +class shlex(Iterable[str]): + commenters: str + wordchars: str + whitespace: str + escape: str + quotes: str + escapedquotes: str + whitespace_split: bool + infile: str + instream: TextIO + source: str + debug: int + lineno: int + token: str + eof: str + if sys.version_info >= (3, 6): + punctuation_chars: str + + if sys.version_info >= (3, 6): + def __init__(self, instream: Union[str, TextIO] = ..., infile: Optional[str] = ..., + posix: bool = ..., punctuation_chars: Union[bool, str] = ...) -> None: ... + else: + def __init__(self, instream: Union[str, TextIO] = ..., infile: Optional[str] = ..., + posix: bool = ...) -> None: ... + def get_token(self) -> Tuple[int, int, str]: ... + def push_token(self, tok: str) -> None: ... + def read_token(self) -> str: ... + def sourcehook(self, filename: str) -> Tuple[str, TextIO]: ... + # TODO argument types + def push_source(self, newstream: Any, newfile: Any = ...) -> None: ... + def pop_source(self) -> None: ... + def error_leader(self, infile: str = ..., + lineno: int = ...) -> None: ... + def __iter__(self: _SLT) -> _SLT: ... + def __next__(self) -> str: ... diff --git a/poezio/roster.py b/poezio/roster.py index bedf477b..4a6350a9 100644 --- a/poezio/roster.py +++ b/poezio/roster.py @@ -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() diff --git a/poezio/tabs/basetabs.py b/poezio/tabs/basetabs.py index a02744aa..d822ea94 100644 --- a/poezio/tabs/basetabs.py +++ b/poezio/tabs/basetabs.py @@ -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' diff --git a/poezio/tabs/muctab.py b/poezio/tabs/muctab.py index 63bf026e..b9c8dad7 100644 --- a/poezio/tabs/muctab.py +++ b/poezio/tabs/muctab.py @@ -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, - nick_or_jid: Union[str, JID], - affiliation: str, - reason=''): + def change_affiliation( + self, + nick_or_jid: Union[str, JID], + affiliation: str, + 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): + def on_lose_focus(self) -> None: if self.joined: if self.input.text: self.state = 'nonempty' @@ -450,7 +471,7 @@ class MucTab(ChatTab): self.send_chat_state('inactive') self.check_scrolled() - def on_gain_focus(self): + def on_gain_focus(self) -> None: self.state = 'current' if (self.text_win.built_lines and self.text_win.built_lines[-1] is None and not config.get('show_useless_separator')): @@ -461,10 +482,8 @@ class MucTab(ChatTab): self.general_jid) and not self.input.get_text(): self.send_chat_state('active') - def handle_presence(self, presence): - """ - Handle MUC presence - """ + def handle_presence(self, presence: Presence) -> None: + """Handle MUC presence""" self.reset_lag() status_codes = set() for status_code in presence.xml.findall(STATUS_XPATH): @@ -492,7 +511,7 @@ class MucTab(ChatTab): self.input.refresh() self.core.doupdate() - def process_presence_buffer(self, last_presence, own): + def process_presence_buffer(self, last_presence: Presence, own: bool) -> None: """ Batch-process all the initial presences """ @@ -516,7 +535,7 @@ class MucTab(ChatTab): self.core.tabs.current_tab.refresh_input() self.core.doupdate() - def handle_presence_unjoined(self, presence: Presence, deterministic, own=False) -> None: + def handle_presence_unjoined(self, presence: Presence, deterministic: bool, own: bool = False) -> None: """ Presence received while we are not in the room (before code=110) """ @@ -538,7 +557,7 @@ class MucTab(ChatTab): status_codes.add(status_code.attrib['code']) self.own_join(from_nick, new_user, status_codes) - def own_join(self, from_nick: str, new_user: User, status_codes: Set[str]): + def own_join(self, from_nick: str, new_user: User, status_codes: Set[str]) -> None: """ Handle the last presence we received, entering the room """ @@ -603,7 +622,7 @@ class MucTab(ChatTab): typ=0) mam.schedule_tab_open(self) - def handle_presence_joined(self, presence: Presence, status_codes) -> None: + def handle_presence_joined(self, presence: Presence, status_codes: Set[str]) -> None: """ Handle new presences when we are already in the room """ @@ -630,7 +649,7 @@ class MucTab(ChatTab): return elif change_nick: self.core.events.trigger('muc_nickchange', presence, self) - self.on_user_nick_change(presence, user, from_nick, from_room) + self.on_user_nick_change(presence, user, from_nick) elif ban: self.core.events.trigger('muc_ban', presence, self) self.core.on_user_left_private_conversation( @@ -656,7 +675,7 @@ class MucTab(ChatTab): self.on_user_change_status(user, from_nick, from_room, affiliation, role, show, status) - def on_non_member_kicked(self): + def on_non_member_kicked(self) -> None: """We have been kicked because the MUC is members-only""" self.add_message( MucOwnLeaveMessage( @@ -666,7 +685,7 @@ class MucTab(ChatTab): typ=2) self.disconnect() - def on_muc_shutdown(self): + def on_muc_shutdown(self) -> None: """We have been kicked because the MUC service is shutting down""" self.add_message( MucOwnLeaveMessage( @@ -676,8 +695,8 @@ class MucTab(ChatTab): typ=2) self.disconnect() - def on_user_join(self, from_nick, affiliation, show, status, role, jid, - color): + def on_user_join(self, from_nick: str, affiliation: str, show: str, status: str, role: str, jid: JID, + color: str) -> None: """ When a new user joins the groupchat """ @@ -693,7 +712,7 @@ class MucTab(ChatTab): self.general_jid): color = dump_tuple(user.color) else: - color = 3 + color = "3" theme = get_theme() info_col = dump_tuple(theme.COLOR_INFORMATION_TEXT) spec_col = dump_tuple(theme.COLOR_JOIN_CHAR) @@ -722,7 +741,7 @@ class MucTab(ChatTab): self.add_message(InfoMessage(msg), typ=2) self.core.on_user_rejoined_private_conversation(self.jid.bare, from_nick) - def on_user_nick_change(self, presence, user, from_nick, from_room): + def on_user_nick_change(self, presence: Presence, user: User, from_nick: str) -> None: new_nick = presence.xml.find( '{%s}x/{%s}item' % (NS_MUC_USER, NS_MUC_USER)).attrib['nick'] old_color = user.color @@ -746,7 +765,7 @@ class MucTab(ChatTab): color = dump_tuple(user.color) old_color = dump_tuple(old_color) else: - old_color = color = 3 + old_color = color = "3" info_col = dump_tuple(get_theme().COLOR_INFORMATION_TEXT) self.add_message( InfoMessage( @@ -763,7 +782,7 @@ class MucTab(ChatTab): # rename the private tabs if needed self.core.rename_private_tabs(self.jid.bare, from_nick, user) - def on_user_banned(self, presence, user, from_nick): + def on_user_banned(self, presence: Presence, user: User, from_nick: str) -> None: """ When someone is banned from a muc """ @@ -818,7 +837,7 @@ class MucTab(ChatTab): self.general_jid): color = dump_tuple(user.color) else: - color = 3 + color = "3" if by: kick_msg = ('\x191}%(spec)s \x19%(color)s}' @@ -846,7 +865,7 @@ class MucTab(ChatTab): } self.add_message(cls(kick_msg), typ=2) - def on_user_kicked(self, presence, user, from_nick): + def on_user_kicked(self, presence: Presence, user: User, from_nick: str) -> None: """ When someone is kicked from a muc """ @@ -899,7 +918,7 @@ class MucTab(ChatTab): self.general_jid): color = dump_tuple(user.color) else: - color = 3 + color = "3" if by: kick_msg = ('\x191}%(spec)s \x19%(color)s}%(nick)s' '\x19%(info_col)s} has been kicked by ' @@ -932,7 +951,7 @@ class MucTab(ChatTab): status: str, from_nick: str, from_room: JID, - server_initiated=False): + server_initiated: bool = False) -> None: """ When a user leaves a groupchat """ @@ -952,7 +971,7 @@ class MucTab(ChatTab): self.general_jid): color = dump_tuple(user.color) else: - color = 3 + color = "3" theme = get_theme() info_col = dump_tuple(theme.COLOR_INFORMATION_TEXT) spec_col = dump_tuple(theme.COLOR_QUIT_CHAR) @@ -992,8 +1011,8 @@ class MucTab(ChatTab): self.add_message(InfoMessage(leave_msg), typ=2) self.core.on_user_left_private_conversation(from_room, user, status) - def on_user_change_status(self, user, from_nick, from_room, affiliation, - role, show, status): + def on_user_change_status(self, user: User, from_nick: str, from_room: str, affiliation: str, + role: str, show: str, status: str) -> None: """ When a user changes her status """ @@ -1004,7 +1023,7 @@ class MucTab(ChatTab): self.general_jid): color = dump_tuple(user.color) else: - color = 3 + color = "3" info_col = dump_tuple(get_theme().COLOR_INFORMATION_TEXT) if from_nick == self.own_nick: msg = '\x19%(color)s}You\x19%(info_col)s} changed: ' % { @@ -1056,7 +1075,7 @@ class MucTab(ChatTab): user.update(affiliation, show, status, role) bisect.insort_left(self.users, user) - def disconnect(self): + def disconnect(self) -> None: """ Set the state of the room as not joined, so we can know if we can join it, send messages to it, etc @@ -1068,25 +1087,25 @@ class MucTab(ChatTab): self.joined = False self.disable_self_ping_event() - def get_single_line_topic(self): + def get_single_line_topic(self) -> str: """ Return the topic as a single-line string (for the window header) """ return self.topic.replace('\n', '|') - def log_message(self, msg: Message, typ=1): + def log_message(self, msg: Message, typ: int = 1) -> None: """ Log the messages in the archives, if it needs to be """ if not isinstance(msg, Message): return - if not msg.history and self.joined: # don't log the history messages + if not msg.history and self.joined and msg.nickname and msg.txt: # don't log the history messages if not logger.log_message(self.jid.bare, msg.nickname, msg.txt, typ=typ): self.core.information('Unable to write in the log file', 'Error') - def get_user_by_name(self, nick): + def get_user_by_name(self, nick: str) -> Optional[User]: """ Gets the user associated with the given nick, or None if not found """ @@ -1095,7 +1114,7 @@ class MucTab(ChatTab): return user return None - def add_message(self, msg: BaseMessage, typ=1) -> None: + def add_message(self, msg: BaseMessage, typ: int = 1) -> None: """Add a message to the text buffer and set various tab status""" # reset self-ping interval if self.self_ping_event: @@ -1108,17 +1127,18 @@ class MucTab(ChatTab): if config.get_by_tabname('notify_messages', self.jid.bare) and self.state != 'current': if msg.nickname != self.own_nick and not msg.history: self.state = 'message' - self.do_highlight(msg.txt, msg.nickname, msg.delayed) + if msg.txt and msg.nickname: + self.do_highlight(msg.txt, msg.nickname, msg.delayed) def modify_message(self, - txt, - old_id, - new_id, - time=None, + txt: str, + old_id: str, + new_id: str, + time: Optional[datetime] = None, delayed: bool = False, - nickname=None, - user=None, - jid=None): + nickname: Optional[str] = None, + user: Optional[User] = None, + jid: Optional[JID] = None) -> bool: highlight = self.message_is_highlight( txt, nickname, delayed, corrected=True ) @@ -1136,10 +1156,10 @@ class MucTab(ChatTab): return highlight return False - def matching_names(self): + def matching_names(self) -> List[Tuple[int, str]]: return [(1, self.jid.user), (3, self.jid.full)] - def enable_self_ping_event(self): + def enable_self_ping_event(self) -> None: delay = config.get_by_tabname( "self_ping_delay", self.general_jid, default=0) interval = int( @@ -1152,12 +1172,12 @@ class MucTab(ChatTab): interval, self.send_self_ping) self.core.add_timed_event(self.self_ping_event) - def disable_self_ping_event(self): + def disable_self_ping_event(self) -> None: if self.self_ping_event is not None: self.core.remove_timed_event(self.self_ping_event) self.self_ping_event = None - def send_self_ping(self): + def send_self_ping(self) -> None: timeout = config.get_by_tabname( "self_ping_timeout", self.general_jid, default=60) to = self.jid.bare + "/" + self.own_nick @@ -1167,7 +1187,7 @@ class MucTab(ChatTab): timeout_callback=self.on_self_ping_failed, timeout=timeout) - def on_self_ping_result(self, iq): + def on_self_ping_result(self, iq: Iq) -> None: if iq["type"] == "error" and iq["error"]["condition"] not in \ ("feature-not-implemented", "service-unavailable", "item-not-found"): self.command_cycle(iq["error"]["text"] or "not in this room") @@ -1176,13 +1196,13 @@ class MucTab(ChatTab): self.reset_lag() self.enable_self_ping_event() - def search_for_color(self, nick): + def search_for_color(self, nick: str) -> str: """ Search for the color of a nick in the config file. Also, look at the colors of its possible aliases if nick_color_aliases is set. """ - color = config.get_by_tabname(nick, 'muc_colors') + color = cast(str, config.get_by_tabname(nick, 'muc_colors')) if color != '': return color nick_color_aliases = config.get_by_tabname('nick_color_aliases', @@ -1192,7 +1212,7 @@ class MucTab(ChatTab): color = config.get_by_tabname(nick_alias, 'muc_colors') return color - def on_self_ping_failed(self, iq): + def on_self_ping_failed(self, iq: Any = None) -> None: if not self.lagged: self.lagged = True self._text_buffer.add_message( @@ -1204,7 +1224,7 @@ class MucTab(ChatTab): self.core.refresh_window() self.enable_self_ping_event() - def reset_lag(self): + def reset_lag(self) -> None: if self.lagged: self.lagged = False self.add_message( @@ -1219,30 +1239,30 @@ class MucTab(ChatTab): ########################## UI ONLY ##################################### @refresh_wrapper.always - def go_to_next_hl(self): + def go_to_next_hl(self) -> None: """ Go to the next HL in the room, or the last """ self.text_win.next_highlight() @refresh_wrapper.always - def go_to_prev_hl(self): + def go_to_prev_hl(self) -> None: """ Go to the previous HL in the room, or the first """ self.text_win.previous_highlight() @refresh_wrapper.always - def scroll_user_list_up(self): + def scroll_user_list_up(self) -> None: "Scroll up in the userlist" self.user_win.scroll_up() @refresh_wrapper.always - def scroll_user_list_down(self): + def scroll_user_list_down(self) -> None: "Scroll down in the userlist" self.user_win.scroll_down() - def resize(self): + def resize(self) -> None: """ Resize the whole window. i.e. all its sub-windows """ @@ -1278,7 +1298,7 @@ class MucTab(ChatTab): 0) self.input.resize(1, self.width, self.height - 1, 0) - def refresh(self): + def refresh(self) -> None: if self.need_resize: self.resize() log.debug(' TAB Refresh: %s', self.__class__.__name__) @@ -1301,7 +1321,7 @@ class MucTab(ChatTab): self.info_win.refresh() self.input.refresh() - def on_info_win_size_changed(self): + def on_info_win_size_changed(self) -> None: if self.core.information_win_size >= self.height - 3: return if config.get("hide_user_list"): @@ -1325,10 +1345,10 @@ class MucTab(ChatTab): # This maxsize is kinda arbitrary, but most users won’t have that many # nicknames anyway. @functools.lru_cache(maxsize=8) - def build_highlight_regex(self, nickname): + def build_highlight_regex(self, nickname: str) -> Pattern: return re.compile(r"(^|\W)" + re.escape(nickname) + r"(\W|$)", re.I) - def message_is_highlight(self, txt: str, nickname: str, delayed: bool, + def message_is_highlight(self, txt: str, nickname: Optional[str], delayed: bool, corrected: bool = False) -> bool: """Highlight algorithm for MUC tabs""" # Don't highlight on info message or our own messages @@ -1358,7 +1378,7 @@ class MucTab(ChatTab): if highlighted and self.joined and not corrected: if self.state != 'current': self.state = 'highlight' - beep_on = config.get('beep_on').split() + beep_on = cast(str, config.get('beep_on')).split() if 'highlight' in beep_on and 'message' not in beep_on: if not config.get_by_tabname('disable_beep', self.jid.bare): curses.beep() @@ -1368,31 +1388,33 @@ class MucTab(ChatTab): ########################## COMMANDS #################################### @command_args_parser.quoted(1, 1, ['']) - def command_invite(self, args): + def command_invite(self, args: List[str]) -> None: """/invite <jid> [reason]""" if args is None: - return self.core.command.help('invite') + self.core.command.help('invite') + return jid, reason = args self.core.command.invite('%s %s "%s"' % (jid, self.jid.bare, reason)) @command_args_parser.quoted(1) - def command_info(self, args): + def command_info(self, args: List[str]) -> None: """ /info <nick> """ if args is None: - return self.core.command.help('info') + self.core.command.help('info') + return nick = args[0] if not self.print_info(nick): self.core.information("Unknown user: %s" % nick, "Error") @command_args_parser.quoted(0) - def command_configure(self, ignored): + def command_configure(self, ignored: Any) -> None: """ /configure """ - def on_form_received(form): + def on_form_received(form: 'Form') -> None: if not form: self.core.information( 'Could not retrieve the configuration form', 'Error') @@ -1402,13 +1424,13 @@ class MucTab(ChatTab): fixes.get_room_form(self.core.xmpp, self.jid.bare, on_form_received) @command_args_parser.raw - def command_cycle(self, msg): + def command_cycle(self, msg: str) -> None: """/cycle [reason]""" self.leave_room(msg) self.join() @command_args_parser.quoted(0, 1, ['']) - def command_recolor(self, args): + def command_recolor(self, args: List[str]) -> None: """ /recolor [random] Re-assigns color to the participants of the room @@ -1417,7 +1439,7 @@ class MucTab(ChatTab): self.recolor(random_colors) @command_args_parser.quoted(2, 2, ['']) - def command_color(self, args): + def command_color(self, args: List[str]) -> None: """ /color <nick> <color> Fix a color for a nick. @@ -1425,24 +1447,28 @@ class MucTab(ChatTab): User "random" to attribute a random color. """ if args is None: - return self.core.command.help('color') + self.core.command.help('color') + return nick = args[0] color = args[1].lower() if nick == self.own_nick: - return self.core.information( + self.core.information( "You cannot change the color of your" - " own nick.", 'Error') + " own nick.", 'Error' + ) elif color not in xhtml.colors and color not in ('unset', 'random'): - return self.core.information("Unknown color: %s" % color, 'Error') - self.set_nick_color(nick, color) + self.core.information("Unknown color: %s" % color, 'Error') + else: + self.set_nick_color(nick, color) @command_args_parser.quoted(1) - def command_version(self, args): + def command_version(self, args: List[str]) -> None: """ /version <jid or nick> """ if args is None: - return self.core.command.help('version') + self.core.command.help('version') + return nick = args[0] try: if nick in [user.nick for user in self.users]: @@ -1451,32 +1477,36 @@ class MucTab(ChatTab): else: jid = JID(nick) except InvalidJID: - return self.core.information('Invalid jid or nick %r' % nick, 'Error') + self.core.information('Invalid jid or nick %r' % nick, 'Error') + return self.core.xmpp.plugin['xep_0092'].get_version( jid, callback=self.core.handler.on_version_result) @command_args_parser.quoted(1) - def command_nick(self, args): + def command_nick(self, args: List[str]) -> None: """ /nick <nickname> """ if args is None: - return self.core.command.help('nick') + self.core.command.help('nick') + return nick = args[0] if not self.joined: - return self.core.information('/nick only works in joined rooms', + self.core.information('/nick only works in joined rooms', 'Info') + return current_status = self.core.get_status() 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.change_nick(self.core, self.jid.bare, nick, current_status.message, current_status.show) @command_args_parser.quoted(0, 1, ['']) - def command_part(self, args): + def command_part(self, args: List[str]) -> None: """ /part [msg] """ @@ -1487,7 +1517,7 @@ class MucTab(ChatTab): self.core.doupdate() @command_args_parser.raw - def command_leave(self, msg): + def command_leave(self, msg: str) -> None: """ /leave [msg] """ @@ -1498,25 +1528,26 @@ class MucTab(ChatTab): self.core.close_tab(self) @command_args_parser.raw - def command_close(self, msg): + def command_close(self, msg: str) -> None: """ /close [msg] """ self.leave_room(msg) self.core.close_tab(self) - def on_close(self): + def on_close(self) -> None: super().on_close() if self.joined: self.leave_room('') @command_args_parser.quoted(1, 1) - def command_query(self, args): + def command_query(self, args: List[str]) -> None: """ /query <nick> [message] """ if args is None: - return self.core.command.help('query') + self.core.command.help('query') + return nick = args[0] r = None for user in self.users: @@ -1524,13 +1555,14 @@ class MucTab(ChatTab): r = self.core.open_private_window(self.jid.bare, user.nick) if r and len(args) == 2: msg = args[1] - self.core.tabs.current_tab.command_say( - xhtml.convert_simple_to_full_colors(msg)) + r.command_say( + xhtml.convert_simple_to_full_colors(msg) + ) if not r: self.core.information("Cannot find user: %s" % nick, 'Error') @command_args_parser.raw - def command_topic(self, subject): + def command_topic(self, subject: str) -> None: """ /topic [new topic] """ @@ -1540,7 +1572,7 @@ class MucTab(ChatTab): self.change_topic(subject) @command_args_parser.quoted(0) - def command_names(self, args): + def command_names(self, args: Any) -> None: """ /names """ @@ -1578,12 +1610,13 @@ class MucTab(ChatTab): self.input.refresh() @command_args_parser.quoted(1, 1) - def command_kick(self, args): + def command_kick(self, args: List[str]) -> None: """ /kick <nick> [reason] """ if args is None: - return self.core.command.help('kick') + self.core.command.help('kick') + return if len(args) == 2: reason = args[1] else: @@ -1592,36 +1625,38 @@ class MucTab(ChatTab): self.change_role(nick, 'none', reason) @command_args_parser.quoted(1, 1) - def command_ban(self, args): + def command_ban(self, args: List[str]) -> None: """ /ban <nick> [reason] """ if args is None: - return self.core.command.help('ban') + self.core.command.help('ban') + return nick = args[0] msg = args[1] if len(args) == 2 else '' self.change_affiliation(nick, 'outcast', msg) @command_args_parser.quoted(2, 1, ['']) - def command_role(self, args): + def command_role(self, args: List[str]) -> None: """ /role <nick> <role> [reason] Changes the role of a user roles can be: none, visitor, participant, moderator """ - def callback(iq): + def callback(iq: Iq) -> None: if iq['type'] == 'error': self.core.room_error(iq, self.jid.bare) if args is None: - return self.core.command.help('role') + self.core.command.help('role') + return nick, role, reason = args[0], args[1].lower(), args[2] self.change_role(nick, role, reason) @command_args_parser.quoted(0, 2) - def command_affiliation(self, args) -> None: + def command_affiliation(self, args: List[str]) -> None: """ /affiliation [<nick or jid> <affiliation>] Changes the affiliation of a user @@ -1639,7 +1674,8 @@ class MucTab(ChatTab): return None if len(args) != 2: - return self.core.command.help('affiliation') + self.core.command.help('affiliation') + return nick, affiliation = args[0], args[1].lower() # Set affiliation @@ -1677,9 +1713,10 @@ class MucTab(ChatTab): ) return None + lines = ['Affiliations for %s' % jid.bare] for iq in iqs: - if isinstance(iq, (IqError, IqTimeout)): + if isinstance(iq, BaseException): continue query = iq.xml.find('{%s}query' % MUC_ADMIN_NS) @@ -1699,7 +1736,7 @@ class MucTab(ChatTab): return None @command_args_parser.raw - def command_say(self, line, correct=False): + def command_say(self, line: str, correct: bool = False) -> None: """ /say <message> Or normal input + enter @@ -1738,19 +1775,20 @@ class MucTab(ChatTab): self.chat_state = needed @command_args_parser.raw - def command_xhtml(self, msg): + def command_xhtml(self, msg: str) -> None: message = self.generate_xhtml_message(msg) if message: message['type'] = 'groupchat' message.send() @command_args_parser.quoted(1) - def command_ignore(self, args): + def command_ignore(self, args: List[str]) -> None: """ /ignore <nick> """ if args is None: - return self.core.command.help('ignore') + self.core.command.help('ignore') + return nick = args[0] user = self.get_user_by_name(nick) @@ -1763,12 +1801,13 @@ class MucTab(ChatTab): self.core.information("%s is now ignored" % nick, 'info') @command_args_parser.quoted(1) - def command_unignore(self, args): + def command_unignore(self, args: List[str]) -> None: """ /unignore <nick> """ if args is None: - return self.core.command.help('unignore') + self.core.command.help('unignore') + return nick = args[0] user = self.get_user_by_name(nick) @@ -1782,7 +1821,7 @@ class MucTab(ChatTab): ########################## COMPLETIONS ################################# - def completion(self): + def completion(self) -> None: """ Called when Tab is pressed, complete the nickname in the input """ @@ -1795,7 +1834,7 @@ class MucTab(ChatTab): for user in sorted(self.users, key=COMPARE_USERS_LAST_TALKED, reverse=True): if user.nick != self.own_nick: word_list.append(user.nick) - after = config.get('after_completion') + ' ' + after = cast(str, config.get('after_completion')) + ' ' input_pos = self.input.pos if ' ' not in self.input.get_text()[:input_pos] or ( self.input.last_completion and self.input.get_text() @@ -1813,7 +1852,7 @@ class MucTab(ChatTab): and not self.input.get_text().startswith('//')) self.send_composing_chat_state(empty_after) - def completion_version(self, the_input): + def completion_version(self, the_input: windows.MessageInput) -> Completion: """Completion for /version""" userlist = [] for user in sorted(self.users, key=COMPARE_USERS_LAST_TALKED, reverse=True): @@ -1828,30 +1867,30 @@ class MucTab(ChatTab): return Completion(the_input.auto_completion, userlist, quotify=False) - def completion_info(self, the_input): + def completion_info(self, the_input: windows.MessageInput) -> Completion: """Completion for /info""" userlist = [] for user in sorted(self.users, key=COMPARE_USERS_LAST_TALKED, reverse=True): userlist.append(user.nick) return Completion(the_input.auto_completion, userlist, quotify=False) - def completion_nick(self, the_input): + def completion_nick(self, the_input: windows.MessageInput) -> Completion: """Completion for /nick""" - nicks = [ + nicks_list = [ os.environ.get('USER'), - config.get('default_nick'), + cast(str, config.get('default_nick')), self.core.get_bookmark_nickname(self.jid.bare) ] - nicks = [i for i in nicks if i] + nicks = [i for i in nicks_list if i] return Completion(the_input.auto_completion, nicks, '', quotify=False) - def completion_recolor(self, the_input): + def completion_recolor(self, the_input: windows.MessageInput) -> Optional[Completion]: if the_input.get_argument_position() == 1: return Completion( the_input.new_completion, ['random'], 1, '', quotify=False) - return True + return None - def completion_color(self, the_input): + def completion_color(self, the_input: windows.MessageInput) -> Optional[Completion]: """Completion for /color""" n = the_input.get_argument_position(quoted=True) if n == 1: @@ -1867,8 +1906,9 @@ class MucTab(ChatTab): colors.append('random') return Completion( the_input.new_completion, colors, 2, '', quotify=False) + return None - def completion_ignore(self, the_input): + def completion_ignore(self, the_input: windows.MessageInput) -> Completion: """Completion for /ignore""" userlist = [user.nick for user in self.users] if self.own_nick in userlist: @@ -1876,7 +1916,7 @@ class MucTab(ChatTab): userlist.sort() return Completion(the_input.auto_completion, userlist, quotify=False) - def completion_role(self, the_input): + def completion_role(self, the_input: windows.MessageInput) -> Optional[Completion]: """Completion for /role""" n = the_input.get_argument_position(quoted=True) if n == 1: @@ -1889,8 +1929,9 @@ class MucTab(ChatTab): possible_roles = ['none', 'visitor', 'participant', 'moderator'] return Completion( the_input.new_completion, possible_roles, 2, '', quotify=True) + return None - def completion_affiliation(self, the_input): + def completion_affiliation(self, the_input: windows.MessageInput) -> Optional[Completion]: """Completion for /affiliation""" n = the_input.get_argument_position(quoted=True) if n == 1: @@ -1913,20 +1954,23 @@ class MucTab(ChatTab): 2, '', quotify=True) + return None - def completion_invite(self, the_input): + def completion_invite(self, the_input: windows.MessageInput) -> Optional[Completion]: """Completion for /invite""" n = the_input.get_argument_position(quoted=True) if n == 1: return Completion( the_input.new_completion, roster.jids(), 1, quotify=True) + return None - def completion_topic(self, the_input): + def completion_topic(self, the_input: windows.MessageInput) -> Optional[Completion]: if the_input.get_argument_position() == 1: return Completion( the_input.auto_completion, [self.topic], '', quotify=False) + return None - def completion_quoted(self, the_input): + def completion_quoted(self, the_input: windows.MessageInput) -> Optional[Completion]: """Nick completion, but with quotes""" if the_input.get_argument_position(quoted=True) == 1: word_list = [] @@ -1936,16 +1980,18 @@ class MucTab(ChatTab): return Completion( the_input.new_completion, word_list, 1, quotify=True) + return None - def completion_unignore(self, the_input): + def completion_unignore(self, the_input: windows.MessageInput) -> Optional[Completion]: if the_input.get_argument_position() == 1: users = [user.nick for user in self.ignores] return Completion(the_input.auto_completion, users, quotify=False) + return None ########################## REGISTER STUFF ############################## - def register_keys(self): + def register_keys(self) -> None: "Register tab-specific keys" self.key_func['^I'] = self.completion self.key_func['M-u'] = self.scroll_user_list_down @@ -1953,7 +1999,7 @@ class MucTab(ChatTab): self.key_func['M-n'] = self.go_to_next_hl self.key_func['M-p'] = self.go_to_prev_hl - def register_commands(self): + def register_commands(self) -> None: "Register tab-specific commands" self.register_commands_batch([{ 'name': 'ignore', diff --git a/poezio/tabs/privatetab.py b/poezio/tabs/privatetab.py index cd2123f3..e4937894 100644 --- a/poezio/tabs/privatetab.py +++ b/poezio/tabs/privatetab.py @@ -145,7 +145,7 @@ class PrivateTab(OneToOneTab): @refresh_wrapper.always @command_args_parser.raw - def command_say(self, line, attention=False, correct=False): + def command_say(self, line: str, attention: bool = False, correct: bool = False) -> None: if not self.on: return our_jid = JID(self.jid.bare) diff --git a/poezio/text_buffer.py b/poezio/text_buffer.py index ff853a67..6ef8e3d4 100644 --- a/poezio/text_buffer.py +++ b/poezio/text_buffer.py @@ -32,6 +32,7 @@ from poezio.ui.types import ( if TYPE_CHECKING: from poezio.windows.text_win import TextWin + from poezio.user import User class CorrectionError(Exception): @@ -249,7 +250,7 @@ class TextBuffer: new_id: str, highlight: bool = False, time: Optional[datetime] = None, - user: Optional[str] = None, + user: Optional['User'] = None, jid: Optional[str] = None) -> Message: """ Correct a message in a text buffer. diff --git a/poezio/user.py b/poezio/user.py index bead0a93..9a14e6b1 100644 --- a/poezio/user.py +++ b/poezio/user.py @@ -55,7 +55,7 @@ class User: else: self.color = choice(get_theme().LIST_COLOR_NICKNAMES) - def set_deterministic_color(self): + def set_deterministic_color(self) -> None: theme = get_theme() if theme.ccg_palette: # use XEP-0392 CCG diff --git a/poezio/windows/info_wins.py b/poezio/windows/info_wins.py index d31130fe..c3975c8c 100644 --- a/poezio/windows/info_wins.py +++ b/poezio/windows/info_wins.py @@ -3,6 +3,8 @@ Module defining all the "info wins", ie the bar which is on top of the info buffer in normal tabs """ +from typing import Optional, Dict, TYPE_CHECKING, Any + import logging log = logging.getLogger(__name__) @@ -13,6 +15,11 @@ from poezio.windows.base_wins import Win from poezio.ui.funcs import truncate_nick from poezio.theming import get_theme, to_curses_attr +if TYPE_CHECKING: + from poezio.user import User + from poezio.tabs import MucTab + from poezio.windows import TextWin + class InfoWin(Win): """ @@ -260,10 +267,16 @@ class MucInfoWin(InfoWin): __slots__ = () - def __init__(self): + def __init__(self) -> None: InfoWin.__init__(self) - def refresh(self, room, window=None, user=None, information=None): + def refresh( + self, + room: 'MucTab', + window: Optional['TextWin'] = None, + user: Optional['User'] = None, + information: Optional[Dict[str, Any]] = None + ) -> None: log.debug('Refresh: %s', self.__class__.__name__) self._win.erase() self.write_room_name(room) |