Commit 4b198be9 authored by mathieui's avatar mathieui

fix: tons of type errors

parent bc4f4f1e
......@@ -250,7 +250,7 @@ class BookmarkList:
if core is not None:
core.information('Bookmarks saved', 'Info')
return result
except (IqError, IqTimeout) as iq:
except (IqError, IqTimeout):
if core is not None:
core.information(
'Could not save remote bookmarks.',
......@@ -318,7 +318,7 @@ class BookmarkList:
self.append(b)
def stanza_storage(bookmarks: BookmarkList) -> Bookmarks:
def stanza_storage(bookmarks: Union[BookmarkList, List[Bookmark]]) -> Bookmarks:
"""Generate a <storage/> stanza with the conference elements."""
storage = Bookmarks()
for b in (b for b in bookmarks if b.method == 'remote'):
......
from typing import Tuple, Dict, List
from typing import Tuple, Dict, List, Union
import curses
import hashlib
import math
......@@ -15,6 +15,9 @@ K_B = 1 - K_R - K_G
def ncurses_color_to_rgb(color: int) -> Tuple[float, float, float]:
if color <= 15:
r: Union[int, float]
g: Union[int, float]
b: Union[int, float]
try:
(r, g, b) = curses.color_content(color)
except: # fallback in faulty terminals (e.g. xterm)
......@@ -83,6 +86,9 @@ def ccg_palette_lookup(palette: Palette, angle: float) -> int:
best_metric = metric
best = color
if best is None:
raise ValueError("No color in palette")
return best
......
......@@ -19,16 +19,19 @@ import pkg_resources
from configparser import RawConfigParser, NoOptionError, NoSectionError
from pathlib import Path
from shutil import copy2
from typing import Callable, Dict, List, Optional, Union, Tuple, cast
from typing import Callable, Dict, List, Optional, Union, Tuple, cast, TypeVar
from poezio.args import parse_args
from poezio import xdg
ConfigValue = Union[str, int, float, bool]
ConfigDict = Dict[str, Dict[str, ConfigValue]]
DEFSECTION = "Poezio"
DEFAULT_CONFIG = {
DEFAULT_CONFIG: ConfigDict = {
'Poezio': {
'ack_message_receipts': True,
'add_space_after_completion': True,
......@@ -155,9 +158,10 @@ DEFAULT_CONFIG = {
'muc_colors': {}
}
T = TypeVar('T', bool, int, float, str)
class PoezioConfigParser(RawConfigParser):
class PoezioConfigParser(RawConfigParser):
def optionxform(self, value) -> str:
return str(value)
......@@ -169,14 +173,14 @@ class Config:
configparser: PoezioConfigParser
file_name: Path
default: Dict[str, Dict[str, ConfigValue]]
default: ConfigDict
def __init__(self, file_name: Path, default=None) -> None:
def __init__(self, file_name: Path, default: Optional[ConfigDict] = None) -> None:
self.configparser = PoezioConfigParser()
# make the options case sensitive
self.file_name = file_name
self.read_file()
self.default = default
self.default = default or {}
def optionxform(self, value):
return str(value)
......@@ -192,8 +196,8 @@ class Config:
def get(self,
option: str,
default: Optional[ConfigValue] = None,
section=DEFSECTION) -> Optional[ConfigValue]:
default: Optional[T] = None,
section=DEFSECTION) -> Optional[T]:
"""
get a value from the config but return
a default value if it is not found
......@@ -201,22 +205,24 @@ class Config:
returned
"""
if default is None:
if self.default:
default = self.default.get(section, {}).get(option)
else:
default = ''
section = self.default.get('section')
if section is not None:
option = section.get(option)
if option is not None:
default = cast(T, option)
res: T
try:
if isinstance(default, bool):
res = self.configparser.getboolean(section, option)
elif isinstance(default, int):
res = self.configparser.getint(section, option)
elif isinstance(default, float):
res = self.configparser.getfloat(section, option)
elif isinstance(default, int):
res = self.configparser.getint(section, option)
else:
res = self.configparser.get(section, option)
except (NoOptionError, NoSectionError, ValueError, AttributeError):
return default if default is not None else ''
return default
if res is None:
return default
......@@ -514,26 +520,26 @@ class Config:
Set a value, save, and return True on success and False on failure
"""
if self.has_section(section):
self.configparser.set(section, option, value)
self.configparser.set(section, option, str(value))
else:
self.add_section(section)
self.configparser.set(section, option, value)
return self.write_in_file(section, option, value)
self.configparser.set(section, option, str(value))
return self.write_in_file(section, option, str(value))
def set(self, option: str, value: ConfigValue, section=DEFSECTION):
"""
Set the value of an option temporarily
"""
try:
self.configparser.set(section, option, value)
self.configparser.set(section, option, str(value))
except NoSectionError:
pass
def to_dict(self) -> Dict[str, Dict[str, ConfigValue]]:
def to_dict(self) -> Dict[str, Dict[str, Optional[ConfigValue]]]:
"""
Returns a dict of the form {section: {option: value, option: value}, …}
"""
res = {} # Dict[str, Dict[str, ConfigValue]]
res: Dict[str, Dict[str, Optional[ConfigValue]]] = {}
for section in self.sections():
res[section] = {}
for option in self.options(section):
......@@ -567,10 +573,10 @@ def file_ok(filepath: Path) -> bool:
return bool(val)
def get_image_cache() -> Path:
def get_image_cache() -> Optional[Path]:
if not config.get('extract_inline_images'):
return None
tmp_dir = config.get('tmp_image_dir')
tmp_dir = config.getstr('tmp_image_dir')
if tmp_dir:
return Path(tmp_dir)
return xdg.CACHE_HOME / 'images'
......@@ -727,7 +733,7 @@ firstrun = False
config = None # type: Config
# The logger object for this module
log = None # type: Optional[logging.Logger]
log = logging.getLogger(__name__) # type: logging.Logger
# The command-line options
options = None
......
......@@ -45,15 +45,18 @@ class Resource:
@property
def priority(self) -> int:
return self._data.get('priority') or 0
try:
return int(self._data.get('priority', 0))
except Exception:
return 0
@property
def presence(self) -> str:
return self._data.get('show') or ''
return str(self._data.get('show')) or ''
@property
def status(self) -> str:
return self._data.get('status') or ''
return str(self._data.get('status')) or ''
def __repr__(self) -> str:
return '<%s>' % self._jid
......
from typing import Callable, List
from typing import Callable, List, Optional
from poezio.core.commands import CommandCore
from poezio.core.completions import CompletionCore
......@@ -22,7 +22,7 @@ CommandDict = TypedDict(
"shortdesc": str,
"desc": str,
"usage": str,
"completion": Callable,
"completion": Optional[Callable],
},
total=False,
)
......
......@@ -16,6 +16,7 @@ import time
import uuid
from collections import defaultdict
from typing import (
Any,
cast,
Callable,
Dict,
......@@ -233,9 +234,9 @@ class Core:
'_dnd': lambda: self.command.status('dnd'),
'_xa': lambda: self.command.status('xa'),
##### Custom actions ########
'_exc_': self.try_execute,
}
self.key_func.update(key_func)
self.key_func.try_execute = self.try_execute
# Add handlers
xmpp_event_handlers = [
......@@ -796,12 +797,14 @@ class Core:
def remove_timed_event(self, event: DelayedEvent) -> None:
"""Remove an existing timed event"""
event.handler.cancel()
if event.handler is not None:
event.handler.cancel()
def add_timed_event(self, event: DelayedEvent) -> None:
"""Add a new timed event"""
event.handler = asyncio.get_event_loop().call_later(
event.delay, event.callback, *event.args)
event.delay, event.callback, *event.args
)
####################### XMPP-related actions ##################################
......@@ -1174,6 +1177,7 @@ class Core:
provided, we open a StaticConversationTab, else a
DynamicConversationTab
"""
new_tab: tabs.ConversationTab
if jid.resource:
new_tab = tabs.StaticConversationTab(self, jid)
else:
......@@ -1196,19 +1200,19 @@ class Core:
self.tabs.set_current_tab(tab)
return tab
# create the new tab
tab = self.tabs.by_name_and_class(room_name, tabs.MucTab)
if not tab:
muc_tab = self.tabs.by_name_and_class(room_name, tabs.MucTab)
if not muc_tab:
return None
new_tab = tabs.PrivateTab(self, complete_jid, tab.own_nick)
tab = tabs.PrivateTab(self, complete_jid, muc_tab.own_nick)
if hasattr(tab, 'directed_presence'):
new_tab.directed_presence = tab.directed_presence
tab.directed_presence = tab.directed_presence
if not focus:
new_tab.state = "private"
tab.state = "private"
# insert it in the tabs
self.add_tab(new_tab, focus)
self.add_tab(tab, focus)
self.refresh_window()
tab.privates.append(new_tab)
return new_tab
muc_tab.privates.append(tab)
return tab
def open_new_room(self,
room: str,
......@@ -1764,18 +1768,23 @@ class Core:
self.refresh_window()
class KeyDict(dict):
class KeyDict(Dict[str, Callable[[str], Any]]):
"""
A dict, with a wrapper for get() that will return a custom value
if the key starts with _exc_
"""
try_execute: Optional[Callable[[str], Any]]
def get(self, key: str, default: Optional[Callable] = None) -> Callable:
if isinstance(key, str) and key.startswith('_exc_') and len(key) > 5:
return lambda: dict.get(self, '_exc_')(key[5:])
if self.try_execute is not None:
try_execute = self.try_execute
return lambda: try_execute(key[5:])
raise ValueError("KeyDict not initialized")
return dict.get(self, key, default)
def replace_key_with_bound(key: str) -> str:
"""
Replace an inputted key with the one defined as its replacement
......
"""
Module defining structures useful to the core class and related methods
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Callable, List, Dict
from typing import Any, Callable, List, TYPE_CHECKING, Optional
if TYPE_CHECKING:
from poezio import windows
__all__ = [
'Command',
......@@ -34,6 +38,7 @@ class Completion:
A completion result essentially currying the input completion call.
"""
__slots__ = ['func', 'args', 'kwargs', 'comp_list']
def __init__(
self,
func: Callable[..., Any],
......@@ -49,11 +54,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]
comp: Optional[Callable[['windows.Input'], Completion]]
short_desc: str
usage: str
......@@ -13,20 +13,19 @@ from typing import (
List,
Optional,
TypeVar,
TYPE_CHECKING,
)
from poezio import common
if TYPE_CHECKING:
from poezio.tabs import RosterInfoTab
T = TypeVar('T', bound=Callable[..., Any])
BeforeFunc = Callable[[List[Any], Dict[str, Any]], Any]
AfterFunc = Callable[[List[Any], Dict[str, Any]], Any]
def wrap_generic(func: Callable, before: BeforeFunc=None, after: AfterFunc=None):
BeforeFunc = Optional[Callable[[List[Any], Dict[str, Any]], Any]]
AfterFunc = Optional[Callable[[Any, List[Any], Dict[str, Any]], Any]]
def wrap_generic(func: Callable, before: BeforeFunc = None, after: AfterFunc = None):
"""
Generic wrapper which can both wrap coroutines and normal functions.
"""
......
......@@ -306,4 +306,4 @@ def create_logger() -> None:
logger = Logger()
logger = None # type: Logger
logger = Logger()
......@@ -32,7 +32,7 @@ log = logging.getLogger(__name__)
if TYPE_CHECKING:
from poezio.core.core import Core
from poezio.tabs import Tab
from poezio.tabs import MucTab
def change_show(
......@@ -79,7 +79,7 @@ def join_groupchat(
status: Optional[str] = None,
show: Optional[str] = None,
seconds: Optional[int] = None,
tab: Optional[Tab] = None
tab: Optional['MucTab'] = None
) -> None:
xmpp = core.xmpp
stanza = xmpp.make_presence(
......
......@@ -10,7 +10,16 @@
Interface for E2EE (End-to-end Encryption) plugins.
"""
from typing import Callable, Dict, List, Optional, Union, Tuple, Set
from typing import (
Callable,
Dict,
List,
Optional,
Union,
Tuple,
Set,
Type,
)
from slixmpp import InvalidJID, JID, Message
from slixmpp.xmlstream import StanzaBase
......@@ -117,7 +126,7 @@ class E2EEPlugin(BasePlugin):
_enabled_tabs: Dict[JID, Callable] = {}
# Tabs that support this encryption mechanism
supported_tab_types: Tuple[ChatTabs] = tuple()
supported_tab_types: Tuple[Type[ChatTabs], ...] = tuple()
# States for each remote entity
trust_states: Dict[str, Set[str]] = {'accepted': set(), 'rejected': set()}
......@@ -224,7 +233,7 @@ class E2EEPlugin(BasePlugin):
except InvalidJID:
return ""
if self._encryption_enabled(jid):
if self._encryption_enabled(jid) and self.encryption_short_name:
return " " + self.encryption_short_name
return ""
......@@ -238,7 +247,7 @@ class E2EEPlugin(BasePlugin):
'{} encryption disabled for {}'.format(self.encryption_name, jid),
'Info',
)
else:
elif self.encryption_short_name:
self._enabled_tabs[jid] = self.encrypt
config.set_and_save('encryption', self.encryption_short_name, section=jid)
self.api.information(
......@@ -368,9 +377,9 @@ class E2EEPlugin(BasePlugin):
# comes from a semi-anonymous MUC for example. Some plugins might be
# fine with this so let them handle it.
jid = message['from']
muctab = tab
if isinstance(muctab, PrivateTab):
muctab = None
if isinstance(tab, PrivateTab):
muctab = tab.parent_muc
jid = None
......@@ -386,7 +395,7 @@ class E2EEPlugin(BasePlugin):
log.debug('Decrypted %s message: %r', self.encryption_name, message['body'])
return None
async def _encrypt(self, stanza: StanzaBase) -> Optional[StanzaBase]:
async def _encrypt(self, stanza: StanzaBase, passthrough: bool = True) -> Optional[StanzaBase]:
if not isinstance(stanza, Message) or stanza['type'] not in ('normal', 'chat', 'groupchat'):
raise NothingToEncrypt()
message = stanza
......
......@@ -26,6 +26,7 @@ from xml.sax import SAXParseException
from typing import (
Any,
Callable,
cast,
Dict,
List,
Optional,
......@@ -48,6 +49,7 @@ from poezio.text_buffer import TextBuffer
from poezio.theming import get_theme, dump_tuple
from poezio.ui.funcs import truncate_nick
from poezio.ui.types import BaseMessage, InfoMessage, Message
from poezio.timed_events import DelayedEvent
from slixmpp import JID, InvalidJID, Message as SMessage
......@@ -117,6 +119,10 @@ class Tab:
# Placeholder values, set on resize
height = 1
width = 1
core: Core
input: Optional[windows.Input]
key_func: Dict[str, Callable[[], Any]]
commands: Dict[str, Command]
def __init__(self, core: Core):
self.core = core
......@@ -234,7 +240,7 @@ class Tab:
*,
desc='',
shortdesc='',
completion: Optional[Callable] = None,
completion: Optional[Callable[[windows.Input], Completion]] = None,
usage=''):
"""
Add a command
......@@ -286,7 +292,6 @@ class Tab:
comp = command.comp(the_input)
if comp:
return comp.run()
return comp
return False
def execute_command(self, provided_text: str) -> bool:
......@@ -294,8 +299,10 @@ class Tab:
Execute the command in the input and return False if
the input didn't contain a command
"""
if self.input is None:
raise NotImplementedError
txt = provided_text or self.input.key_enter()
if txt.startswith('/') and not txt.startswith('//') and\
if txt and txt.startswith('/') and not txt.startswith('//') and\
not txt.startswith('/me '):
command = txt.strip().split()[0][1:]
arg = txt[2 + len(command):] # jump the '/' and the ' '
......@@ -461,6 +468,9 @@ class Tab:
class GapTab(Tab):
def __init__(self, core: Optional[Core], *args, **kwargs):
super().__init__(core, **args, **kwargs)
def __bool__(self):
return False
......@@ -485,7 +495,10 @@ class ChatTab(Tab):
"""
plugin_commands: Dict[str, Command] = {}
plugin_keys: Dict[str, Callable] = {}
last_sent_message: Optional[SMessage]
message_type = 'chat'
timed_event_paused: Optional[DelayedEvent]
timed_event_not_paused: Optional[DelayedEvent]
def __init__(self, core, jid: Union[JID, str]):
Tab.__init__(self, core)
......@@ -507,7 +520,7 @@ class ChatTab(Tab):
self.timed_event_paused = None
self.timed_event_not_paused = None
# Keeps the last sent message to complete it easily in completion_correct, and to replace it.
self.last_sent_message = {}
self.last_sent_message = None
self.key_func['M-v'] = self.move_separator
self.key_func['M-h'] = self.scroll_separator
self.key_func['M-/'] = self.last_words_completion
......@@ -625,6 +638,8 @@ class ChatTab(Tab):
self.input.auto_completion(words, ' ', quotify=False)
def on_enter(self):
if self.input is None:
raise NotImplementedError
txt = self.input.key_enter()
if txt:
if not self.execute_command(txt):
......@@ -740,16 +755,18 @@ class ChatTab(Tab):
if self.timed_event_paused is not None:
self.core.remove_timed_event(self.timed_event_paused)
self.timed_event_paused = None
self.core.remove_timed_event(self.timed_event_not_paused)
self.timed_event_not_paused = None
if self.timed_event_not_paused is not None:
self.core.remove_timed_event(self.timed_event_not_paused)
self.timed_event_not_paused = None
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
# afterwards? Who knows.
msg = copy(msg)
msg['id'] = self.last_sent_message['id']
msg = cast(SMessage, copy(msg))
if self.last_sent_message is not None:
msg['id'] = self.last_sent_message['id']
self.last_sent_message = msg
@command_args_parser.raw
......@@ -783,7 +800,8 @@ class ChatTab(Tab):
self.text_win.remove_line_separator()
self.text_win.add_line_separator(self._text_buffer)
self.text_win.refresh()
self.input.refresh()
if self.input:
self.input.refresh()
def get_conversation_messages(self):
return self._text_buffer.messages
......
......@@ -34,8 +34,8 @@ class BookmarksTab(Tab):
self.new_bookmarks: List[Bookmark] = []
self.removed_bookmarks: List[Bookmark] = []
self.header_win = windows.ColumnHeaderWin(
('name', 'room@server/nickname', 'password', 'autojoin',
'storage'))
['name', 'room@server/nickname', 'password', 'autojoin',
'storage'])
self.bookmarks_win = windows.BookmarksWin(
self.bookmarks, self.height - 4, self.width, 1, 0)
self.help_win = windows.HelpText('Ctrl+Y: save, Ctrl+G: cancel, '
......
......@@ -166,6 +166,7 @@ class MucTab(ChatTab):
"""
Join the room
"""
seconds: Optional[int]
status = self.core.get_status()
if self.last_connection:
delta = to_utc(datetime.now()) - to_utc(self.last_connection)
......@@ -725,7 +726,7 @@ class MucTab(ChatTab):
new_nick = presence.xml.find(
'{%s}x/{%s}item' % (NS_MUC_USER, NS_MUC_USER)
).attrib['nick']
old_color = user.color
old_color_tuple = user.color
if user.nick == self.own_nick:
self.own_nick = new_nick
# also change our nick in all private discussions of this room
......@@ -741,7 +742,7 @@ class MucTab(ChatTab):
if config.get_by_tabname('display_user_color_in_join_part',
self.general_jid):
color = dump_tuple(user.color)
old_color = dump_tuple(old_color)
old_color = dump_tuple(old_color_tuple)
else:
old_color = color = "3"