*: Fix all typing hint related errors

- Try to mark class variables in component classes.

- Leave typing hints generic, such as 'list' and 'dict' where content is usually
not filled, too complex, or context is unimportant.

- backups: Handle failure for tarfile extraction so that methods are not called
on potentially None valued variables.

- backups: Prevent potentially passing a keyword argument twice.

- dynamicdns: Deal properly with outcome of urlparsing.

- ejabberd: Deal with failed regex match

- email: Fix a mypy compliant when iterating a filtered list.

- tor: Don't reuse variables for different typed values.

- tor: Don't reuse variables for different typed values.

- operation: Return None explicitly.

- operation: Ensure that keyword argument is not repeated.

Tests:

- Where only typing hints were modified and no syntax error came up, additional
testing was not done.

- `mypy --ignore-missing-imports .` run successfully.

- Generate developer documentation.

- Service runs without errors upon start up.

- backups: Listing and restoring specific apps from a backup works.

- backups: Mounting a remote backup repository works.

- NOT TESTED: dynamicdns: Migrating from old style configuration works.

- ejabberd: Verify that setting coturn configuration works.

- email: Test that showing configuration from postfix works.

- tor: Orport value is properly shown.

- transmission: Configuration values are properly set.

- users: Running unit tests as root works.

- operation: Operation status messages are show properly during app install.

- ./setup.py install runs

Signed-off-by: Sunil Mohan Adapa <sunil@medhas.org>
Reviewed-by: James Valleroy <jvalleroy@mailbox.org>
This commit is contained in:
Sunil Mohan Adapa 2023-09-23 18:25:19 -07:00 committed by James Valleroy
parent a709f3a6a8
commit 2dd00a8f08
No known key found for this signature in database
GPG Key ID: 77C0C75E7B650808
43 changed files with 111 additions and 87 deletions

View File

