Commit 291233bb authored by mathieui's avatar mathieui

Merge branch 'more-typing' into 'master'

More typing

See merge request !135
parents e662bdee c9e219c1
......@@ -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.
......
......@@ -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
......
"""
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)
......@@ -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)
......
......@@ -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()
......
......@@ -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
......
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: ...
......@@ -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