Commit d54d4bc2 authored by Philipp Hörist's avatar Philipp Hörist

Add annotations and fix pylint/mypy errors

parent a62f348a
Pipeline #2182 passed with stages
in 3 minutes and 58 seconds
......@@ -273,7 +273,8 @@ class GajimApplication(Gtk.Application):
self.activate()
return 0
def _handle_local_options(self, application,
def _handle_local_options(self,
application: Gtk.Application,
options: GLib.VariantDict) -> int:
# Parse all options that have to be executed before ::startup
if options.contains('profile'):
......
......@@ -413,7 +413,7 @@ class ChatControl(ChatControlBase):
img = self._pep_images[pep_type]
if pep_type in pep:
img.set_from_pixbuf(gtkgui_helpers.get_pep_as_pixbuf(pep[pep_type]))
img.set_tooltip_markup(pep[pep_type].asMarkupText())
img.set_tooltip_markup(pep[pep_type].as_markup_text())
img.show()
else:
img.hide()
......
......@@ -31,62 +31,77 @@ to automatic discovery and dispatching, also features manual control
over the process.
"""
from typing import Any # pylint: disable=unused-import
from typing import Dict # pylint: disable=unused-import
from gajim.command_system.tools import remove
COMMANDS = {}
CONTAINERS = {}
COMMANDS = {} # type: Dict[Any, Any]
CONTAINERS = {} # type: Dict[Any, Any]
def add_host(host):
CONTAINERS[host] = []
def remove_host(host):
remove(CONTAINERS, host)
def add_container(container):
for host in container.HOSTS:
CONTAINERS[host].append(container)
def remove_container(container):
for host in container.HOSTS:
remove(CONTAINERS[host], container)
def add_commands(container):
commands = COMMANDS.setdefault(container, {})
for command in traverse_commands(container):
for name in command.names:
commands[name] = command
def remove_commands(container):
remove(COMMANDS, container)
def traverse_commands(container):
for name in dir(container):
attribute = getattr(container, name)
if is_command(attribute):
yield attribute
def is_command(attribute):
from gajim.command_system.framework import Command
return isinstance(attribute, Command)
def is_root(namespace):
metaclass = namespace.get("__metaclass__", None)
if not metaclass:
return False
return issubclass(metaclass, Dispatchable)
def get_command(host, name):
for container in CONTAINERS[host]:
command = COMMANDS[container].get(name)
if command:
return command
def list_commands(host):
for container in CONTAINERS[host]:
commands = COMMANDS[container]
for name, command in commands.items():
yield name, command
class Dispatchable(type):
# pylint: disable=no-value-for-parameter
def __init__(self, name, bases, namespace):
......@@ -99,6 +114,7 @@ class Dispatchable(type):
if self.AUTOMATIC:
self.enable()
class Host(Dispatchable):
def enable(self):
......@@ -107,6 +123,7 @@ class Host(Dispatchable):
def disable(self):
remove_host(self)
class Container(Dispatchable):
def enable(self):
......
......@@ -34,6 +34,7 @@ code in here will not be executed and commands defined here will not be
detected.
"""
from gajim.common.i18n import _
from gajim.command_system.framework import CommandContainer, command, doc
from gajim.command_system.implementation.hosts import ChatCommands, PrivateChatCommands, GroupChatCommands
......
......@@ -37,6 +37,7 @@ from os.path import expanduser
from gi.repository import GLib
from gajim.common.i18n import _
from gajim.command_system.framework import CommandContainer, command, doc
from gajim.command_system.implementation.hosts import ChatCommands, PrivateChatCommands, GroupChatCommands
......
......@@ -23,6 +23,7 @@ from datetime import date
from gajim import dialogs
from gajim.common import app
from gajim.common import helpers
from gajim.common.i18n import _
from gajim.common.exceptions import GajimGeneralException
from gajim.common.const import KindConstant
......
......@@ -42,7 +42,7 @@ from gajim.common.contacts import LegacyContactsAPI
from gajim.common.events import Events
from gajim.common.css_config import CSSConfig
interface = None # The actual interface (the gtk one for the moment)
interface = None # type: gajim.interface.Interface
thread_interface = lambda *args: None # Interface to run a thread and then a callback
config = c_config.Config()
version = gajim.__version__
......@@ -52,10 +52,10 @@ ipython_window = None
app = None # Gtk.Application
ged = ged_module.GlobalEventsDispatcher() # Global Events Dispatcher
nec = None # Network Events Controller
nec = None # type: gajim.common.nec.NetworkEventsController
plugin_manager = None # Plugins Manager
logger = None
logger = None # type: gajim.common.logger.Logger
# For backwards compatibility needed
# some plugins use that
......@@ -122,7 +122,7 @@ SHOW_LIST = ['offline', 'connecting', 'online', 'chat', 'away', 'xa', 'dnd',
ZEROCONF_ACC_NAME = 'Local'
# These will be set in app.gui_interface.
idlequeue = None
idlequeue = None # type: nbxmpp.idlequeue.IdleQueue
socks5queue = None
gajim_identity = {'type': 'pc', 'category': 'client', 'name': 'Gajim'}
......
......@@ -32,6 +32,7 @@ from gi.repository import GLib
from enum import IntEnum, unique
import gajim
from gajim.common.i18n import _
@unique
......
......@@ -19,82 +19,90 @@
# You should have received a copy of the GNU General Public License
# along with Gajim. If not, see <http://www.gnu.org/licenses/>.
from typing import Dict # pylint: disable=unused-import
from typing import List
from typing import Generator
from typing import Optional # pylint: disable=unused-import
from typing import Tuple
from typing import Union
import os
import sys
import tempfile
from pathlib import Path
import gajim
from gajim.common.i18n import _
from gajim.common.const import PathType, PathLocation
from gajim.common.types import PathTuple
def get(key):
def get(key: str) -> Union[str, List[str]]:
if key == 'PLUGINS_DIRS':
if gajim.IS_FLATPAK:
return ['/app/plugins',
_paths['PLUGINS_BASE']]
else:
return [_paths['PLUGINS_BASE'],
_paths['PLUGINS_USER']]
return [_paths['PLUGINS_BASE'],
_paths['PLUGINS_USER']]
return _paths[key]
def get_paths(type_):
def get_paths(type_: PathType) -> Generator[str, None, None]:
for key, value in _paths.items():
location, path, path_type = value
path_type = value[2]
if type_ != path_type:
continue
yield _paths[key]
def override_path(*args, **kwargs):
_paths._add(*args, **kwargs)
_paths.add(*args, **kwargs)
def set_separation(active: bool):
def set_separation(active: bool) -> None:
_paths.profile_separation = active
def set_profile(profile: str):
def set_profile(profile: str) -> None:
_paths.profile = profile
def set_config_root(config_root: str):
def set_config_root(config_root: str) -> None:
_paths.custom_config_root = config_root
def init():
def init() -> None:
_paths.init()
def create_paths():
def create_paths() -> None:
for path in get_paths(PathType.FOLDER):
if not isinstance(path, Path):
path = Path(path)
path_ = Path(path)
if path.is_file():
print(_('%s is a file but it should be a directory') % path)
if path_.is_file():
print(_('%s is a file but it should be a directory') % path_)
print(_('Gajim will now exit'))
sys.exit()
if not path.exists():
for parent_path in reversed(path.parents):
if not path_.exists():
for parent_path in reversed(path_.parents):
# Create all parent folders
# don't use mkdir(parent=True), as it ignores `mode`
# when creating the parents
if not parent_path.exists():
print(('creating %s directory') % parent_path)
parent_path.mkdir(mode=0o700)
print(('creating %s directory') % path)
path.mkdir(mode=0o700)
print(('creating %s directory') % path_)
path_.mkdir(mode=0o700)
class ConfigPaths:
def __init__(self):
self._paths = {}
def __init__(self) -> None:
self._paths = {} # type: Dict[str, PathTuple]
self.profile = ''
self.profile_separation = False
self.custom_config_root = None
self.custom_config_root = None # type: Optional[str]
if os.name == 'nt':
try:
......@@ -133,23 +141,23 @@ class ConfigPaths:
]
for path in source_paths:
self._add(*path)
self.add(*path)
def __getitem__(self, key):
def __getitem__(self, key: str) -> str:
location, path, _ = self._paths[key]
if location == PathLocation.CONFIG:
return os.path.join(self.config_root, path)
elif location == PathLocation.CACHE:
if location == PathLocation.CACHE:
return os.path.join(self.cache_root, path)
elif location == PathLocation.DATA:
if location == PathLocation.DATA:
return os.path.join(self.data_root, path)
return path
def items(self):
def items(self) -> Generator[Tuple[str, PathTuple], None, None]:
for key, value in self._paths.items():
yield (key, value)
def _prepare(self, path, unique):
def _prepare(self, path: str, unique: bool) -> str:
if os.name == 'nt':
path = path.capitalize()
if self.profile:
......@@ -157,7 +165,12 @@ class ConfigPaths:
return '%s.%s' % (path, self.profile)
return path
def _add(self, name, path, location=None, path_type=None, unique=False):
def add(self,
name: str,
path: str,
location: PathLocation = None,
path_type: PathType = None,
unique: bool = False) -> None:
if path and location is not None:
path = self._prepare(path, unique)
self._paths[name] = (location, path, path_type)
......@@ -175,7 +188,7 @@ class ConfigPaths:
]
for path in user_dir_paths:
self._add(*path)
self.add(*path)
# These paths are unique per profile
unique_profile_paths = [
......@@ -191,7 +204,7 @@ class ConfigPaths:
]
for path in unique_profile_paths:
self._add(*path, unique=True)
self.add(*path, unique=True)
# These paths are only unique per profile if the commandline arg
# `separate` is passed
......@@ -219,7 +232,7 @@ class ConfigPaths:
]
for path in paths:
self._add(*path)
self.add(*path)
_paths = ConfigPaths()
......
This diff is collapsed.
from enum import IntEnum, Enum, unique
from collections import namedtuple
from gajim.common.i18n import _
Option = namedtuple('Option', 'kind label type value name callback data desc enabledif props')
Option.__new__.__defaults__ = (None,) * len(Option._fields)
......@@ -168,13 +170,14 @@ class PEPHandlerType(IntEnum):
@unique
class PEPEventType(IntEnum):
ACTIVITY = 0
TUNE = 1
MOOD = 2
LOCATION = 3
NICKNAME = 4
AVATAR = 5
ATOM = 6
ABSTRACT = 0
ACTIVITY = 1
TUNE = 2
MOOD = 3
LOCATION = 4
NICKNAME = 5
AVATAR = 6
ATOM = 7
ACTIVITIES = {
......
......@@ -25,6 +25,7 @@ import logging
from gajim.common import app
from gajim.common import exceptions
from gajim.common.i18n import _
_GAJIM_ERROR_IFACE = 'org.gajim.dbus.Error'
......
......@@ -48,6 +48,7 @@ from string import Template
import nbxmpp
from gajim.common.i18n import Q_
from gajim.common.i18n import _
from gajim.common.i18n import ngettext
from gajim.common import configpaths
......
......@@ -132,7 +132,7 @@ if os.name == 'nt':
_localedir = get_locale_dir()
if hasattr(locale, 'bindtextdomain'):
locale.bindtextdomain(DOMAIN, _localedir)
locale.bindtextdomain(DOMAIN, _localedir) # type: ignore
gettext.textdomain(DOMAIN)
gettext.install(DOMAIN, _localedir)
......@@ -142,7 +142,7 @@ try:
except OSError:
_ = gettext.gettext
if gettext._translations:
_translations = list(gettext._translations.values())[0]
if gettext._translations: # type: ignore
_translations = list(gettext._translations.values())[0] # type: ignore
else:
_translations = gettext.NullTranslations()
......@@ -12,14 +12,18 @@
# You should have received a copy of the GNU General Public License
# along with Gajim. If not, see <http://www.gnu.org/licenses/>.
from typing import Any
from typing import Dict # pylint: disable=unused-import
from typing import List
from typing import Tuple
import logging
from typing import List # noqa
from typing import Dict # noqa
from typing import Any # noqa
from importlib import import_module
from pathlib import Path
from unittest.mock import MagicMock
from gajim.common.types import ConnectionT
log = logging.getLogger('gajim.c.m')
ZEROCONF_MODULES = ['adhoc_commands', 'receipts', 'discovery']
......@@ -70,7 +74,7 @@ class ModuleMock:
return MagicMock()
def register(con, *args, **kwargs):
def register(con: ConnectionT, *args: Any, **kwargs: Any) -> None:
if con in _modules:
return
_modules[con.name] = {}
......@@ -83,20 +87,20 @@ def register(con, *args, **kwargs):
_modules[con.name][name] = instance
def register_single(con, instance, name):
def register_single(con: ConnectionT, instance: Any, name: str) -> None:
if con.name not in _modules:
raise ValueError('Unknown account name: %s' % con.name)
_modules[con.name][name] = instance
def unregister(con):
def unregister(con: ConnectionT) -> None:
for instance in _modules[con.name].values():
if hasattr(instance, 'cleanup'):
instance.cleanup()
del _modules[con.name]
def unregister_single(con, name):
def unregister_single(con: ConnectionT, name: str) -> None:
if con.name not in _modules:
return
if name not in _modules[con.name]:
......@@ -111,8 +115,8 @@ def get(account: str, name: str) -> Any:
return ModuleMock(name)
def get_handlers(con):
handlers = []
def get_handlers(con: ConnectionT) -> List[Tuple[Any, ...]]:
handlers = [] # type: List[Tuple[Any, ...]]
for module in _modules[con.name].values():
handlers += module.handlers
return handlers
......@@ -14,24 +14,30 @@
# XEP-0145: Annotations
from typing import Any
from typing import Dict # pylint: disable=unused-import
from typing import List # pylint: disable=unused-import
from typing import Tuple
import logging
import nbxmpp
from gajim.common import app
from gajim.common import helpers
from gajim.common.types import ConnectionT
log = logging.getLogger('gajim.c.m.annotations')
class Annotations:
def __init__(self, con):
def __init__(self, con: ConnectionT) -> None:
self._con = con
self._account = con.name
self._server = self._con.get_own_jid().getDomain()
self.handlers = []
self.annotations = {}
self.handlers = [] # type: List[Tuple[Any, ...]]
self.annotations = {} # type: Dict[str, str]
def get_annotations(self) -> None:
if not app.account_is_connected(self._account):
......@@ -93,5 +99,5 @@ class Annotations:
log.info('Storing rosternotes successful')
def get_instance(*args, **kwargs):
def get_instance(*args: Any, **kwargs: Any) -> Tuple[Annotations, str]:
return Annotations(*args, **kwargs), 'Annotations'
......@@ -14,6 +14,11 @@
# XEP-0163: Personal Eventing Protocol
from typing import Any
from typing import Dict
from typing import List
from typing import Tuple
import logging
import nbxmpp
......@@ -22,12 +27,16 @@ from gajim.common import app
from gajim.common.exceptions import StanzaMalformed
from gajim.common.nec import NetworkIncomingEvent
from gajim.common.const import PEPHandlerType, PEPEventType
from gajim.common.types import ConnectionT
from gajim.common.types import PEPHandlersDict # pylint: disable=unused-import
from gajim.common.types import PEPNotifyCallback
from gajim.common.types import PEPRetractCallback
log = logging.getLogger('gajim.c.m.pep')
class PEP:
def __init__(self, con):
def __init__(self, con: ConnectionT) -> None:
self._con = con
self._account = con.name
......@@ -37,29 +46,40 @@ class PEP:
]
self.supported = False
self._pep_handlers = {}
self._store_publish_modules = []
def pass_disco(self, from_, identities, _features, _data, _node):
self._pep_handlers = {} # type: PEPHandlersDict
self._store_publish_modules = [] # type: List[Any]
def pass_disco(self,
from_: nbxmpp.JID,
identities: List[Dict[str, str]],
_features: List[str],
_data: List[nbxmpp.DataForm],
_node: str) -> None:
for identity in identities:
if identity['category'] == 'pubsub':
if identity.get('type') == 'pep':
log.info('Discovered PEP support: %s', from_)
self.supported = True
def register_pep_handler(self, namespace, notify_handler, retract_handler):
def register_pep_handler(
self,
namespace: str,
notify_handler: PEPNotifyCallback,
retract_handler: PEPRetractCallback) -> None:
if namespace in self._pep_handlers:
self._pep_handlers[namespace].append(
(notify_handler, retract_handler))
else:
self._pep_handlers[namespace] = [(notify_handler, retract_handler)]
if notify_handler:
module_instance = notify_handler.__self__
module_instance = notify_handler.__self__ # type: ignore
if module_instance.store_publish:
if module_instance not in self._store_publish_modules:
self._store_publish_modules.append(module_instance)
def _pep_event_received(self, _con, stanza):
def _pep_event_received(self,
_con: ConnectionT,
stanza: nbxmpp.Message) -> None:
jid = stanza.getFrom()
event = stanza.getTag('event', namespace=nbxmpp.NS_PUBSUB_EVENT)
items = event.getTag('items')
......@@ -98,17 +118,47 @@ class PEP:
handler[PEPHandlerType.NOTIFY](jid, items_[0])
raise nbxmpp.NodeProcessed
def send_stored_publish(self):
def send_stored_publish(self) -> None:
for module in self._store_publish_modules:
module.send_stored_publish()
def reset_stored_publish(self):
def reset_stored_publish(self) -> None:
for module in self._store_publish_modules:
module.reset_stored_publish()
class AbstractPEPData:
type_ = PEPEventType.ABSTRACT
def __init__(self, data: Any) -> None:
self.data = data
def as_markup_text(self) -> str: # pylint: disable=no-self-use
'''SHOULD be implemented by subclasses'''
return ''
def __eq__(self, other: Any) -> bool:
return other == self.type_
def __bool__(self) -> bool:
return self.data is not None
def __str__(self) -> str:
return str(self.data)
class AbstractPEPModule:
def __init__(self, con, account):
name = ''
namespace = ''
pep_class = AbstractPEPData
store_publish = True
_log = log
def __init__(self,
con: ConnectionT,
account: str) -> None:
self._account = account
self._con = con
......@@ -119,7 +169,7 @@ class AbstractPEPModule:
self._pep_notify_received,
self._pep_retract_received)
def _pep_notify_received(self, jid, item):
def _pep_notify_received(self, jid: nbxmpp.JID, item: nbxmpp.Node) -> None:
try:
data = self._extract_info(item)
except StanzaMalformed as error:
......@@ -129,19 +179,19 @@ class AbstractPEPModule:
self._log.info('Received: %s %s', jid, data)
self._push_event(jid, self.pep_class(data))
def _pep_retract_received(self, jid, id_):
def _pep_retract_received(self, jid: nbxmpp.JID, id_: str) -> None:
self._log.info('Retract: %s %s', jid, id_)
self._push_event(jid, self.pep_class(None))
def _extract_info(self, item):
def _extract_info(self, item: nbxmpp.Node) -> Any:
'''To be implemented by subclasses'''
raise NotImplementedError
def _build_node(self, data):
def _build_node(self, data: Any) -> nbxmpp.Node:
'''To be implemented by subclasses'''
raise NotImplementedError
def _push_event(self, jid, user_pep):
def _push_event(self, jid: nbxmpp.JID, user_pep: Any) -> None:
self._notification_received(jid, user_pep)
app.nec.push_incoming_event(
PEPReceivedEvent(None, conn=self._con,
......@@ -149,7 +199,7 @@ class AbstractPEPModule:
pep_type=self.name,
user_pep=user_pep))
def _notification_received(self, jid, user_pep):