@ -111,7 +111,7 @@ htmlhelp_basename = 'FreedomBoxdoc'
# -- Options for LaTeX output ------------------------------------------------
latex_elements = {
latex_elements: dict = {
# The paper size ('letterpaper' or 'a4paper').
#
# 'papersize': 'letterpaper',

View File

@ -7,6 +7,7 @@ import collections
import enum
import inspect
import logging
from typing import ClassVar
from plinth import cfg
from plinth.signals import post_app_loading
@ -39,14 +40,15 @@ class App:
"""
app_id = None
app_id: str | None = None
can_be_disabled = True
can_be_disabled: bool = True
locked = False # Whether user interaction with the app is allowed.
locked: bool = False # Whether user interaction with the app is allowed.
# XXX: Lockdown the application UI by implementing a middleware
_all_apps = collections.OrderedDict()
_all_apps: ClassVar[collections.OrderedDict[
str, 'App']] = collections.OrderedDict()
class SetupState(enum.Enum):
"""Various states of app being setup."""

View File

@ -4,6 +4,7 @@
import json
import logging
import pathlib
from typing import ClassVar
from plinth import app, cfg
from plinth.modules.users import privileged as users_privileged
@ -14,7 +15,7 @@ logger = logging.getLogger(__name__)
class Shortcut(app.FollowerComponent):
"""An application component for handling shortcuts."""
_all_shortcuts = {}
_all_shortcuts: ClassVar[dict[str, 'Shortcut']] = {}
def __init__(self, component_id, name, short_description=None, icon=None,
url=None, description=None, manual_page=None,

View File

@ -1,5 +1,7 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
from typing import ClassVar
from django.urls import reverse_lazy
from plinth import app
@ -8,7 +10,7 @@ from plinth import app
class Menu(app.FollowerComponent):
"""Component to manage a single menu item."""
_all_menus = set()
_all_menus: ClassVar[set['Menu']] = set()
def __init__(self, component_id, name=None, short_description=None,
icon=None, url_name=None, url_args=None, url_kwargs=None,

View File

@ -17,7 +17,7 @@ class Migration(migrations.Migration):
initial = True
dependencies = []
dependencies: list = []
operations = [
migrations.CreateModel(

View File

@ -3,4 +3,4 @@
URLs for the Apache module.
"""
urlpatterns = []
urlpatterns: list = []

View File

@ -7,4 +7,4 @@ Application manifest for avahi.
# /etc/avahi/services. Currently, we don't intend to make that customizable.
# There is no necessity for backup and restore. This manifest will ensure that
# avahi enable/disable setting is preserved.
backup = {}
backup: dict = {}

View File

@ -6,4 +6,4 @@ Application manifest for backups.
# Currently, backup application does not have any settings. However, settings
# such as scheduler settings, backup location, secrets to connect to remove
# servers need to be backed up.
backup = {}
backup: dict = {}

View File

@ -31,7 +31,7 @@ def mount(mountpoint: str, remote_path: str, ssh_keyfile: Optional[str] = None,
except AlreadyMountedError:
return
kwargs = {}
input_ = None
# the shell would expand ~/ to the local home directory
remote_path = remote_path.replace('~/', '').replace('~', '')
# 'reconnect', 'ServerAliveInternal' and 'ServerAliveCountMax' allow the
@ -55,9 +55,9 @@ def mount(mountpoint: str, remote_path: str, ssh_keyfile: Optional[str] = None,
if not password:
raise ValueError('mount requires either a password or ssh_keyfile')
cmd += ['-o', 'password_stdin']
kwargs['input'] = password.encode()
input_ = password.encode()
subprocess.run(cmd, check=True, timeout=TIMEOUT, **kwargs)
subprocess.run(cmd, check=True, timeout=TIMEOUT, input=input_)
@privileged
@ -266,7 +266,12 @@ def get_exported_archive_apps(path: str) -> list[str]:
for name in filenames:
if 'var/lib/plinth/backups-manifests/' in name \
and name.endswith('.json'):
manifest_data = tar_handle.extractfile(name).read()
file_handle = tar_handle.extractfile(name)
if not file_handle:
raise RuntimeError(
'Unable to extract app manifest from backup file.')
manifest_data = file_handle.read()
manifest = json.loads(manifest_data)
break

View File

@ -72,9 +72,9 @@ KNOWN_ERRORS = [
class BaseBorgRepository(abc.ABC):
"""Base class for all kinds of Borg repositories."""
flags = {}
flags: dict[str, bool] = {}
is_mounted = True
known_credentials = []
known_credentials: list[str] = []
def __init__(self, path, credentials=None, uuid=None, schedule=None,
**kwargs):

View File

@ -17,4 +17,4 @@ clients = [{
# triggered on every Plinth domain change (and cockpit application install) and
# will set the value of allowed domains correctly. This is the only key the is
# customized in cockpit.conf.
backup = {}
backup: dict = {}

View File

@ -2,8 +2,6 @@
"""App component for other apps to manage their STUN/TURN server configuration.
"""
from __future__ import annotations # Can be removed in Python 3.10
import base64
import hashlib
import hmac
@ -11,6 +9,7 @@ import json
import re
from dataclasses import dataclass, field
from time import time
from typing import ClassVar, Iterable
from plinth import app
@ -36,9 +35,9 @@ class TurnConfiguration:
that must be used by a STUN/TURN client after advice from the server.
"""
domain: str = None
domain: str | None = None
uris: list[str] = field(default_factory=list)
shared_secret: str = None
shared_secret: str | None = None
def __post_init__(self):
"""Generate URIs after object initialization if necessary."""
@ -76,8 +75,8 @@ class UserTurnConfiguration(TurnConfiguration):
time.
"""
username: str = None
credential: str = None
username: str | None = None
credential: str | None = None
def to_json(self) -> str:
"""Return a JSON representation of the configuration."""
@ -102,7 +101,7 @@ class TurnConsumer(app.FollowerComponent):
"""
_all = {}
_all: ClassVar[dict[str, 'TurnConsumer']] = {}
def __init__(self, component_id):
"""Initialize the component.
@ -115,7 +114,7 @@ class TurnConsumer(app.FollowerComponent):
self._all[component_id] = self
@classmethod
def list(cls) -> list[TurnConsumer]: # noqa
def list(cls) -> Iterable['TurnConsumer']:
"""Return a list of all Coturn components."""
return cls._all.values()

View File

@ -3,4 +3,4 @@
Application manifest for diagnostics.
"""
backup = {}
backup: dict = {}

View File

@ -3,6 +3,7 @@
import pathlib
import urllib
from typing import Any
from plinth.actions import privileged
@ -30,7 +31,7 @@ def _read_configuration(path, separator='='):
@privileged
def export_config() -> dict[str, object]:
def export_config() -> dict[str, bool | dict[str, dict[str, str | None]]]:
"""Return the old ez-ipupdate configuration in JSON format."""
input_config = {}
if _active_config.exists():
@ -68,12 +69,12 @@ def export_config() -> dict[str, object]:
update_url = domain['update_url']
try:
server = urllib.parse.urlparse(update_url).netloc
service_types = {
service_types: dict[str, str] = {
'dynupdate.noip.com': 'noip.com',
'dynupdate.no-ip.com': 'noip.com',
'freedns.afraid.org': 'freedns.afraid.org'
}
domain['service_type'] = service_types.get(server, 'other')
domain['service_type'] = service_types.get(str(server), 'other')
except ValueError:
pass
@ -86,7 +87,7 @@ def export_config() -> dict[str, object]:
and _active_config.exists()):
enabled = True
output_config = {'enabled': enabled, 'domains': {}}
output_config: dict[str, Any] = {'enabled': enabled, 'domains': {}}
if domain['domain']:
output_config['domains'][domain['domain']] = domain

View File

@ -28,7 +28,7 @@ MOD_IRC_DEPRECATED_VERSION = Version('18.06')
yaml = YAML()
yaml.allow_duplicate_keys = True
yaml.preserve_quotes = True
yaml.preserve_quotes = True # type: ignore [assignment]
TURN_URI_REGEX = r'(stun|turn):(.*):([0-9]{4})\?transport=(tcp|udp)'
@ -286,7 +286,11 @@ def mam(command: str) -> Optional[bool]:
def _generate_service(uri: str) -> dict:
"""Generate ejabberd mod_stun_disco service config from Coturn URI."""
pattern = re.compile(TURN_URI_REGEX)
typ, domain, port, transport = pattern.match(uri).groups()
match = pattern.match(uri)
if not match:
raise ValueError('URL does not match TURN URI')
typ, domain, port, transport = match.groups()
return {
"host": domain,
"port": int(port),

View File

@ -23,7 +23,7 @@ class Service: # NOQA, pylint: disable=too-many-instance-attributes
wakeup: str
maxproc: str
command: str
options: str
options: dict[str, str]
def __str__(self) -> str:
parts = [
@ -49,7 +49,8 @@ def get_config(keys: list) -> dict:
output = _run(args)
result = {}
for line in filter(None, output.split('\n')):
lines: list[str] = list(filter(None, output.split('\n')))
for line in lines:
key, sep, value = line.partition('=')
if not sep:
raise ValueError('Invalid output detected')

View File

@ -32,7 +32,7 @@ default_config = {
])
}
submission_options = {
submission_options: dict[str, str] = {
'syslog_name': 'postfix/submission',
'smtpd_tls_security_level': 'encrypt',
'smtpd_client_restrictions': 'permit_sasl_authenticated,reject',
@ -43,7 +43,7 @@ submission_service = postconf.Service(service='submission', type_='inet',
wakeup='-', maxproc='-', command='smtpd',
options=submission_options)
smtps_options = {
smtps_options: dict[str, str] = {
'syslog_name': 'postfix/smtps',
'smtpd_tls_wrappermode': 'yes',
'smtpd_sasl_auth_enable': 'yes',

View File

@ -27,7 +27,7 @@ _description = [
'security threat from the Internet.'), box_name=cfg.box_name)
]
_port_details = {}
_port_details: dict[str, list[str]] = {}
logger = logging.getLogger(__name__)

View File

@ -5,6 +5,7 @@ App component for other apps to use firewall functionality.
import logging
import re
from typing import ClassVar
from django.utils.text import format_lazy
from django.utils.translation import gettext_lazy as _
@ -18,7 +19,7 @@ logger = logging.getLogger(__name__)
class Firewall(app.FollowerComponent):
"""Component to open/close firewall ports for an app."""
_all_firewall_components = {}
_all_firewall_components: ClassVar[dict[str, 'Firewall']] = {}
def __init__(self, component_id, name=None, ports=None, is_external=False):
"""Initialize the firewall component."""

View File

@ -3,4 +3,4 @@
Application manifest for firewall.
"""
backup = {}
backup: dict = {}

View File

@ -11,4 +11,4 @@ clients = [{
}]
}]
backup = {}
backup: dict = {}

View File

@ -11,4 +11,4 @@ clients = [{
}]
}]
backup = {}
backup: dict = {}

View File

@ -4,6 +4,7 @@
import logging
import pathlib
import threading
from typing import ClassVar
from plinth import app
from plinth.modules.names.components import DomainName
@ -38,7 +39,7 @@ class LetsEncrypt(app.FollowerComponent):
"""
_all = {}
_all: ClassVar[dict[str, 'LetsEncrypt']] = {}
def __init__(self, component_id, domains=None, daemons=None,
should_copy_certificates=False, private_key_path=None,

View File

@ -3,7 +3,6 @@
import logging
import os
from typing import List
from django.urls import reverse_lazy
from django.utils.translation import gettext_lazy as _
@ -191,7 +190,7 @@ def get_configured_domain_name():
return config['server_name']
def get_turn_configuration() -> (List[str], str, bool):
def get_turn_configuration() -> tuple[TurnConfiguration, bool]:
"""Return TurnConfiguration if setup else empty."""
for file_path, managed in ((privileged.OVERRIDDEN_TURN_CONF_PATH, False),
(privileged.TURN_CONF_PATH, True)):

View File

@ -3,6 +3,8 @@
App component to introduce a new domain type.
"""
from typing import ClassVar
from django.utils.translation import gettext_lazy as _
from plinth import app
@ -39,7 +41,7 @@ class DomainType(app.FollowerComponent):
"""
_all = {}
_all: ClassVar[dict[str, 'DomainType']] = {}
def __init__(self, component_id, display_name, configuration_url,
can_have_certificate=True):
@ -90,7 +92,7 @@ class DomainName(app.FollowerComponent):
primary reason for making a domain name available as a component.
"""
_all = {}
_all: ClassVar[dict[str, 'DomainName']] = {}
def __init__(self, component_id, name, domain_type, services):
"""Initialize a domain name.

View File

@ -3,4 +3,4 @@
Application manifest for names.
"""
backup = {}
backup: dict = {}

View File

@ -2,7 +2,6 @@
"""Configure PageKite."""
import os
from typing import Union
import augeas
@ -118,7 +117,7 @@ def set_config(frontend: str, kite_name: str, kite_secret: str):
@privileged
def remove_service(service: dict[str, Union[str, bool]]):
def remove_service(service: dict[str, str]):
"""Search and remove the service(s) that match all given parameters."""
aug = _augeas_load()
service = utils.load_service(service)
@ -171,7 +170,7 @@ def _add_service(aug, service):
@privileged
def add_service(service: dict[str, Union[str, bool]]):
def add_service(service: dict[str, str]):
"""Add one service."""
aug = _augeas_load()
service = utils.load_service(service)

View File

@ -13,4 +13,4 @@ clients = [{
}]
}]
backup = {}
backup: dict = {}

View File

@ -3,4 +3,4 @@
Application manifest for power.
"""
backup = {}
backup: dict = {}

View File

@ -3,4 +3,4 @@
Application manifest for privoxy.
"""
backup = {}
backup: dict = {}

View File

@ -3,4 +3,4 @@
Application manifest for storage.
"""
backup = {}
backup: dict = {}

View File

@ -14,7 +14,7 @@ gio = import_from_gi('Gio', '2.0')
_DBUS_NAME = 'org.freedesktop.UDisks2'
_INTERFACES = {
_INTERFACES: dict[str, str] = {
'Ata': 'org.freedesktop.UDisks2.Drive.Ata',
'Block': 'org.freedesktop.UDisks2.Block',
'Drive': 'org.freedesktop.UDisks2.Drive',
@ -28,14 +28,14 @@ _INTERFACES = {
'UDisks2': 'org.freedesktop.UDisks2',
}
_OBJECTS = {
_OBJECTS: dict[str, str] = {
'drives': '/org/freedesktop/UDisks2/drives/',
'jobs': '/org/freedesktop/UDisks2/jobs/',
'Manager': '/org/freedesktop/UDisks2/Manager',
'UDisks2': '/org/freedesktop/UDisks2',
}
_ERRORS = {
_ERRORS: dict[str, str] = {
'AlreadyMounted': 'org.freedesktop.UDisks2.Error.AlreadyMounted',
'Failed': 'org.freedesktop.UDisks2.Error.Failed',
}
@ -54,8 +54,8 @@ def _get_dbus_proxy(object_, interface):
class Proxy:
"""Base methods for abstraction over UDisks2 DBus proxy objects."""
interface = None
properties = {}
interface: str | None = None
properties: dict[str, tuple[str, str]] = {}
def __init__(self, object_path):
"""Return an object instance."""

View File

@ -254,8 +254,8 @@ def _get_ports() -> dict[str, str]:
def _get_orport() -> str:
"""Return the ORPort by querying running instance."""
cookie = open(TOR_AUTH_COOKIE, 'rb').read()
cookie = codecs.encode(cookie, 'hex').decode()
cookie_bytes = open(TOR_AUTH_COOKIE, 'rb').read()
cookie = codecs.encode(cookie_bytes, 'hex').decode()
commands = '''AUTHENTICATE {cookie}
GETINFO net/listeners/or
@ -270,6 +270,9 @@ QUIT
line = response.split(b'\r\n')[1].decode()
matches = re.match(r'.*=".+:(\d+)"', line)
if not matches:
raise ValueError('Invalid orport value returned by Tor')
return matches.group(1)

View File

@ -22,12 +22,13 @@ def get_configuration() -> dict[str, str]:
@privileged
def merge_configuration(configuration: dict[str, Union[str, bool]]):
"""Merge given JSON configuration with existing configuration."""
current_configuration = _transmission_config.read_bytes()
current_configuration = json.loads(current_configuration)
current_configuration_bytes = _transmission_config.read_bytes()
current_configuration = json.loads(current_configuration_bytes)
new_configuration = current_configuration
new_configuration.update(configuration)
new_configuration = json.dumps(new_configuration, indent=4, sort_keys=True)
new_configuration_bytes = json.dumps(new_configuration, indent=4,
sort_keys=True)
_transmission_config.write_text(new_configuration, encoding='utf-8')
_transmission_config.write_text(new_configuration_bytes, encoding='utf-8')
action_utils.service_reload('transmission-daemon')

View File

@ -4,6 +4,7 @@ App component to manage users and groups.
"""
import itertools
from typing import ClassVar
from plinth import app
@ -12,7 +13,7 @@ class UsersAndGroups(app.FollowerComponent):
"""Component to manage users and groups of an app."""
# Class variable to hold a list of user groups for apps
_all_components = set()
_all_components: ClassVar[set['UsersAndGroups']] = set()
def __init__(self, component_id, reserved_usernames=[], groups={}):
"""Store reserved_usernames and groups of the app.

View File

@ -16,11 +16,6 @@ from plinth import action_utils
from plinth.modules.users import privileged
from plinth.tests import config as test_config
pytestmark = pytest.mark.usefixtures('mock_privileged')
privileged_modules_to_mock = [
'plinth.modules.users.privileged', 'plinth.modules.security.privileged'
]
_cleanup_users = None
_cleanup_groups = None
@ -39,10 +34,13 @@ def _is_ldap_set_up():
return False
pytestmark = [
pytest.mark.usefixtures('needs_root', 'load_cfg'),
pytestmark: list[pytest.MarkDecorator] = [
pytest.mark.usefixtures('needs_root', 'load_cfg', 'mock_privileged'),
pytest.mark.skipif(not _is_ldap_set_up(), reason="LDAP is not configured")
]
privileged_modules_to_mock = [
'plinth.modules.users.privileged', 'plinth.modules.security.privileged'
]
def _random_string(length=8):

View File

@ -19,7 +19,7 @@ def get_info() -> dict[str, dict]:
if not line:
continue
fields = [
fields: list = [
field if field != '(none)' else None for field in line.split()
]
interface_name = fields[0]

View File

@ -89,10 +89,10 @@ class Operation:
return self.return_value
@staticmethod
def get_operation():
def get_operation() -> 'Operation':
"""Return the operation associated with this thread."""
thread = threading.current_thread()
return thread._operation
return thread._operation # type: ignore [attr-defined]
def on_update(self, message: Optional[str] = None,
exception: Optional[Exception] = None):
@ -106,7 +106,7 @@ class Operation:
self._update_notification()
@property
def message(self):
def message(self) -> str | None:
"""Return a message about status of the operation."""
from django.utils.translation import gettext_noop
if self._message: # Progress has been set by the operation itself
@ -124,6 +124,8 @@ class Operation:
if self.state == Operation.State.COMPLETED:
return gettext_noop('Finished: {name}')
return None
@property
def translated_message(self):
"""Return a message about status of operation after translating.
@ -183,8 +185,8 @@ class OperationsManager:
def new(self, *args, **kwargs):
"""Create a new operation instance and add to global list."""
with self._lock:
operation = Operation(*args, **kwargs,
on_complete=self._on_operation_complete)
kwargs['on_complete'] = self._on_operation_complete
operation = Operation(*args, **kwargs)
self._operations.append(operation)
logger.info('%s: added', operation)
self._schedule_next()

View File

@ -124,7 +124,7 @@ LOGIN_URL = 'users:login'
LOGIN_REDIRECT_URL = 'index'
# Overridden before initialization
MESSAGE_TAGS = {}
MESSAGE_TAGS: dict = {}
MIDDLEWARE = (
'django.middleware.security.SecurityMiddleware',

View File

@ -92,7 +92,7 @@ class TestYAMLFileUtil:
kv_pair = {'key': 'value'}
yaml = ruamel.yaml.YAML()
yaml.preserve_quotes = True
yaml.preserve_quotes = True # type: ignore [assignment]
def test_update_empty_yaml_file(self):
"""

View File

@ -9,6 +9,7 @@ import urllib.parse
from django.contrib import messages
from django.core.exceptions import ImproperlyConfigured
from django.forms import Form
from django.http import Http404, HttpResponseBadRequest, HttpResponseRedirect
from django.shortcuts import redirect
from django.template.response import TemplateResponse
@ -164,9 +165,9 @@ class AppView(FormView):
instead of the simple appearance provided by default.
"""
form_class = None
app_id = None
template_name = 'app.html'
form_class: Form | None = None
app_id: str | None = None
template_name: str = 'app.html'
def __init__(self, *args, **kwargs):
"""Initialize the view."""

View File

@ -7,6 +7,7 @@ import logging
import os
import sys
import warnings
from typing import ClassVar
import cherrypy
@ -104,7 +105,7 @@ class StaticFiles(app_module.FollowerComponent):
"""
_all_instances = {}
_all_instances: ClassVar[dict[str, 'StaticFiles']] = {}
def __init__(self, component_id, directory_map=None):
"""Initialize the component.

View File

@ -66,7 +66,7 @@ LOCALE_PATHS = ['plinth/locale']
class DjangoCommand(Command):
"""Setup command to run a Django management command."""
user_options = []
user_options: list = []
def initialize_options(self):
"""Declare the options for this command."""