Merge pull request #3468 from modoboa/replaced-pylint-by-ruff

Replaced pylint by ruff
This commit is contained in:
Antoine Nguyen
2025-02-18 16:42:10 +01:00
committed by GitHub
207 changed files with 773 additions and 1568 deletions
+4 -4
View File
@@ -3,12 +3,12 @@
repos:
- repo: local
hooks:
- id: pylint
name: pylint
entry: pylint
- id: ruff
name: ruff
entry: ruff
language: system
types: [python]
args: ["-rn", "-sn", "--fail-under=5"]
args: ["check"]
- id: prettier
name: prettier
entry: prettier --config frontend/.prettierrc.json --write --ignore-unknown
+11 -8
View File
@@ -2,7 +2,6 @@
from typing import List
from django.conf import settings
from django.contrib.auth import password_validation
from django.core.exceptions import ValidationError
from django.shortcuts import get_object_or_404
@@ -179,7 +178,7 @@ class DomainAliasSerializer(serializers.ModelSerializer):
object_type="domain_aliases",
)
except lib_exceptions.ModoboaException as inst:
raise serializers.ValidationError({"domain": force_str(inst)})
raise serializers.ValidationError({"domain": force_str(inst)}) from None
domain_alias.save(creator=creator)
return domain_alias
@@ -312,7 +311,7 @@ class AccountPasswordSerializer(serializers.ModelSerializer):
try:
password_validation.validate_password(value, self.instance)
except ValidationError as exc:
raise serializers.ValidationError(exc.messages[0])
raise serializers.ValidationError(exc.messages[0]) from None
return value
def update(self, instance, validated_data):
@@ -364,7 +363,7 @@ class WritableAccountSerializer(AccountSerializer):
try:
validators.UTF8EmailValidator()(username)
except ValidationError as err:
raise ValidationError({"username": err.message})
raise ValidationError({"username": err.message}) from None
mailbox = data.get("mailbox")
if mailbox is None:
if not self.instance:
@@ -397,7 +396,9 @@ class WritableAccountSerializer(AccountSerializer):
object_type="mailboxes",
)
except lib_exceptions.ModoboaException as inst:
raise serializers.ValidationError({"mailbox": force_str(inst)})
raise serializers.ValidationError(
{"mailbox": force_str(inst)}
) from None
condition = not data.get("random_password") and (
data.get("password") or not self.partial
)
@@ -409,7 +410,9 @@ class WritableAccountSerializer(AccountSerializer):
data["password"], self.instance
)
except ValidationError as exc:
raise serializers.ValidationError({"password": exc.messages[0]})
raise serializers.ValidationError(
{"password": exc.messages[0]}
) from None
elif not self.instance:
raise serializers.ValidationError(
{"password": _("This field is required.")}
@@ -524,9 +527,9 @@ class AliasSerializer(serializers.ModelSerializer):
value, self.context["request"].user, instance=self.instance
)
except ValidationError as err:
raise serializers.ValidationError(err)
raise serializers.ValidationError(err) from None
except AliasExists:
raise serializers.ValidationError(_("This alias already exists"))
raise serializers.ValidationError(_("This alias already exists")) from None
return value.lower()
def create(self, validated_data):
+9 -9
View File
@@ -161,7 +161,7 @@ class DomainAliasAPITestCase(ModoAPITestCase):
@classmethod
def setUpTestData(cls): # NOQA:N802
"""Create test data."""
super(DomainAliasAPITestCase, cls).setUpTestData()
super().setUpTestData()
factories.populate_database()
cls.dom_alias1 = factories.DomainAliasFactory(
name="dalias1.com", target__name="test.com"
@@ -185,7 +185,7 @@ class DomainAliasAPITestCase(ModoAPITestCase):
self.assertEqual(response.data["name"], "dalias1.com")
url = reverse("v1:domain_alias-list")
response = self.client.get("{}?domain=test.com".format(url))
response = self.client.get(f"{url}?domain=test.com")
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data), 1)
@@ -246,7 +246,7 @@ class AccountAPITestCase(ModoAPITestCase):
@classmethod
def setUpTestData(cls): # NOQA:N802
"""Create test data."""
super(AccountAPITestCase, cls).setUpTestData()
super().setUpTestData()
factories.populate_database()
cls.da_token = Token.objects.create(
user=core_models.User.objects.get(username="admin@test.com")
@@ -271,17 +271,17 @@ class AccountAPITestCase(ModoAPITestCase):
response = response.json()
self.assertEqual(len(response), 5)
response = self.client.get("{}?domain=test.com".format(url))
response = self.client.get(f"{url}?domain=test.com")
self.assertEqual(response.status_code, 200)
response = response.json()
self.assertEqual(len(response), 2)
response = self.client.get("{}?domain=pouet.com".format(url))
response = self.client.get(f"{url}?domain=pouet.com")
self.assertEqual(response.status_code, 200)
response = response.json()
self.assertEqual(len(response), 0)
response = self.client.get("{}?search=user@test.com".format(url))
response = self.client.get(f"{url}?search=user@test.com")
self.assertEqual(response.status_code, 200)
response = response.json()
self.assertEqual(len(response), 1)
@@ -608,7 +608,7 @@ class AliasAPITestCase(ModoAPITestCase):
@classmethod
def setUpTestData(cls): # NOQA:N802
"""Create test data."""
super(AliasAPITestCase, cls).setUpTestData()
super().setUpTestData()
cls.localconfig.parameters.set_value("enable_admin_limits", False, app="limits")
cls.localconfig.save()
factories.populate_database()
@@ -624,7 +624,7 @@ class AliasAPITestCase(ModoAPITestCase):
response = response.json()
self.assertEqual(len(response), 3)
response = self.client.get("{}?domain=test.com".format(url))
response = self.client.get(f"{url}?domain=test.com")
self.assertEqual(response.status_code, 200)
response = response.json()
self.assertEqual(len(response), 3)
@@ -744,7 +744,7 @@ class SenderAddressAPITestCase(ModoAPITestCase):
@classmethod
def setUpTestData(cls): # NOQA:N802
"""Create test data."""
super(SenderAddressAPITestCase, cls).setUpTestData()
super().setUpTestData()
cls.localconfig.parameters.set_value("enable_admin_limits", False, app="limits")
cls.localconfig.save()
factories.populate_database()
+1 -1
View File
@@ -135,7 +135,7 @@ class AccountViewSet(
try:
user = core_models.User.objects.get(pk=pk)
except core_models.User.DoesNotExist:
raise http.Http404
raise http.Http404 from None
serializer = self.get_serializer(user, data=request.data)
if serializer.is_valid():
serializer.save()
+10 -12
View File
@@ -94,7 +94,7 @@ class DomainSerializer(v1_serializers.DomainSerializer):
return domain
# 1. Create a domain administrator
username = "%s@%s" % (domain_admin["username"], domain.name)
username = "{}@{}".format(domain_admin["username"], domain.name)
try:
da = core_models.User.objects.get(username=username)
except core_models.User.DoesNotExist:
@@ -136,7 +136,7 @@ class DomainSerializer(v1_serializers.DomainSerializer):
core_signals.can_create_object.send(
self.__class__, context=user, klass=models.Alias
)
address = "postmaster@{}".format(domain.name)
address = f"postmaster@{domain.name}"
alias = models.Alias.objects.create(
address=address, domain=domain, enabled=True
)
@@ -240,13 +240,11 @@ class AdminGlobalParametersSerializer(serializers.Serializer):
if not value:
return value
try:
ip_addresses = [
ipaddress.ip_network(v.strip()) for v in value.split() if v.strip()
]
[ipaddress.ip_network(v.strip()) for v in value.split() if v.strip()]
except ValueError:
raise serializers.ValidationError(
_("This field only allows valid IP addresses (or networks)")
)
) from None
return value
def validate_auto_create_domain_and_mailbox(self, value):
@@ -366,9 +364,7 @@ class AccountMeSerializer(v1_serializers.AccountSerializer):
permissions = (
user.user_permissions.all() | Permission.objects.filter(group__user=user)
).select_related("content_type")
return [
"{}.{}".format(p.content_type.app_label, p.codename) for p in permissions
]
return [f"{p.content_type.app_label}.{p.codename}" for p in permissions]
class AccountSerializer(v1_serializers.AccountSerializer):
@@ -483,7 +479,7 @@ class WritableAccountSerializer(v1_serializers.WritableAccountSerializer):
try:
validators.UTF8EmailValidator()(username)
except ValidationError as err:
raise ValidationError({"username": err.message})
raise ValidationError({"username": err.message}) from None
mailbox = data.get("mailbox")
if mailbox is None:
if not self.instance:
@@ -505,7 +501,7 @@ class WritableAccountSerializer(v1_serializers.WritableAccountSerializer):
object_type="mailboxes",
)
except lib_exceptions.ModoboaException as inst:
raise serializers.ValidationError({"mailbox": str(inst)})
raise serializers.ValidationError({"mailbox": str(inst)}) from None
if len(self.address) > 64:
raise serializers.ValidationError(
{
@@ -522,7 +518,9 @@ class WritableAccountSerializer(v1_serializers.WritableAccountSerializer):
data["password"], self.instance
)
except ValidationError as exc:
raise serializers.ValidationError({"password": exc.messages[0]})
raise serializers.ValidationError(
{"password": exc.messages[0]}
) from None
elif not self.instance:
raise serializers.ValidationError(
{"password": _("This field is required.")}
+4 -6
View File
@@ -212,7 +212,7 @@ class AdminParametersForm(param_forms.AdminParametersForm):
dpath = fpath
if dpath:
try:
code, version = exec_cmd("%s --version" % dpath)
code, version = exec_cmd(f"{dpath} --version")
except OSError:
hide_fields = True
else:
@@ -255,18 +255,16 @@ class AdminParametersForm(param_forms.AdminParametersForm):
if not value:
return value
try:
ip_addresses = [
ipaddress.ip_network(v.strip()) for v in value.split() if v.strip()
]
[ipaddress.ip_network(v.strip()) for v in value.split() if v.strip()]
except ValueError:
raise forms.ValidationError(
gettext_lazy("This field only allows valid IP addresses (or networks)")
)
) from None
return value
def clean(self):
"""Check MX options."""
cleaned_data = super(AdminParametersForm, self).clean()
cleaned_data = super().clean()
condition = (
cleaned_data.get("enable_mx_checks")
and cleaned_data.get("domains_must_have_authorized_mx")
+1 -1
View File
@@ -15,7 +15,7 @@ class DomainFactory(PermissionFactory):
model = models.Domain
django_get_or_create = ("name",)
name = factory.Sequence(lambda n: "domain{}.test".format(n))
name = factory.Sequence(lambda n: f"domain{n}.test")
type = "domain" # NOQA:A003
quota = 0
default_mailbox_quota = 10
+16 -16
View File
@@ -71,7 +71,7 @@ class AccountFormGeneral(forms.ModelForm):
labels = {"is_active": gettext_lazy("Enabled")}
def __init__(self, user, *args, **kwargs):
super(AccountFormGeneral, self).__init__(*args, **kwargs)
super().__init__(*args, **kwargs)
self.fields = OrderedDict(
(key, self.fields[key])
for key in [
@@ -163,7 +163,7 @@ class AccountFormGeneral(forms.ModelForm):
def clean(self):
"""Check master user mode."""
super(AccountFormGeneral, self).clean()
super().clean()
if self.errors:
return self.cleaned_data
condition = (
@@ -200,7 +200,7 @@ class AccountFormGeneral(forms.ModelForm):
return self.cleaned_data
def save(self, commit=True):
account = super(AccountFormGeneral, self).save(commit=False)
account = super().save(commit=False)
if self.user == account and not self.cleaned_data["is_active"]:
raise lib_exceptions.PermDeniedException(
_("You can't disable your own account")
@@ -292,12 +292,12 @@ class AccountFormMail(forms.Form, DynamicForm):
self.fields["email"].required = True
qset = self.mb.aliasrecipient_set.filter(alias__internal=False)
for cpt, ralias in enumerate(qset):
name = "aliases_{}".format(cpt + 1)
name = f"aliases_{cpt + 1}"
self._create_field(
lib_fields.UTF8AndEmptyUserEmailField, name, ralias.alias.address
)
for cpt, saddress in enumerate(self.mb.senderaddress_set.all()):
name = "senderaddress_{}".format(cpt + 1)
name = f"senderaddress_{cpt + 1}"
self._create_field(
lib_fields.UTF8AndEmptyUserEmailField, name, saddress.address
)
@@ -336,14 +336,14 @@ class AccountFormMail(forms.Form, DynamicForm):
try:
self.domain = models.Domain.objects.get(name=domname)
except models.Domain.DoesNotExist:
raise forms.ValidationError(_("Domain does not exist"))
raise forms.ValidationError(_("Domain does not exist")) from None
if not self.mb:
try:
core_signals.can_create_object.send(
sender=self.__class__, context=self.domain, object_type="mailboxes"
)
except lib_exceptions.ModoboaException as inst:
raise forms.ValidationError(inst)
raise forms.ValidationError(inst) from None
return email
def clean_quota(self):
@@ -355,7 +355,7 @@ class AccountFormMail(forms.Form, DynamicForm):
Check if quota is >= 0 only when the domain value is not used.
"""
cleaned_data = super(AccountFormMail, self).clean()
cleaned_data = super().clean()
use_default_domain_quota = cleaned_data["quota_act"]
condition = (
not use_default_domain_quota
@@ -510,13 +510,13 @@ class AccountPermissionsForm(forms.Form, DynamicForm):
self.account = kwargs["instance"]
del kwargs["instance"]
super(AccountPermissionsForm, self).__init__(*args, **kwargs)
super().__init__(*args, **kwargs)
if not hasattr(self, "account") or self.account is None:
return
qset = models.Domain.objects.get_for_admin(self.account)
for pos, dom in enumerate(qset):
name = "domains_%d" % (pos + 1)
name = f"domains_{pos + 1}"
self._create_field(lib_fields.DomainNameField, name, dom.name)
if len(args) and isinstance(args[0], QueryDict):
self._load_from_qdict(args[0], "domains", lib_fields.DomainNameField)
@@ -580,7 +580,7 @@ class AccountForm(TabForms):
cbargs["account"] = kwargs["instances"]["general"]
results = signals.extra_account_forms.send(sender=self.__class__, **cbargs)
self.forms += reduce(lambda a, b: a + b, [result[1] for result in results])
super(AccountForm, self).__init__(request, *args, **kwargs)
super().__init__(request, *args, **kwargs)
def extra_context(self, context):
account = self.instances["general"]
@@ -604,8 +604,8 @@ class AccountForm(TabForms):
if form["id"] == "general":
return True
if hasattr(self, "check_%s" % form["id"]):
if not getattr(self, "check_%s" % form["id"])(self.account):
if hasattr(self, "check_{}".format(form["id"])):
if not getattr(self, "check_{}".format(form["id"]))(self.account):
return False
return True
@@ -620,9 +620,9 @@ class AccountForm(TabForms):
def is_valid(self):
"""Two steps validation."""
self.instances["general"].oldgroup = self.instances["general"].role
if super(AccountForm, self).is_valid(mandatory_only=True):
if super().is_valid(mandatory_only=True):
self.account = self.forms[0]["instance"].save()
return super(AccountForm, self).is_valid(optional_only=True)
return super().is_valid(optional_only=True)
return False
def save(self):
@@ -646,7 +646,7 @@ class AccountWizard(WizardForm):
"""Account creation wizard."""
def __init__(self, request):
super(AccountWizard, self).__init__(request)
super().__init__(request)
self.add_step(
WizardStep(
"general", AccountFormGeneral, _("General"), new_args=[request.user]
+7 -9
View File
@@ -49,7 +49,7 @@ class AliasForm(forms.ModelForm, DynamicForm):
def __init__(self, user, *args, **kwargs):
self.user = user
super(AliasForm, self).__init__(*args, **kwargs)
super().__init__(*args, **kwargs)
self.fields = OrderedDict(
(key, self.fields[key])
for key in [
@@ -80,7 +80,7 @@ class AliasForm(forms.ModelForm, DynamicForm):
self.fields["enabled"].widget.attrs["disabled"] = "disabled"
cpt = 1
for rcpt in alias.aliasrecipient_set.filter(alias__internal=False):
name = "recipients_%d" % cpt
name = f"recipients_{cpt}"
self._create_field(forms.EmailField, name, rcpt.address, 2)
cpt += 1
@@ -92,7 +92,7 @@ class AliasForm(forms.ModelForm, DynamicForm):
try:
domain = Domain.objects.get(name=domname)
except Domain.DoesNotExist:
raise forms.ValidationError(_("Domain does not exist"))
raise forms.ValidationError(_("Domain does not exist")) from None
if not self.user.can_access(domain):
raise forms.ValidationError(_("You don't have access to this domain"))
if not self.instance.pk:
@@ -101,12 +101,12 @@ class AliasForm(forms.ModelForm, DynamicForm):
sender=self.__class__, context=domain, object_type="mailbox_aliases"
)
except lib_exceptions.ModoboaException as inst:
raise forms.ValidationError(inst)
raise forms.ValidationError(inst) from None
return self.cleaned_data["address"].lower()
def clean(self):
"""Check it there is at least one recipient."""
super(AliasForm, self).clean()
super().clean()
random_address = self.cleaned_data.get("random_address")
if not random_address:
if not self.cleaned_data.get("address"):
@@ -121,11 +121,9 @@ class AliasForm(forms.ModelForm, DynamicForm):
def save(self, commit=True):
"""Custom save method."""
alias = super(AliasForm, self).save(commit=False)
alias = super().save(commit=False)
if self.cleaned_data.get("random_address"):
alias.address = "{}@{}".format(
Alias.generate_random_address(), alias.domain
)
alias.address = f"{Alias.generate_random_address()}@{alias.domain}"
else:
local_part, domname = split_mailbox(self.cleaned_data["address"])
alias.domain = Domain.objects.get(name=domname)
+11 -11
View File
@@ -80,7 +80,7 @@ class DomainFormGeneral(forms.ModelForm, DynamicForm):
if "instance" in kwargs:
self.old_dkim_key_length = kwargs["instance"].dkim_key_length
self.oldname = kwargs["instance"].name
super(DomainFormGeneral, self).__init__(*args, **kwargs)
super().__init__(*args, **kwargs)
params = dict(param_tools.get_global_parameters("admin"))
self.fields["quota"].initial = params["default_domain_quota"]
if params["default_domain_message_limit"] is not None:
@@ -97,7 +97,7 @@ class DomainFormGeneral(forms.ModelForm, DynamicForm):
elif "instance" in kwargs:
d = kwargs["instance"]
for pos, dalias in enumerate(d.domainalias_set.all()):
name = "aliases_%d" % (pos + 1)
name = f"aliases_{pos + 1}"
self._create_field(forms.CharField, name, dalias.name, 3)
def clean_name(self):
@@ -148,7 +148,7 @@ class DomainFormGeneral(forms.ModelForm, DynamicForm):
The validation way is not very smart...
"""
cleaned_data = super(DomainFormGeneral, self).clean()
cleaned_data = super().clean()
if self._errors:
return cleaned_data
condition = (
@@ -203,7 +203,7 @@ class DomainFormGeneral(forms.ModelForm, DynamicForm):
quotas update.
"""
d = super(DomainFormGeneral, self).save(commit=False)
d = super().save(commit=False)
core_signals.can_create_object.send(
sender=self.__class__, context=user, klass=Domain, instance=d
)
@@ -280,7 +280,7 @@ class DomainFormOptions(forms.Form):
)
def __init__(self, user, *args, **kwargs):
super(DomainFormOptions, self).__init__(*args, **kwargs)
super().__init__(*args, **kwargs)
results = core_signals.user_can_set_role.send(
sender=self.__class__, user=user, role="DomainAdmins"
)
@@ -296,7 +296,7 @@ class DomainFormOptions(forms.Form):
def clean(self):
"""Check required values."""
cleaned_data = super(DomainFormOptions, self).clean()
cleaned_data = super().clean()
if cleaned_data.get("create_dom_admin"):
if not cleaned_data.get("dom_admin_username"):
self.add_error("dom_admin_username", _("This field is required."))
@@ -311,7 +311,7 @@ class DomainFormOptions(forms.Form):
return
user = kwargs.pop("user")
domain = kwargs.pop("domain")
username = "%s@%s" % (self.cleaned_data["dom_admin_username"], domain.name)
username = f"{self.cleaned_data['dom_admin_username']}@{domain.name}"
try:
da = User.objects.get(username=username)
except User.DoesNotExist:
@@ -349,7 +349,7 @@ class DomainFormOptions(forms.Form):
core_signals.can_create_object.send(
self.__class__, context=user, klass=Alias
)
address = "postmaster@{}".format(domain.name)
address = f"postmaster@{domain.name}"
alias = Alias.objects.create(
address=address, domain=domain, enabled=True
)
@@ -386,7 +386,7 @@ class DomainForm(TabForms):
self.forms += reduce(lambda a, b: a + b, [result[1] for result in results])
if not self.forms:
self.active_id = "admins"
super(DomainForm, self).__init__(request, *args, **kwargs)
super().__init__(request, *args, **kwargs)
def extra_context(self, context):
"""Add information to template context."""
@@ -409,7 +409,7 @@ class DomainForm(TabForms):
if "general" in self.instances:
instance = self.instances["general"]
instance.oldname = instance.name
return super(DomainForm, self).is_valid()
return super().is_valid()
def save(self):
"""Custom save method.
@@ -437,7 +437,7 @@ class DomainWizard(WizardForm):
"""Domain creation wizard."""
def __init__(self, request):
super(DomainWizard, self).__init__(request)
super().__init__(request)
self.add_step(
WizardStep(
"general",
+1 -1
View File
@@ -13,7 +13,7 @@ class ExportDataForm(forms.Form):
)
def __init__(self, *args, **kwargs):
super(ExportDataForm, self).__init__(*args, **kwargs)
super().__init__(*args, **kwargs)
self.fields["sepchar"].widget.attrs = {"class": "col-md-1 form-control"}
def clean_sepchar(self):
+1 -1
View File
@@ -21,7 +21,7 @@ class ImportDataForm(forms.Form):
)
def __init__(self, *args, **kwargs):
super(ImportDataForm, self).__init__(*args, **kwargs)
super().__init__(*args, **kwargs)
self.fields["sepchar"].widget.attrs = {"class": "col-md-1 form-control"}
def clean_sepchar(self):
+6 -10
View File
@@ -27,13 +27,9 @@ def update_domain_mxs_and_mailboxes(sender, instance, **kwargs):
)
if instance.old_mail_homes is None:
return
qset = models.Quota.objects.filter(
username__contains="@{}".format(instance.oldname)
)
qset = models.Quota.objects.filter(username__contains=f"@{instance.oldname}")
for q in qset:
username = q.username.replace(
"@{}".format(instance.oldname), "@{}".format(instance.name)
)
username = q.username.replace(f"@{instance.oldname}", f"@{instance.name}")
models.Quota.objects.create(
username=username, bytes=q.bytes, messages=q.messages
)
@@ -62,17 +58,17 @@ def create_alias_for_domainalias(sender, instance, **kwargs):
if not kwargs.get("created"):
return
alias = models.Alias.objects.create(
address="@{}".format(instance.name), enabled=True, internal=True
address=f"@{instance.name}", enabled=True, internal=True
)
models.AliasRecipient.objects.create(
address="@{}".format(instance.target.name), alias=alias
address=f"@{instance.target.name}", alias=alias
)
@receiver(signals.post_delete, sender=models.DomainAlias)
def remove_alias_for_domainalias(sender, instance, **kwargs):
"""Remove the alias associated to domain alias."""
models.Alias.objects.filter(address="@{}".format(instance.name)).delete()
models.Alias.objects.filter(address=f"@{instance.name}").delete()
@receiver(signals.post_save, sender=models.Mailbox)
@@ -256,7 +252,7 @@ def import_account_mailbox(sender, user, account, row, **kwargs):
raise exceptions.BadRequest(
_("Account import failed (%s): wrong quota value")
% account.username
)
) from None
use_domain_quota = True if not quota else False
mb = models.Mailbox(
address=mailbox,
+3 -3
View File
@@ -205,7 +205,7 @@ def get_authoritative_server(domain, resolver=None):
return str(first_answer.target)
if hasattr(first_answer, "address"):
return str(first_answer.address)
except dns.resolver.NoAnswer as e:
except dns.resolver.NoAnswer:
dnsname = dnsname.parent()
else:
dnsname = dnsname.parent()
@@ -253,7 +253,7 @@ def get_dns_records(name, typ, resolver=None):
exc_info=e,
)
except dns.name.NameTooLong as e:
logger.error(_("DNS name is too long: %s" % name), exc_info=e)
logger.error(_("DNS name is too long: {}".format(name)), exc_info=e)
else:
return dns_answers
return None
@@ -359,7 +359,7 @@ def import_data(user, file_object, options: dict):
raise Conflict(
_("Object already exists: %s")
% options["sepchar"].join(row[:2])
)
) from None
cpt += 1
msg = _("%d objects imported successfully") % cpt
return True, msg
@@ -25,7 +25,7 @@ class Command(BaseCommand):
help = "Handles rename and delete operations on mailboxes" # NOQA:A003
def __init__(self, *args, **kwargs):
super(Command, self).__init__(*args, **kwargs)
super().__init__(*args, **kwargs)
self.logger = logging.getLogger("modoboa.admin")
def add_arguments(self, parser):
@@ -44,9 +44,9 @@ class Command(BaseCommand):
if not os.path.exists(dirname):
try:
os.makedirs(dirname)
except os.error as e:
raise OperationError(str(e))
code, output = exec_cmd("mv %s %s" % (operation.argument, new_mail_home))
except OSError as e:
raise OperationError(str(e)) from None
code, output = exec_cmd(f"mv {operation.argument} {new_mail_home}")
if code:
raise OperationError(output)
@@ -57,7 +57,7 @@ class Command(BaseCommand):
def onerror(function, path, excinfo):
"""Handle errors."""
self.logger.critical("delete failed (reason: {})".format(excinfo))
self.logger.critical(f"delete failed (reason: {excinfo})")
shutil.rmtree(operation.argument, False, onerror)
@@ -71,7 +71,7 @@ class Command(BaseCommand):
with open(path) as fp:
pid = fp.read().strip()
code, output = exec_cmd(
"grep handle_mailbox_operations /proc/%s/cmdline" % pid
f"grep handle_mailbox_operations /proc/{pid}/cmdline"
)
if not code:
return False
@@ -88,7 +88,7 @@ class Command(BaseCommand):
return
for ope in MailboxOperation.objects.all():
try:
f = getattr(self, "%s_mailbox" % ope.type)
f = getattr(self, f"{ope.type}_mailbox")
except AttributeError:
continue
try:
@@ -1,13 +1,11 @@
"""Django management command to import admin objects."""
import csv
import io
import os
import progressbar
from chardet.universaldetector import UniversalDetector
from django.core.exceptions import ValidationError
from django.core.management.base import BaseCommand, CommandError
from django.db import transaction
from django.utils.translation import gettext as _
@@ -50,12 +48,12 @@ class ImportCommand(BaseCommand):
if not os.path.isfile(filename):
raise CommandError("File not found")
num_lines = sum(1 for line in io.open(filename, encoding=encoding) if line)
num_lines = sum(1 for line in open(filename, encoding=encoding) if line)
pbar = progressbar.ProgressBar(
widgets=[progressbar.Percentage(), progressbar.Bar(), progressbar.ETA()],
maxval=num_lines,
).start()
with io.open(filename, encoding=encoding, newline="") as f:
with open(filename, encoding=encoding, newline="") as f:
reader = csv.reader(f, delimiter=options["sepchar"])
i = 0
for row in reader:
@@ -77,7 +75,7 @@ class ImportCommand(BaseCommand):
"Object already exists at line {}: {}".format(
i + 1, options["sepchar"].join(row[:2])
)
)
) from None
i += 1
pbar.update(i)
@@ -102,7 +100,7 @@ class ImportCommand(BaseCommand):
)
)
detector = UniversalDetector()
with io.open(filename, "rb") as fp:
with open(filename, "rb") as fp:
for line in fp:
detector.feed(line)
if detector.done:
@@ -19,7 +19,7 @@ class ManageDKIMKeys(BaseCommand):
def create_new_dkim_key(self, domain):
"""Create a new DKIM key."""
storage_dir = param_tools.get_global_parameter("dkim_keys_storage_dir")
pkey_path = os.path.join(storage_dir, "{}.pem".format(domain.name))
pkey_path = os.path.join(storage_dir, f"{domain.name}.pem")
alarm_qset = domain.alarms.filter(internal_name=DKIM_WRITE_ERROR)
if not os.access(storage_dir, os.W_OK):
@@ -43,26 +43,20 @@ class ManageDKIMKeys(BaseCommand):
if domain.dkim_key_length
else self.default_key_length
)
code, output = sysutils.exec_cmd(
"openssl genrsa -out {} {}".format(pkey_path, key_size)
)
code, output = sysutils.exec_cmd(f"openssl genrsa -out {pkey_path} {key_size}")
if code:
print(
"Failed to generate DKIM private key for domain {}: {}".format(
domain.name, smart_str(output)
)
f"Failed to generate DKIM private key for domain {domain.name}: {smart_str(output)}"
)
domain.alarms.create(
title=_("Failed to generate DKIM private key"), internal_name=DKIM_ERROR
)
return
domain.dkim_private_key_path = pkey_path
code, output = sysutils.exec_cmd("openssl rsa -in {} -pubout".format(pkey_path))
code, output = sysutils.exec_cmd(f"openssl rsa -in {pkey_path} -pubout")
if code:
print(
"Failed to generate DKIM public key for domain {}: {}".format(
domain.name, smart_str(output)
)
f"Failed to generate DKIM public key for domain {domain.name}: {smart_str(output)}"
)
domain.alarms.create(
title=_("Failed to generate DKIM public key"), internal_name=DKIM_ERROR
@@ -87,7 +87,7 @@ class CheckMXRecords(BaseCommand):
else:
delim = "." if ip.version == 4 else ":"
reverse = delim.join(ip.exploded.split(delim)[::-1])
pattern = "{}.{}.".format(reverse, provider)
pattern = f"{reverse}.{provider}."
try:
result = socket.gethostbyname(pattern)
# result from dnsbl is in ipv4 format
@@ -128,7 +128,7 @@ class CheckMXRecords(BaseCommand):
dnsbl_result.save()
if not dnsbl_result.status and result:
trigger = True
alarm_name = "domain_mx_in_dnsbl_{}".format(provider)
alarm_name = f"domain_mx_in_dnsbl_{provider}"
if trigger:
title = _("MX {} listed by DNSBL provider {}").format(mx.name, provider)
domain.alarms.create(
@@ -165,7 +165,7 @@ class CheckMXRecords(BaseCommand):
"""
alerts = []
check = False
mxs = [(mx, ipaddress.ip_address("%s" % mx.address)) for mx in mx_list]
mxs = [(mx, ipaddress.ip_address(str(mx.address))) for mx in mx_list]
valid_mxs = self.valid_mxs
if not mxs:
alarm, created = domain.alarms.get_or_create(
@@ -187,7 +187,7 @@ class CheckMXRecords(BaseCommand):
mx.managed = check = True
mx.save()
if check is False:
mx_names = ["{0.name} ({0.address})".format(mx) for mx in mx_list]
mx_names = [f"{mx.name} ({mx.address})" for mx in mx_list]
alarm, created = domain.alarms.get_or_create(
internal_name="domain_invalid_mx",
status=constants.ALARM_OPENED,
@@ -76,18 +76,16 @@ def sometimes_mailbox_have_no_alias(**options):
)
if created:
alias_created += 1
log("Alias {0} created".format(alias), **options)
log(f"Alias {alias} created", **options)
recipient, created = models.AliasRecipient.objects.get_or_create(
alias=alias, address=instance.full_address, r_mailbox=instance
)
if created:
recipient_created += 1
log("AliasRecipient {0} created".format(recipient), **options)
log(f"AliasRecipient {recipient} created", **options)
if alias_created or recipient_created:
log(
"{0} alias created. {1} alias recipient created".format(
alias_created, recipient_created
),
f"{alias_created} alias created. {recipient_created} alias recipient created",
**options,
)
@@ -129,5 +127,5 @@ class Repair(BaseCommand):
for func in known_problems:
title = func.__doc__.strip()
log("", **options)
log("Checking for... {}...".format(title), **options)
log(f"Checking for... {title}...", **options)
func(**options)
@@ -1,4 +1,4 @@
from django.db import models, migrations
from django.db import migrations
def rename_and_clean(apps, schema_editor):
+1 -3
View File
@@ -48,9 +48,7 @@ class Alarm(models.Model):
ordering = ["created"]
def __str__(self):
return "[{}] {} - {}".format(
self.created, self.domain, self.get_status_display()
)
return f"[{self.created}] {self.domain} - {self.get_status_display()}"
def close(self):
"""Close this alarm."""
+6 -6
View File
@@ -54,7 +54,7 @@ def validate_alias_address(
object_type="mailbox_aliases",
)
except ModoboaException as inst:
raise ValidationError(str(inst))
raise ValidationError(str(inst)) from None
return local_part, domain
@@ -132,7 +132,7 @@ class Alias(AdminObject):
return "---"
rcpts = self.recipients
if rcpts_count > 1:
return "%s, ..." % rcpts[0]
return f"{rcpts[0]}, ..."
return rcpts[0]
@property
@@ -150,7 +150,7 @@ class Alias(AdminObject):
def post_create(self, creator):
from modoboa.lib.permissions import grant_access_to_object
super(Alias, self).post_create(creator)
super().post_create(creator)
if creator.is_superuser:
for admin in self.domain.admins:
grant_access_to_object(admin, self)
@@ -174,7 +174,7 @@ class Alias(AdminObject):
address, return_extension=True
)
if domname is None:
raise BadRequest("%s %s" % (_("Invalid address"), address))
raise BadRequest("{} {}".format(_("Invalid address"), address))
domain = Domain.objects.filter(name=domname).first()
kwargs = {"address": address, "alias": self}
if (domain is not None) and (
@@ -189,7 +189,7 @@ class Alias(AdminObject):
rcpt = Mailbox.objects.filter(domain=domain, address=local_part).first()
if rcpt is None:
rcpt = Alias.objects.filter(
address="%s@%s" % (local_part, domname)
address=f"{local_part}@{domname}"
).first()
if rcpt is None:
raise NotFound(
@@ -306,4 +306,4 @@ class AliasRecipient(models.Model):
def __str__(self):
"""Return alias and recipient."""
return smart_str("{} -> {}".format(self.alias.address, self.address))
return smart_str(f"{self.alias.address} -> {self.address}")
+2 -2
View File
@@ -62,10 +62,10 @@ class AdminObject(models.Model):
def save(self, *args, **kwargs):
creator = kwargs.pop("creator", None)
super(AdminObject, self).save(*args, **kwargs)
super().save(*args, **kwargs)
if creator is not None:
self.post_create(creator)
def delete(self):
ungrant_access_to_object(self)
super(AdminObject, self).delete()
super().delete()
+9 -9
View File
@@ -271,22 +271,20 @@ class Domain(mixins.MessageLimitMixin, AdminObject):
@property
def bind_format_dkim_public_key(self):
"""Return DKIM public key using Bind format."""
record = "v=DKIM1;k=rsa;p=%s" % self.dkim_public_key
record = f"v=DKIM1;k=rsa;p={self.dkim_public_key}"
# TXT records longer than 255 characters need split into chunks
# split record into 74 character chunks (+2 for indent, +2 for quotes(")
# == 78 characters) to make more readable in text editors.
split_record = []
while record:
if len(record) > 74:
split_record.append(' "%s"' % record[:74])
split_record.append(f' "{record[:74]}"')
record = record[74:]
else:
split_record.append(' "%s"' % record)
split_record.append(f' "{record}"')
break
record = "\n".join(split_record)
return "{}._domainkey.{}. IN TXT (\n{})".format(
self.dkim_key_selector, self.name, record
)
return f"{self.dkim_key_selector}._domainkey.{self.name}. IN TXT (\n{record})"
def add_admin(self, account):
"""Add a new administrator to this domain.
@@ -345,7 +343,7 @@ class Domain(mixins.MessageLimitMixin, AdminObject):
if param_tools.get_global_parameter("auto_account_removal"):
User.objects.filter(mailbox__domain=self).delete()
elif self.mailbox_set.count():
Quota.objects.filter(username__contains="@%s" % self.name).delete()
Quota.objects.filter(username__contains=f"@{self.name}").delete()
ungrant_access_to_objects(self.mailbox_set.all())
super().delete()
@@ -380,7 +378,9 @@ class Domain(mixins.MessageLimitMixin, AdminObject):
try:
self.quota = int(row[2].strip())
except ValueError:
raise BadRequest(_("{}: invalid quota value for domain").format(self.name))
raise BadRequest(
_("{}: invalid quota value for domain").format(self.name)
) from None
try:
self.default_mailbox_quota = int(row[3].strip())
except ValueError:
@@ -388,7 +388,7 @@ class Domain(mixins.MessageLimitMixin, AdminObject):
_("{}: invalid default mailbox quota value for domain").format(
self.name
)
)
) from None
if self.quota != 0 and self.default_mailbox_quota > self.quota:
raise BadRequest(
_(
+1 -1
View File
@@ -76,7 +76,7 @@ class DomainAlias(AdminObject):
try:
self.target = Domain.objects.get(name=domname)
except Domain.DoesNotExist:
raise BadRequest(_("Unknown domain %s") % domname)
raise BadRequest(_("Unknown domain %s") % domname) from None
core_signals.can_create_object.send(
sender="import", context=self.target, object_type="domain_aliases"
)
+5 -5
View File
@@ -25,7 +25,7 @@ class QuotaManager(models.Manager):
def get_domain_usage(self, domain):
"""Return current usage for domain."""
qset = self.get_queryset().filter(username__endswith="@{}".format(domain.name))
qset = self.get_queryset().filter(username__endswith=f"@{domain.name}")
result = qset.aggregate(usage=models.Sum("bytes")).get("usage", 0)
if result is None:
result = 0
@@ -114,7 +114,7 @@ class Mailbox(mixins.MessageLimitMixin, AdminObject):
return smart_str(self.full_address)
def __full_address(self, localpart):
return "%s@%s" % (localpart, self.domain.name)
return f"{localpart}@{self.domain.name}"
@property
def full_address(self):
@@ -147,7 +147,7 @@ class Mailbox(mixins.MessageLimitMixin, AdminObject):
if not admin_params.get("handle_mailboxes"):
return None
if self.__mail_home is None:
code, output = doveadm_cmd("user -f home %s" % self.full_address)
code, output = doveadm_cmd(f"user -f home {self.full_address}")
if code:
raise lib_exceptions.InternalError(
_("Failed to retrieve mailbox location (%s)") % output
@@ -384,5 +384,5 @@ class MailboxOperation(models.Model):
def __str__(self):
if self.type == "rename":
return "Rename %s -> %s" % (self.argument, self.mailbox.mail_home)
return "Delete %s" % self.argument
return f"Rename {self.argument} -> {self.mailbox.mail_home}"
return f"Delete {self.argument}"
+2 -2
View File
@@ -45,7 +45,7 @@ class MXRecordManager(models.Manager):
record = self.get_queryset().create(
domain=domain,
name="{}".format(mx_addr.strip(".")),
address="{}".format(mx_ip_addr),
address=f"{mx_ip_addr}",
updated=now + delta,
)
yield record
@@ -68,7 +68,7 @@ class MXRecord(models.Model):
return bool(param_tools.get_global_parameter("valid_mxs").strip())
def __str__(self):
return "{0.name} ({0.address}) for {0.domain} ".format(self)
return f"{self.name} ({self.address}) for {self.domain} "
class DNSBLQuerySet(models.QuerySet):
+5 -5
View File
@@ -1,7 +1,7 @@
"""Map file definitions for postfix."""
class DomainsMap(object):
class DomainsMap:
"""Map to list all domains."""
filename = "sql-domains.cf"
@@ -18,7 +18,7 @@ class DomainsMap(object):
)
class DomainsAliasesMap(object):
class DomainsAliasesMap:
"""Map to list all domain aliases."""
filename = "sql-domain-aliases.cf"
@@ -39,7 +39,7 @@ class DomainsAliasesMap(object):
)
class AliasesMap(object):
class AliasesMap:
"""A map to list all mailbox aliases."""
filename = "sql-aliases.cf"
@@ -63,7 +63,7 @@ class AliasesMap(object):
)
class MaintainMap(object):
class MaintainMap:
"""Map files to list non available mailboxes."""
filename = "sql-maintain.cf"
@@ -90,7 +90,7 @@ class MaintainMap(object):
)
class SenderLoginMap(object):
class SenderLoginMap:
"""Map file to list authorized sender addresses for a given account:
* Its mailbox
* Its aliases
+4 -4
View File
@@ -151,7 +151,7 @@ def domain_actions(user, domain):
actions = [
{
"name": "listidentities",
"url": "{0}#list/?searchquery=@{1}".format(
"url": "{}#list/?searchquery=@{}".format(
reverse("admin:identity_list"), domain.name
),
"title": _("View the domain's identities"),
@@ -268,7 +268,7 @@ def domain_aliases(domain):
return "---"
res = ""
for alias in domain.aliases.all():
res += "%s<br/>" % alias.name
res += f"{alias.name}<br/>"
return mark_safe(res)
@@ -286,7 +286,7 @@ def identity_modify_link(identity, active_tab="default"):
linkdef = {"label": identity.identity, "modal": True}
if identity.__class__.__name__ == "User":
linkdef["url"] = reverse("admin:account_change", args=[identity.id])
linkdef["url"] += "?active_tab=%s" % active_tab
linkdef["url"] += f"?active_tab={active_tab}"
linkdef["modalcb"] = "admin.editaccount_cb"
else:
linkdef["url"] = reverse("admin:alias_change", args=[identity.id])
@@ -299,7 +299,7 @@ def domadmin_actions(daid, domid):
actions = [
{
"name": "removeperm",
"url": "{0}?domid={1}&daid={2}".format(
"url": "{}?domid={}&daid={}".format(
reverse("admin:permission_remove"), domid, daid
),
"img": "fa fa-trash",
+8 -8
View File
@@ -17,7 +17,7 @@ class AuthenticationTestCase(ModoTestCase):
@classmethod
def setUpTestData(cls): # NOQA:N802
"""Create test data."""
super(AuthenticationTestCase, cls).setUpTestData()
super().setUpTestData()
cls.mb = factories.MailboxFactory(
domain__name="test.com",
address="user",
@@ -42,7 +42,7 @@ class AccountTestCase(ModoTestCase):
@classmethod
def setUpTestData(cls): # NOQA:N802
"""Create test data."""
super(AccountTestCase, cls).setUpTestData()
super().setUpTestData()
factories.populate_database()
def test_crud(self):
@@ -576,14 +576,14 @@ class AccountTestCase(ModoTestCase):
self.assertIn("user@test.com", response["rows"])
old_rows = response["rows"]
response = self.ajax_get("{}?sort_order=-quota_value__bytes".format(url))
response = self.ajax_get(f"{url}?sort_order=-quota_value__bytes")
self.assertNotEqual(old_rows, response["rows"])
old_rows = response["rows"]
response = self.ajax_get("{}?sort_order=-quota_usage".format(url))
response = self.ajax_get(f"{url}?sort_order=-quota_usage")
self.assertEqual(old_rows, response["rows"])
response = self.ajax_get("{}?sort_order=-unknown".format(url), status=400)
response = self.ajax_get(f"{url}?sort_order=-unknown", status=400)
@skipIf(NO_LDAP, "No ldap module installed")
@@ -612,10 +612,10 @@ class PermissionsTestCase(ModoTestCase):
@classmethod
def setUpTestData(cls): # NOQA:N802
"""Create test data."""
super(PermissionsTestCase, cls).setUpTestData()
super().setUpTestData()
parameters = {}
for name, _definition in limits_utils.get_user_limit_templates():
parameters["deflt_user_{0}_limit".format(name)] = 2
parameters[f"deflt_user_{name}_limit"] = 2
cls.localconfig.parameters.set_values(parameters, app="limits")
cls.localconfig.save()
factories.populate_database()
@@ -625,7 +625,7 @@ class PermissionsTestCase(ModoTestCase):
def setUp(self):
"""Initiate test context."""
super(PermissionsTestCase, self).setUp()
super().setUp()
self.values = {
"username": self.user.username,
"role": "DomainAdmins",
-2
View File
@@ -1,7 +1,5 @@
"""Alarm related tests."""
from django.urls import reverse
from modoboa.lib.tests import ModoTestCase
from ..factories import populate_database
from ..models import Alarm, Domain
+2 -2
View File
@@ -13,7 +13,7 @@ class AliasTestCase(ModoTestCase):
@classmethod
def setUpTestData(cls): # NOQA:N802
"""Create test data."""
super(AliasTestCase, cls).setUpTestData()
super().setUpTestData()
factories.populate_database()
def test_alias(self):
@@ -173,7 +173,7 @@ class AliasTestCase(ModoTestCase):
self.ajax_post(reverse("admin:alias_change", args=[fwd.id]), values)
self.assertEqual(fwd.recipients_count, 1)
self.ajax_delete(reverse("admin:alias_delete") + "?selection=%d" % fwd.id, {})
self.ajax_delete(reverse("admin:alias_delete") + f"?selection={fwd.id}", {})
self.assertRaises(
Alias.DoesNotExist, Alias.objects.get, address="forward2@test.com"
)
+7 -7
View File
@@ -32,7 +32,7 @@ class DomainTestCase(ModoTestCase):
@classmethod
def setUpTestData(cls): # NOQA:N802
"""Create test data."""
super(DomainTestCase, cls).setUpTestData()
super().setUpTestData()
factories.populate_database()
def test_create(self):
@@ -364,10 +364,10 @@ class DomainTestCase(ModoTestCase):
self.assertIn("test.com", response["rows"])
old_rows = response["rows"]
response = self.ajax_get("{}?sort_order=-name".format(url))
response = self.ajax_get(f"{url}?sort_order=-name")
self.assertNotEqual(old_rows, response["rows"])
response = self.ajax_get("{}?sort_order=allocated_quota".format(url))
response = self.ajax_get(f"{url}?sort_order=allocated_quota")
self.assertNotEqual(old_rows, response["rows"])
def test_domain_logs_list_view(self):
@@ -386,10 +386,10 @@ class DomainTestCase(ModoTestCase):
self.assertIn("ID1", response["rows"])
old_rows = response["rows"]
response = self.ajax_get("{}?sort_order=-sender".format(url))
response = self.ajax_get(f"{url}?sort_order=-sender")
self.assertNotEqual(old_rows, response["rows"])
response = self.ajax_get("{}?searchquery=ID1".format(url))
response = self.ajax_get(f"{url}?searchquery=ID1")
self.assertNotEqual(old_rows, response["rows"])
admin = User.objects.get(username="admin@test.com")
@@ -421,7 +421,7 @@ class DomainTestCase(ModoTestCase):
response = self.ajax_get(url)
self.assertIn("handle_mailboxes", response)
self.assertIn("listalarms", response["rows"])
response = self.ajax_get("{}?objtype=quota".format(url))
response = self.ajax_get(f"{url}?objtype=quota")
self.assertIn("progress-bar", response["rows"])
@@ -431,7 +431,7 @@ class DKIMTestCase(ModoTestCase):
@classmethod
def setUpTestData(cls): # NOQA:N802
"""Create test data."""
super(DKIMTestCase, cls).setUpTestData()
super().setUpTestData()
factories.populate_database()
def setUp(self):
+3 -5
View File
@@ -10,7 +10,7 @@ class DomainAliasTestCase(ModoTestCase):
@classmethod
def setUpTestData(cls): # NOQA:N802
"""Create test data."""
super(DomainAliasTestCase, cls).setUpTestData()
super().setUpTestData()
factories.populate_database()
cls.dom = Domain.objects.get(name="test.com")
@@ -21,15 +21,13 @@ class DomainAliasTestCase(ModoTestCase):
domal.target = dom
domal.save()
self.assertEqual(dom.domainalias_count, 1)
self.assertTrue(Alias.objects.filter(address="@{}".format(domal.name)).exists())
self.assertTrue(Alias.objects.filter(address=f"@{domal.name}").exists())
domal.name = "domalias.org"
domal.save()
domal.delete()
self.assertFalse(
Alias.objects.filter(address="@{}".format(domal.name)).exists()
)
self.assertFalse(Alias.objects.filter(address=f"@{domal.name}").exists())
def test_form(self):
dom = Domain.objects.get(name="test.com")
+2 -2
View File
@@ -17,7 +17,7 @@ class ExportTestCase(ModoTestCase):
@classmethod
def setUpTestData(cls): # NOQA:N802
"""Create test data."""
super(ExportTestCase, cls).setUpTestData()
super().setUpTestData()
factories.populate_database()
def __export_domains(self, domfilter=""):
@@ -31,7 +31,7 @@ class ExportTestCase(ModoTestCase):
def __export_identities(self, idtfilter="", grpfilter=""):
self.client.get(
reverse("admin:_identity_list")
+ "?grpfilter=%s&idtfilter=%s" % (grpfilter, idtfilter)
+ f"?grpfilter={grpfilter}&idtfilter={idtfilter}"
)
return self.client.post(
reverse("admin:identity_export"), {"filename": "test.csv"}
+2 -4
View File
@@ -21,7 +21,7 @@ class ImportTestCase(ModoTestCase):
@classmethod
def setUpTestData(cls): # NOQA:N802
"""Create test data."""
super(ImportTestCase, cls).setUpTestData()
super().setUpTestData()
cls.localconfig.parameters.set_value("enable_admin_limits", False, app="limits")
cls.localconfig.save()
factories.populate_database()
@@ -326,9 +326,7 @@ alias;user@test.com;True;admin@test.com
f = ContentFile(
"""domain; dómªin1.com; 1000; 100; True
dómain; dómªin2.com; 1000; 100; True
""".encode(
"utf8"
),
""".encode(),
name="dómains.csv",
)
self.client.post(reverse("admin:domain_import"), {"sourcefile": f})
@@ -13,21 +13,21 @@ from modoboa.lib.tests import ModoTestCase
from .. import factories, models
@override_settings(DOVECOT_LOOKUP_PATH=["{}/dovecot".format(os.path.dirname(__file__))])
@override_settings(DOVECOT_LOOKUP_PATH=[f"{os.path.dirname(__file__)}/dovecot"])
class MailboxOperationTestCase(ModoTestCase):
"""Test management command."""
@classmethod
def setUpTestData(cls): # NOQA:N802
"""Create test data."""
super(MailboxOperationTestCase, cls).setUpTestData()
super().setUpTestData()
factories.populate_database()
def setUp(self):
"""Initiate test env."""
super(MailboxOperationTestCase, self).setUp()
super().setUp()
self.workdir = tempfile.mkdtemp()
path = "{}/test.com/admin".format(self.workdir)
path = f"{self.workdir}/test.com/admin"
os.makedirs(path)
self.set_global_parameter("handle_mailboxes", True)
self.set_global_parameter("enable_admin_limits", False, app="limits")
@@ -39,7 +39,7 @@ class MailboxOperationTestCase(ModoTestCase):
@mock.patch("modoboa.admin.models.Mailbox.mail_home")
def test_delete_account(self, mail_home_mock):
"""Check delete operation."""
path = "{}/test.com/admin".format(self.workdir)
path = f"{self.workdir}/test.com/admin"
mail_home_mock.__get__ = mock.Mock(return_value=path)
mb = models.Mailbox.objects.select_related("user").get(
address="admin", domain__name="test.com"
@@ -52,7 +52,7 @@ class MailboxOperationTestCase(ModoTestCase):
@mock.patch("modoboa.admin.models.Mailbox.mail_home")
def test_rename_account(self, mail_home_mock):
"""Check rename operation."""
path = "{}/test.com/admin".format(self.workdir)
path = f"{self.workdir}/test.com/admin"
mail_home_mock.__get__ = mock.Mock(return_value=path)
mb = models.Mailbox.objects.select_related("user").get(
address="admin", domain__name="test.com"
@@ -67,7 +67,7 @@ class MailboxOperationTestCase(ModoTestCase):
"content": "content",
}
self.ajax_post(reverse("admin:account_change", args=[mb.user.pk]), values)
path = "{}/test.com/admin2".format(self.workdir)
path = f"{self.workdir}/test.com/admin2"
mail_home_mock.__get__ = mock.Mock(return_value=path)
call_command("handle_mailbox_operations")
self.assertFalse(models.MailboxOperation.objects.exists())
@@ -76,7 +76,7 @@ class MailboxOperationTestCase(ModoTestCase):
@mock.patch("modoboa.admin.models.Mailbox.mail_home")
def test_delete_domain(self, mail_home_mock):
"""Check delete operation."""
path = "{}/test.com/admin".format(self.workdir)
path = f"{self.workdir}/test.com/admin"
mail_home_mock.__get__ = mock.Mock(return_value=path)
domain = models.Domain.objects.get(name="test.com")
self.ajax_post(reverse("admin:domain_delete", args=[domain.pk]))
+3 -3
View File
@@ -23,7 +23,7 @@ class MXTestCase(ModoTestCase):
@classmethod
def setUpTestData(cls): # NOQA:N802
"""Create some data."""
super(MXTestCase, cls).setUpTestData()
super().setUpTestData()
cls.domain = factories.DomainFactory(name="modoboa.org")
# should not exist
cls.bad_domain = factories.DomainFactory(name="does-not-exist.example.com")
@@ -110,7 +110,7 @@ class MXTestCase(ModoTestCase):
mb = factories.MailboxFactory(
address="admin",
domain=domain,
user__username="admin@{}".format(domain.name),
user__username=f"admin@{domain.name}",
user__groups=("DomainAdmins",),
)
domain.add_admin(mb.user)
@@ -297,7 +297,7 @@ class DNSChecksTestCase(ModoTestCase):
@classmethod
def setUpTestData(cls): # NOQA:N802
"""Create some data."""
super(DNSChecksTestCase, cls).setUpTestData()
super().setUpTestData()
cls.domain = factories.DomainFactory(name="dns-checks.com")
@mock.patch("gevent.socket.gethostbyname")
+1 -1
View File
@@ -10,7 +10,7 @@ class PasswordSchemesTestCase(ModoTestCase):
@classmethod
def setUpTestData(cls): # NOQA:N802
"""Create test data."""
super(PasswordSchemesTestCase, cls).setUpTestData()
super().setUpTestData()
factories.populate_database()
def _create_account(self):
+1 -1
View File
@@ -13,7 +13,7 @@ class RepairTestCase(ModoTestCase):
@classmethod
def setUpTestData(cls): # NOQA:N802
"""Create some data."""
super(RepairTestCase, cls).setUpTestData()
super().setUpTestData()
factories.populate_database()
def test_management_command(self):
+1 -1
View File
@@ -12,7 +12,7 @@ class ForwardTestCase(ModoTestCase):
"""User forward test cases."""
def setUp(self):
super(ForwardTestCase, self).setUp()
super().setUp()
populate_database()
def test_set_forward(self):
+2 -2
View File
@@ -11,13 +11,13 @@ from dns.rdtypes.ANY.TXT import TXT
from dns.resolver import NXDOMAIN, NoAnswer, NoNameservers, Timeout
class RRset(object):
class RRset:
def __init__(self):
self.address = "192.0.2.1"
class RRsetInvalid(object):
class RRsetInvalid:
def __init__(self):
self.address = "000.0.0.0"
+3 -3
View File
@@ -27,7 +27,7 @@ def _validate_alias(request, form, successmsg, callback=None):
try:
alias = form.save()
except IntegrityError:
raise Conflict(_("Alias with this name already exists"))
raise Conflict(_("Alias with this name already exists")) from None
if callback:
callback(request.user, alias)
return render_to_json_response(successmsg)
@@ -110,13 +110,13 @@ class AliasDetailView(auth_mixins.PermissionRequiredMixin, generic.DetailView):
def has_permission(self):
"""Check object-level access."""
result = super(AliasDetailView, self).has_permission()
result = super().has_permission()
if not result:
return result
return self.request.user.can_access(self.get_object())
def get_context_data(self, **kwargs):
"""Add information to context."""
context = super(AliasDetailView, self).get_context_data(**kwargs)
context = super().get_context_data(**kwargs)
context["selection"] = "identities"
return context
+4 -12
View File
@@ -16,15 +16,11 @@ class MXDomainDetailView(auth_mixins.PermissionRequiredMixin, generic.DetailView
def get_queryset(self):
"""Add some prefetching."""
return (
super(MXDomainDetailView, self)
.get_queryset()
.prefetch_related("mxrecord_set")
)
return super().get_queryset().prefetch_related("mxrecord_set")
def get_context_data(self, **kwargs):
"""Add extra variables."""
context = super(MXDomainDetailView, self).get_context_data(**kwargs)
context = super().get_context_data(**kwargs)
context.update({"title": _("MX records of {}").format(self.object.name)})
return context
@@ -38,14 +34,10 @@ class DNSBLDomainDetailView(auth_mixins.PermissionRequiredMixin, generic.DetailV
def get_queryset(self):
"""Add some prefetching."""
return (
super(DNSBLDomainDetailView, self)
.get_queryset()
.prefetch_related("dnsblresult_set")
)
return super().get_queryset().prefetch_related("dnsblresult_set")
def get_context_data(self, **kwargs):
"""Add extra variables."""
context = super(DNSBLDomainDetailView, self).get_context_data(**kwargs)
context = super().get_context_data(**kwargs)
context.update({"title": _("DNSBL summary for {}").format(self.object.name)})
return context
+4 -4
View File
@@ -150,10 +150,10 @@ def list_quotas(request):
domains = Domain.objects.get_for_admin(request.user)
domains = domains.exclude(quota=0)
if sort_order in ["name", "quota"]:
domains = domains.order_by("{}{}".format(sort_dir, sort_order))
domains = domains.order_by(f"{sort_dir}{sort_order}")
elif sort_order == "allocated_quota":
domains = domains.annotate(allocated_quota=Sum("mailbox__quota")).order_by(
"{}{}".format(sort_dir, sort_order)
f"{sort_dir}{sort_order}"
)
page = get_listing_page(domains, request.GET.get("page", 1))
context = {
@@ -182,7 +182,7 @@ def list_logs(request):
)
else:
logs = ml_models.Maillog.objects.all()
logs = logs.order_by("{}{}".format(sort_dir, sort_order))
logs = logs.order_by(f"{sort_dir}{sort_order}")
if search:
logs = logs.filter(
Q(sender__icontains=search)
@@ -266,7 +266,7 @@ class DomainDetailView(auth_mixins.PermissionRequiredMixin, generic.DetailView):
def get_context_data(self, **kwargs):
"""Include extra widgets."""
context = super(DomainDetailView, self).get_context_data(**kwargs)
context = super().get_context_data(**kwargs)
result = signals.extra_domain_dashboard_widgets.send(
self.__class__, user=self.request.user, domain=self.object
)
+1 -1
View File
@@ -27,7 +27,7 @@ def _export(content, filename):
resp = HttpResponse(content)
resp["Content-Type"] = "text/csv"
resp["Content-Length"] = len(content)
resp["Content-Disposition"] = 'attachment; filename="{}"'.format(filename)
resp["Content-Disposition"] = f'attachment; filename="{filename}"'
return resp
+8 -8
View File
@@ -75,7 +75,7 @@ def list_quotas(request):
)
mboxes = mboxes.exclude(quota=0)
if sort_order in ["address", "quota"]:
mboxes = mboxes.order_by("%s%s" % (sort_dir, sort_order))
mboxes = mboxes.order_by(f"{sort_dir}{sort_order}")
elif sort_order in ("quota_value__bytes", "quota_usage"):
db_type = db_type()
if db_type == "mysql":
@@ -85,9 +85,9 @@ def list_quotas(request):
if sort_order == "quota_value__bytes":
mboxes = mboxes.extra(
select={"quota_value__bytes": "admin_quota.bytes"},
where=["admin_quota.username=%s" % where],
where=[f"admin_quota.username={where}"],
tables=["admin_quota", "admin_domain"],
order_by=["%s%s" % (sort_dir, sort_order)],
order_by=[f"{sort_dir}{sort_order}"],
)
else:
if db_type == "postgres":
@@ -102,9 +102,9 @@ def list_quotas(request):
)
mboxes = mboxes.extra(
select={"quota_usage": select},
where=["admin_quota.username=%s" % where],
where=[f"admin_quota.username={where}"],
tables=["admin_quota", "admin_domain"],
order_by=["%s%s" % (sort_dir, sort_order)],
order_by=[f"{sort_dir}{sort_order}"],
)
else:
raise BadRequest(_("Invalid request"))
@@ -199,7 +199,7 @@ def remove_permission(request):
account = User.objects.get(pk=daid)
domain = Domain.objects.get(pk=domid)
except (User.DoesNotExist, Domain.DoesNotExist):
raise BadRequest(_("Invalid request"))
raise BadRequest(_("Invalid request")) from None
if not request.user.can_access(account) or not request.user.can_access(domain):
raise PermDeniedException
domain.remove_admin(account)
@@ -215,14 +215,14 @@ class AccountDetailView(auth_mixins.PermissionRequiredMixin, generic.DetailView)
def has_permission(self):
"""Check object-level access."""
result = super(AccountDetailView, self).has_permission()
result = super().has_permission()
if not result:
return result
return self.request.user.can_access(self.get_object())
def get_context_data(self, **kwargs):
"""Add information to context."""
context = super(AccountDetailView, self).get_context_data(**kwargs)
context = super().get_context_data(**kwargs)
del context["user"]
result = signals.extra_account_dashboard_widgets.send(
self.__class__, user=self.request.user, account=self.object
-7
View File
@@ -1,8 +1,5 @@
"""Import related views."""
import csv
import io
from reversion import revisions as reversion
from django.contrib.auth.decorators import (
@@ -10,15 +7,11 @@ from django.contrib.auth.decorators import (
permission_required,
user_passes_test,
)
from django.db import transaction
from django.shortcuts import render
from django.urls import reverse
from django.utils.encoding import smart_str
from django.utils.translation import gettext as _
from modoboa.lib.exceptions import Conflict, ModoboaException
from .. import lib
from .. import signals
from ..forms import ImportDataForm, ImportIdentitiesForm
+1 -1
View File
@@ -10,7 +10,7 @@ class CalendarsConfig(AppConfig):
verbose_name = "Modoboa calendars"
def ready(self):
from modoboa.calendars import handlers
from modoboa.calendars import handlers # noqa
from modoboa.calendars.app_settings import load_settings
load_settings()
+3 -6
View File
@@ -2,10 +2,8 @@
from importlib import import_module
from modoboa.lib.cryptutils import decrypt
class CalendarBackend(object):
class CalendarBackend:
"""Base backend class."""
def __init__(self, calendar=None):
@@ -31,11 +29,10 @@ class CalendarBackend(object):
def get_backend(name, *args, **kwargs):
"""Return a backend instance."""
module = import_module("modoboa.calendars.backends.{}".format(name))
return getattr(module, "{}Backend".format(name.capitalize()))(*args, **kwargs)
module = import_module(f"modoboa.calendars.backends.{name}")
return getattr(module, f"{name.capitalize()}Backend")(*args, **kwargs)
def get_backend_from_request(name, request, calendar=None):
"""Return a backend instance from a request."""
# password = decrypt(request.session["password"])
return get_backend(name, request.user.email, request.auth, calendar=calendar)
+4 -4
View File
@@ -94,7 +94,7 @@ class Caldav_Backend(CalendarBackend):
def update_event(self, uid, original_data):
"""Update an existing event."""
data = dict(original_data)
url = "{}/{}.ics".format(self.remote_cal.url.geturl(), uid)
url = f"{self.remote_cal.url.geturl()}/{uid}.ics"
cal = self.remote_cal.event_by_url(url)
orig_evt = cal.vobject_instance.vevent
if "title" in data:
@@ -125,7 +125,7 @@ class Caldav_Backend(CalendarBackend):
# Calendar has been changed, remove old event first.
self.remote_cal.client.delete(url)
remote_cal = Calendar(self.client, data["calendar"].encoded_path)
url = "{}/{}.ics".format(remote_cal.url.geturl(), uid)
url = f"{remote_cal.url.geturl()}/{uid}.ics"
else:
remote_cal = self.remote_cal
remote_cal.add_event(cal.instance)
@@ -133,7 +133,7 @@ class Caldav_Backend(CalendarBackend):
def get_event(self, uid):
"""Retrieve and event using its uid."""
url = "{}/{}.ics".format(self.remote_cal.url.geturl(), uid)
url = f"{self.remote_cal.url.geturl()}/{uid}.ics"
event = self.remote_cal.event_by_url(url)
return self._serialize_event(event)
@@ -147,7 +147,7 @@ class Caldav_Backend(CalendarBackend):
def delete_event(self, uid):
"""Delete an event using its uid."""
url = "{}/{}.ics".format(self.remote_cal.url.geturl(), uid)
url = f"{self.remote_cal.url.geturl()}/{uid}.ics"
self.remote_cal.client.delete(url)
def import_events(self, fp):
+2 -2
View File
@@ -12,7 +12,7 @@ class UserCalendarFactory(factory.django.DjangoModelFactory):
class Meta:
model = models.UserCalendar
name = factory.Sequence(lambda n: "User calendar %s" % n)
name = factory.Sequence(lambda n: f"User calendar {n}")
mailbox = factory.SubFactory(admin_factories.MailboxFactory)
@@ -21,7 +21,7 @@ class SharedCalendarFactory(factory.django.DjangoModelFactory):
class Meta:
model = models.SharedCalendar
name = factory.Sequence(lambda n: "Shared calendar %s" % n)
name = factory.Sequence(lambda n: f"Shared calendar {n}")
domain = factory.SubFactory(admin_factories.DomainFactory)
+2 -2
View File
@@ -48,7 +48,7 @@ def set_user_calendar_path(sender, instance, **kwargs):
"""Set path at creation."""
if instance.pk:
return
instance._path = "{}/{}".format(instance.mailbox.full_address, instance.name)
instance._path = f"{instance.mailbox.full_address}/{instance.name}"
if not instance.access_token:
instance.access_token = token_hex(16)
@@ -58,6 +58,6 @@ def set_shared_calendar_path(sender, instance, **kwargs):
"""Set path at creation."""
if instance.pk:
return
instance._path = "{}/{}".format(instance.domain.name, instance.name)
instance._path = f"{instance.domain.name}/{instance.name}"
if not instance.access_token:
instance.access_token = token_hex(16)
@@ -30,21 +30,20 @@ class Command(BaseCommand):
def _generate_acr(self, name, user, collection, perm="rw", comment=None):
"""Write a new access control rule to the config file."""
if comment is not None:
self._cfgfile.write("\n# %s" % comment)
self._cfgfile.write(f"\n# {comment}")
self._cfgfile.write(
"""
[%s]
user = %s
collection = %s
permissions = %s
f"""
[{name}]
user = {user}
collection = {collection}
permissions = {perm}
"""
% (name, user, collection, perm)
)
def _user_access_rules(self):
"""Create user access rules."""
for acr in models.AccessRule.objects.select_related().all():
section = "%s-to-%s-acr" % (acr.mailbox, acr.calendar)
section = f"{acr.mailbox}-to-{acr.calendar}-acr"
permission = ""
if acr.read:
permission += "r"
@@ -61,7 +60,7 @@ permissions = %s
"""Create token access rules."""
for model in [models.UserCalendar, models.SharedCalendar]:
for calendar in model.objects.all():
section = "token-{}-access".format(calendar._path)
section = f"token-{calendar._path}-access"
self._generate_acr(
section,
calendar.access_token,
@@ -73,15 +72,15 @@ permissions = %s
def _super_admin_rules(self):
"""Generate access rules for super administrators."""
for sa in User.objects.filter(is_superuser=True):
section = "sa-%s-acr" % sa.username
section = f"sa-{sa.username}-acr"
self._generate_acr(section, sa.username, ".*")
def _domain_admin_rules(self):
"""Generate access rules for domain adminstrators."""
for da in User.objects.filter(groups__name="DomainAdmins"):
for domain in Domain.objects.get_for_admin(da):
section = "da-%s-to-%s-acr" % (da.email, domain.name)
self._generate_acr(section, da.email, "%s/user/.*" % domain.name)
section = f"da-{da.email}-to-{domain.name}-acr"
self._generate_acr(section, da.email, f"{domain.name}/user/.*")
def _generate_file(self, target):
"""
@@ -89,11 +88,10 @@ permissions = %s
"""
self._cfgfile = open(target, "w")
self._cfgfile.write(
"""# Rights management file for Radicale
# This file was generated by Modoboa on %s
f"""# Rights management file for Radicale
# This file was generated by Modoboa on {datetime.datetime.today()}
# DO NOT EDIT MANUALLY!
"""
% datetime.datetime.today()
)
allow_calendars_administration = param_tools.get_global_parameter(
+2 -2
View File
@@ -31,7 +31,7 @@ END:VCALENDAR
"""
class Url(object):
class Url:
def __init__(self, path):
self.path = path
@@ -85,5 +85,5 @@ class DAVClientMock:
def proppatch(self, url, body, dummy=None):
return Response(200)
def request(self, url, method="GET", body="", headers={}):
def request(self, url, method="GET", body="", headers=None):
return Response(200)
+4 -3
View File
@@ -53,7 +53,7 @@ class Calendar(models.Model):
@property
def share_url(self):
"""Return calendar share url."""
return "{}?token={}".format(self.full_url, self.access_token)
return f"{self.full_url}?token={self.access_token}"
@property
def encoded_url(self):
@@ -162,6 +162,7 @@ class AccessRule(models.Model):
access = "r" if self.read else ""
access += "w" if self.write else ""
return smart_str(
"%s access rule to %s -> %s"
% (self.mailbox, self.calendar, access if access else "no access")
"{} access rule to {} -> {}".format(
self.mailbox, self.calendar, access if access else "no access"
)
)
+16 -18
View File
@@ -1,5 +1,3 @@
# coding: utf-8
"""Radicale extension unit tests."""
import os
@@ -24,13 +22,13 @@ from . import models
from . import mocks
class TestDataMixin(object):
class TestDataMixin:
"""Create some data."""
@classmethod
def setUpTestData(cls):
"""Create test data."""
super(TestDataMixin, cls).setUpTestData()
super().setUpTestData()
admin_factories.populate_database()
cls.account = core_models.User.objects.get(username="user@test.com")
cls.calendar = factories.UserCalendarFactory(mailbox=cls.account.mailbox)
@@ -60,12 +58,12 @@ class AccessRuleTestCase(ModoTestCase):
@classmethod
def setUpTestData(cls):
"""Create test data."""
super(AccessRuleTestCase, cls).setUpTestData()
super().setUpTestData()
populate_database()
def setUp(self):
"""Initialize tests."""
super(AccessRuleTestCase, self).setUp()
super().setUp()
self.rights_file_path = tempfile.mktemp()
self.set_global_parameter(
"rights_file_path", self.rights_file_path, app="calendars"
@@ -394,12 +392,12 @@ class EventViewSetTestCase(TestDataMixin, ModoAPITestCase):
self.client.post(reverse("core:login"), data)
# FIXME: drf nested routers does not handle reverse() properly
url = "/api/v2/user-calendars/{}/events/{}/".format(self.calendar.pk, 1234)
url = f"/api/v2/user-calendars/{self.calendar.pk}/events/{1234}/"
response = self.client.get(url)
self.assertEqual(response.status_code, 200)
self.assertTrue(response.json()["title"].startswith("Bastille"))
url = "/api/v2/user-calendars/{}/events/".format(self.calendar.pk)
url = f"/api/v2/user-calendars/{self.calendar.pk}/events/"
url = "{}?start={}&end={}".format(url, "20060712T182145Z", "20070712T182145Z")
response = self.client.get(url)
self.assertEqual(response.status_code, 200)
@@ -411,12 +409,12 @@ class EventViewSetTestCase(TestDataMixin, ModoAPITestCase):
self.client.post(reverse("core:login"), data)
# FIXME: drf nested routers does not handle reverse() properly
url = "/api/v2/shared-calendars/{}/events/{}/".format(self.scalendar.pk, 1234)
url = f"/api/v2/shared-calendars/{self.scalendar.pk}/events/{1234}/"
response = self.client.get(url)
self.assertEqual(response.status_code, 200)
self.assertTrue(response.json()["title"].startswith("Bastille"))
url = "/api/v2/shared-calendars/{}/events/".format(self.scalendar.pk)
url = f"/api/v2/shared-calendars/{self.scalendar.pk}/events/"
url = "{}?start={}&end={}".format(url, "20060712T182145Z", "20070712T182145Z")
response = self.client.get(url)
self.assertEqual(response.status_code, 200)
@@ -427,7 +425,7 @@ class EventViewSetTestCase(TestDataMixin, ModoAPITestCase):
data = {"username": "user@test.com", "password": "toto"}
self.client.post(reverse("core:login"), data)
url = "/api/v2/user-calendars/{}/events/".format(self.calendar.pk)
url = f"/api/v2/user-calendars/{self.calendar.pk}/events/"
data = {
"title": "Test event",
"start": "2018-03-27T00:00:00Z",
@@ -441,7 +439,7 @@ class EventViewSetTestCase(TestDataMixin, ModoAPITestCase):
self.assertEqual(response.status_code, 201)
self.assertIn("id", response.json())
url = "/api/v2/shared-calendars/{}/events/".format(self.scalendar.pk)
url = f"/api/v2/shared-calendars/{self.scalendar.pk}/events/"
data = {
"title": "Test event 2",
"start": "2018-03-27T00:00:00Z",
@@ -460,7 +458,7 @@ class EventViewSetTestCase(TestDataMixin, ModoAPITestCase):
data = {"username": "user@test.com", "password": "toto"}
self.client.post(reverse("core:login"), data)
url = "/api/v2/user-calendars/{}/events/1234/".format(self.calendar.pk)
url = f"/api/v2/user-calendars/{self.calendar.pk}/events/1234/"
data = {
"title": "Test event",
"start": "2018-03-27T00:00:00Z",
@@ -483,7 +481,7 @@ class EventViewSetTestCase(TestDataMixin, ModoAPITestCase):
data = {"username": "user@test.com", "password": "toto"}
self.client.post(reverse("core:login"), data)
url = "/api/v2/user-calendars/{}/events/1234/".format(self.calendar.pk)
url = f"/api/v2/user-calendars/{self.calendar.pk}/events/1234/"
data = {
"start": "2018-03-27T00:00:00Z",
"end": "2018-03-28T00:00:00Z",
@@ -506,7 +504,7 @@ class EventViewSetTestCase(TestDataMixin, ModoAPITestCase):
"calendar": self.scalendar.pk,
"new_calendar_type": "shared",
}
url = "/api/v2/user-calendars/{}/events/1234/".format(self.calendar.pk)
url = f"/api/v2/user-calendars/{self.calendar.pk}/events/1234/"
response = self.client.put(url, data=data, format="json")
self.assertEqual(response.status_code, 200)
@@ -515,7 +513,7 @@ class EventViewSetTestCase(TestDataMixin, ModoAPITestCase):
data = {"username": "user@test.com", "password": "toto"}
self.client.post(reverse("core:login"), data)
url = "/api/v2/shared-calendars/{}/events/1234/".format(self.scalendar.pk)
url = f"/api/v2/shared-calendars/{self.scalendar.pk}/events/1234/"
response = self.client.delete(url)
self.assertEqual(response.status_code, 200)
@@ -546,7 +544,7 @@ class AttendeeViewSetTestCase(ModoAPITestCase):
@classmethod
def setUpTestData(cls):
super(AttendeeViewSetTestCase, cls).setUpTestData()
super().setUpTestData()
populate_database()
cls.account = core_models.User.objects.get(username="user@test.com")
@@ -567,7 +565,7 @@ class MailboxViewSetTestCase(ModoAPITestCase):
@classmethod
def setUpTestData(cls):
super(MailboxViewSetTestCase, cls).setUpTestData()
super().setUpTestData()
populate_database()
cls.account = core_models.User.objects.get(username="user@test.com")
+1 -1
View File
@@ -7,7 +7,7 @@ class ModoboaContactsConfig(AppConfig):
name = "modoboa.contacts"
def ready(self):
from . import handlers
from . import handlers # noqa
from modoboa.contacts.app_settings import load_settings
load_settings()
+1 -3
View File
@@ -51,9 +51,7 @@ class ContactFactory(factory.django.DjangoModelFactory):
first_name = "Homer"
last_name = "Simpson"
display_name = factory.LazyAttribute(
lambda c: "{}{}".format(c.first_name, c.last_name)
)
display_name = factory.LazyAttribute(lambda c: f"{c.first_name}{c.last_name}")
@factory.post_generation
def categories(self, create, extracted, **dummy_kwargs):
+3 -3
View File
@@ -53,9 +53,9 @@ nsmap = {
def ns(prefix, tag=None):
name = "{%s}" % nsmap[prefix]
name = f"{{{nsmap[prefix]}}}"
if tag is not None:
name = "%s%s" % (name, tag)
name = f"{name}{tag}"
return name
@@ -91,7 +91,7 @@ class SyncCollectionQuery(elements.base.BaseElement):
tag = ns("D", "sync-collection")
class PyCardDAV(object):
class PyCardDAV:
"""class for interacting with a CardDAV server
Since PyCardDAV relies heavily on Requests [1] its SSL verification is also
@@ -44,7 +44,7 @@ class Command(BaseCommand):
).first()
if not addressbook:
raise CommandError(
"Address Book for email '%s' not found" % options["email"]
"Address Book for email '{}' not found".format(options["email"])
)
try:
import_csv_file(
@@ -55,5 +55,5 @@ class Command(BaseCommand):
options.get("carddav_password"),
)
except RuntimeError as err:
raise CommandError(err)
raise CommandError(err) from None
self.stdout.write(self.style.SUCCESS("File was imported successfuly"))
+3 -5
View File
@@ -98,10 +98,10 @@ def report_mock(url, request):
def get_mock(url, request):
"""Simulate a GET request."""
uid = url.path.split("/")[-1]
content = """
content = f"""
BEGIN:VCARD
VERSION:3.0
UID:{}
UID:{uid}
N:Gump;Forrest
FN:Forrest Gump
ORG:Bubba Gump Shrimp Co.
@@ -111,7 +111,5 @@ TEL;TYPE=HOME;VOICE:(404) 555-1212
ADR;TYPE=HOME:;;42 Plantation St.;Baytown;LA;30314;United States of America
EMAIL;TYPE=PREF,INTERNET:forrestgump@example.com
END:VCARD
""".format(
uid
)
"""
return {"status_code": 200, "content": content}
+4 -6
View File
@@ -82,18 +82,16 @@ class Contact(models.Model):
"""Set uid for new object."""
super().__init__(*args, **kwargs)
if not self.pk:
self.uid = "{}.vcf".format(uuid.uuid4())
self.uid = f"{uuid.uuid4()}.vcf"
@property
def url(self):
return "{}/{}.vcf".format(self.addressbook.url, self.uid)
return f"{self.addressbook.url}/{self.uid}.vcf"
def to_vcard(self):
"""Convert this contact to a vCard."""
card = vobject.vCard()
card.add("prodid").value = "-//Modoboa//Contacts plugin {}//EN".format(
__version__
)
card.add("prodid").value = f"-//Modoboa//Contacts plugin {__version__}//EN"
card.add("uid").value = self.uid
card.add("n").value = vobject.vcard.Name(
family=self.last_name, given=self.first_name
@@ -193,4 +191,4 @@ class PhoneNumber(models.Model):
db_table = "modoboa_contacts_phonenumber"
def __str__(self):
return "{}: {}".format(self.type, self.number)
return f"{self.type}: {self.number}"
+1 -1
View File
@@ -38,7 +38,7 @@ class EmailAddressWithNameSerializer(serializers.ModelSerializer):
"""Return display name."""
if obj.contact.display_name:
return obj.contact.display_name
return "{} {}".format(obj.contact.first_name, obj.contact.last_name)
return f"{obj.contact.first_name} {obj.contact.last_name}"
class PhoneNumberSerializer(serializers.ModelSerializer):
+9 -10
View File
@@ -1,4 +1,3 @@
# coding: utf-8
"""Contacts backend tests."""
import os
@@ -25,7 +24,7 @@ class TestDataMixin:
@classmethod
def setUpTestData(cls):
"""Create test data."""
super(TestDataMixin, cls).setUpTestData()
super().setUpTestData()
admin_factories.populate_database()
cls.user = core_models.User.objects.get(username="user@test.com")
cls.addressbook = cls.user.addressbook_set.first()
@@ -172,14 +171,14 @@ class ContactViewSetTestCase(TestDataMixin, ModoAPITestCase):
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data), 3)
response = self.client.get("{}?search=homer".format(url))
response = self.client.get(f"{url}?search=homer")
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data), 1)
def test_filter_by_category(self):
"""Check filtering."""
url = reverse("api:contact-list")
response = self.client.get("{}?category={}".format(url, self.category.name))
response = self.client.get(f"{url}?category={self.category.name}")
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data), 1)
@@ -300,15 +299,15 @@ class EmailAddressViewSetTestCase(TestDataMixin, ModoAPITestCase):
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data), 3)
response = self.client.get("{}?search=homer".format(url))
response = self.client.get(f"{url}?search=homer")
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data), 1)
response = self.client.get("{}?search=Marge".format(url))
response = self.client.get(f"{url}?search=Marge")
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data), 1)
response = self.client.get("{}?search=Simpson".format(url))
response = self.client.get(f"{url}?search=Simpson")
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data), 3)
@@ -339,7 +338,7 @@ class ImportTestCase(TestDataMixin, ModoTestCase):
def test_import_from_outlook_auto(self):
management.call_command("import_contacts", "user@test.com", self.path)
address = models.EmailAddress.objects.get(address="toto@titi.com")
phone = models.PhoneNumber.objects.get(number="12345678")
models.PhoneNumber.objects.get(number="12345678")
self.assertEqual(address.contact.first_name, "Toto Tata")
self.assertEqual(address.contact.addressbook.user.email, "user@test.com")
self.assertEqual(address.contact.address, "Street 1 Street 2")
@@ -352,7 +351,7 @@ class ImportTestCase(TestDataMixin, ModoTestCase):
backend="outlook",
)
address = models.EmailAddress.objects.get(address="toto@titi.com")
phone = models.PhoneNumber.objects.get(number="12345678")
models.PhoneNumber.objects.get(number="12345678")
self.assertEqual(address.contact.first_name, "Toto Tata")
self.assertEqual(address.contact.addressbook.user.email, "user@test.com")
self.assertEqual(address.contact.address, "Street 1 Street 2")
@@ -366,7 +365,7 @@ class ImportTestCase(TestDataMixin, ModoTestCase):
carddav_password="Toto1234",
)
address = models.EmailAddress.objects.get(address="toto@titi.com")
phone = models.PhoneNumber.objects.get(number="12345678")
models.PhoneNumber.objects.get(number="12345678")
self.assertEqual(address.contact.first_name, "Toto Tata")
self.assertEqual(address.contact.addressbook.user.email, "user@test.com")
self.assertEqual(address.contact.address, "Street 1 Street 2")
+4 -4
View File
@@ -29,7 +29,7 @@ class AccountViewSet(GetThrottleViewsetMixin, viewsets.ViewSet):
def tfa_setup(self, request):
"""Initiate TFA setup."""
instance, created = TOTPDevice.objects.get_or_create(
user=request.user, defaults={"name": "{} TOTP device".format(request.user)}
user=request.user, defaults={"name": f"{request.user} TOTP device"}
)
return response.Response()
@@ -43,9 +43,9 @@ class AccountViewSet(GetThrottleViewsetMixin, viewsets.ViewSet):
serializer.is_valid(raise_exception=True)
# create static device for recovery purposes
device = StaticDevice.objects.create(
user=request.user, name="{} static device".format(request.user)
user=request.user, name=f"{request.user} static device"
)
for cpt in range(10):
for _cpt in range(10):
token = StaticToken.random_token()
device.token_set.create(token=token)
django_otp.login(self.request, request.user.totpdevice_set.first())
@@ -82,7 +82,7 @@ class AccountViewSet(GetThrottleViewsetMixin, viewsets.ViewSet):
if device is None:
return response.Response({"error": _("2FA is not enabled")}, status=403)
device.token_set.all().delete()
for cpt in range(10):
for _cpt in range(10):
token = StaticToken.random_token()
device.token_set.create(token=token)
return response.Response(
+6 -6
View File
@@ -180,21 +180,21 @@ class CoreGlobalParametersSerializer(serializers.Serializer):
try:
value % {"user": "toto"}
except (KeyError, ValueError):
raise serializers.ValidationError(_("Invalid syntax"))
raise serializers.ValidationError(_("Invalid syntax")) from None
return value
def validate_ldap_sync_account_dn_template(self, value):
try:
value % {"user": "toto"}
except (KeyError, ValueError):
raise serializers.ValidationError(_("Invalid syntax"))
raise serializers.ValidationError(_("Invalid syntax")) from None
return value
def validate_ldap_search_filter(self, value):
try:
value % {"user": "toto"}
except (KeyError, ValueError, TypeError):
raise serializers.ValidationError(_("Invalid syntax"))
raise serializers.ValidationError(_("Invalid syntax")) from None
return value
def validate_rounds_number(self, value):
@@ -416,7 +416,7 @@ class PasswordRecoverySmsConfirmSerializer(serializers.Serializer):
except KeyError:
raise CustomValidationError(
{"reason": "totp secret not set in session"}, 403
)
) from None
if not oath.accept_totp(self.context["totp_secret"], data["sms_totp"])[0]:
raise CustomValidationError({"reason": "Wrong totp"}, 400)
@@ -445,7 +445,7 @@ class PasswordRecoverySmsResendSerializer(serializers.Serializer):
except KeyError:
raise CustomValidationError(
{"reason": "totp secret not set in session"}, status.HTTP_403_FORBIDDEN
)
) from None
# Attempt to get user, will raise an error if pk is not valid
self.context["user"] = User.objects.filter(
@@ -501,7 +501,7 @@ class PasswordRecoveryConfirmSerializer(serializers.Serializer):
try:
password_validation.validate_password(data["new_password1"], user)
except djangoValidationError as e:
raise PasswordRequirementsFailure(e.messages)
raise PasswordRequirementsFailure(e.messages) from None
self.context["user"] = user
return super().validate(data)
+2 -2
View File
@@ -25,11 +25,11 @@ from . import serializers
logger = logging.getLogger("modoboa.auth")
def delete_cache_key(class_target, throttles, request):
def delete_cache_key(class_target, throttles: list, request) -> None:
"""Attempt to delete cache key from throttling on login/password reset success."""
for throttle in throttles:
if type(throttle) == class_target:
if isinstance(throttle, class_target):
throttle.reset_cache(request)
return
+2 -2
View File
@@ -32,12 +32,12 @@ def create_static_tokens(request):
device = request.user.staticdevice_set.first()
if device is None:
device = StaticDevice.objects.create(
user=request.user, name="{} static device".format(request.user)
user=request.user, name=f"{request.user} static device"
)
elif device is not None:
return None
tokens = []
for cpt in range(10):
for _cpt in range(10):
token = StaticToken.random_token()
device.token_set.create(token=token)
tokens.append(token)
+6 -9
View File
@@ -512,7 +512,7 @@ class GeneralParametersForm(param_forms.AdminParametersForm):
}
def __init__(self, *args, **kwargs):
super(GeneralParametersForm, self).__init__(*args, **kwargs)
super().__init__(*args, **kwargs)
self.fields["default_top_redirection"].choices = enabled_applications()
self._add_visibilty_rules(sms_backends.get_all_backend_visibility_rules())
self.fields["password_scheme"].choices = get_password_scheme()
@@ -532,7 +532,7 @@ class GeneralParametersForm(param_forms.AdminParametersForm):
try:
tpl % {"user": "toto"}
except (KeyError, ValueError):
raise forms.ValidationError(_("Invalid syntax"))
raise forms.ValidationError(_("Invalid syntax")) from None
return tpl
def clean_ldap_sync_account_dn_template(self):
@@ -540,7 +540,7 @@ class GeneralParametersForm(param_forms.AdminParametersForm):
try:
tpl % {"user": "toto"}
except (KeyError, ValueError):
raise forms.ValidationError(_("Invalid syntax"))
raise forms.ValidationError(_("Invalid syntax")) from None
return tpl
def clean_ldap_search_filter(self):
@@ -548,7 +548,7 @@ class GeneralParametersForm(param_forms.AdminParametersForm):
try:
ldap_filter % {"user": "toto"}
except (KeyError, ValueError, TypeError):
raise forms.ValidationError(_("Invalid syntax"))
raise forms.ValidationError(_("Invalid syntax")) from None
return ldap_filter
def clean_rounds_number(self):
@@ -613,10 +613,7 @@ class GeneralParametersForm(param_forms.AdminParametersForm):
{"first_name": "givenName", "email": "mail", "last_name": "sn"},
)
ldap_uri = "ldaps://" if values["ldap_secured"] == "ssl" else "ldap://"
ldap_uri += "%s:%s" % (
values[backend.srv_address_setting_name],
values[backend.srv_port_setting_name],
)
ldap_uri += f"{values[backend.srv_address_setting_name]}:{values[backend.srv_port_setting_name]}"
setattr(settings, backend.setting_fullname("SERVER_URI"), ldap_uri)
if values["ldap_secured"] == "starttls":
setattr(settings, backend.setting_fullname("START_TLS"), True)
@@ -680,7 +677,7 @@ class GeneralParametersForm(param_forms.AdminParametersForm):
Doing so, we can use the django_auth_ldap module.
"""
try:
import ldap
import ldap # noqa
ldap_available = True
except ImportError:
+8 -7
View File
@@ -7,7 +7,7 @@ from django.template import Context, Template
from django.utils.encoding import smart_str
class Command(object):
class Command:
"""Base command class.
A valid administrative command must inherit from this class.
@@ -25,7 +25,7 @@ class Command(object):
{"BACKEND": ("django.template.backends.django.DjangoTemplates")}
]
)
self._templates_dir = "%s/templates" % os.path.dirname(__file__)
self._templates_dir = f"{os.path.dirname(__file__)}/templates"
def _render_template(self, tplfile, env):
"""Render an HTML template."""
@@ -70,7 +70,7 @@ def scan_for_commands(dirname=""):
else:
cmdclassname = cmdname.capitalize()
try:
cmdclass = getattr(cmdmod, "%sCommand" % cmdclassname)
cmdclass = getattr(cmdmod, f"{cmdclassname}Command")
except AttributeError:
continue
result[cmdname] = cmdclass
@@ -83,9 +83,10 @@ def handle_command_line():
parser = argparse.ArgumentParser(
description="A set of utilities to ease the installation of Modoboa.",
epilog="""Available commands:
%s
"""
% "\n".join(["\t%s" % c for c in sorted(commands)]),
{}
""".format(
"\n".join([f"\t{c}" for c in sorted(commands)])
),
)
parser.add_argument(
"--verbose", action="store_true", help="Activate verbose output"
@@ -94,7 +95,7 @@ def handle_command_line():
(args, remaining) = parser.parse_known_args()
if args.command not in commands:
print("Unknown command '%s'" % args.command, file=sys.stderr)
print(f"Unknown command '{args.command}'", file=sys.stderr)
sys.exit(1)
commands[args.command](commands, verbose=args.verbose).run(remaining)
+13 -13
View File
@@ -114,10 +114,10 @@ class DeployCommand(Command):
if p.returncode:
if output:
print(
"\n".join([l.decode() for l in output if l is not None]),
"\n".join([line.decode() for line in output if line is not None]),
file=sys.stderr,
)
print("%s failed, check your configuration" % cmd, file=sys.stderr)
print(f"{cmd} failed, check your configuration", file=sys.stderr)
def ask_db_info(self, name="default"):
"""Prompt the user for database information
@@ -127,7 +127,7 @@ class DeployCommand(Command):
:param name: the connection name
"""
print("Configuring database connection: %s" % name)
print(f"Configuring database connection: {name}")
info = {
"conn_name": name,
"ENGINE": input("Database type (mysql, postgres or sqlite3): "),
@@ -137,7 +137,7 @@ class DeployCommand(Command):
if info["ENGINE"] == "sqlite3":
info["ENGINE"] = "django.db.backends.sqlite3"
info["NAME"] = "%s.db" % name
info["NAME"] = f"{name}.db"
return info
if info["ENGINE"] == "postgres":
info["ENGINE"] = "django.db.backends.postgresql"
@@ -146,7 +146,7 @@ class DeployCommand(Command):
info["ENGINE"] = "django.db.backends.mysql"
default_port = 3306
info["HOST"] = input("Database host (default: 'localhost'): ")
info["PORT"] = input("Database port (default: '%s'): " % default_port)
info["PORT"] = input(f"Database port (default: '{default_port}'): ")
# leave port setting empty, if default value is supplied and
# leave it to django
if info["PORT"] == default_port:
@@ -181,7 +181,7 @@ class DeployCommand(Command):
for extension in extensions:
module = __import__(extension[1], locals(), globals(), [])
basedir = os.path.dirname(module.__file__)
if not os.path.exists("{0}/settings.py".format(basedir)):
if not os.path.exists(f"{basedir}/settings.py"):
continue
extra_settings.append(extension[1])
return extra_settings
@@ -189,7 +189,7 @@ class DeployCommand(Command):
def handle(self, parsed_args):
django.setup()
management.call_command("startproject", parsed_args.name, verbosity=False)
path = "%(name)s/%(name)s" % {"name": parsed_args.name}
path = f"{parsed_args.name}/{parsed_args.name}"
sys.path.append(parsed_args.name)
conn_tpl = Template(DBCONN_TPL)
@@ -240,7 +240,7 @@ class DeployCommand(Command):
mod = __import__(parsed_args.name, globals(), locals(), [smart_str("settings")])
tpl = self._render_template(
"%s/settings.py.tpl" % self._templates_dir,
f"{self._templates_dir}/settings.py.tpl",
{
"db_connections": connections,
"secret_key": management.utils.get_random_secret_key(),
@@ -254,15 +254,15 @@ class DeployCommand(Command):
"extra_settings": extra_settings,
},
)
with open("%s/settings.py" % path, "w") as fp:
with open(f"{path}/settings.py", "w") as fp:
fp.write(tpl)
generate_rsa_private_key(parsed_args.name)
shutil.copyfile("%s/urls.py.tpl" % self._templates_dir, "%s/urls.py" % path)
os.mkdir("%s/media" % parsed_args.name)
shutil.copyfile(f"{self._templates_dir}/urls.py.tpl", f"{path}/urls.py")
os.mkdir(f"{parsed_args.name}/media")
if isfile("%s/settings.pyc" % path):
os.unlink("%s/settings.pyc" % path)
if isfile(f"{path}/settings.pyc"):
os.unlink(f"{path}/settings.pyc")
self._exec_django_command("migrate", parsed_args.name, "--noinput")
self._exec_django_command(
"load_initial_data",
+3 -3
View File
@@ -8,14 +8,14 @@ class HelpCommand(Command):
help = "Display the help message associated to a specific command" # NOQA:A003
def __init__(self, *args, **kwargs):
super(HelpCommand, self).__init__(*args, **kwargs)
super().__init__(*args, **kwargs)
self._parser.add_argument("name", type=str, help="A command name")
def handle(self, parsed_args):
if parsed_args.name not in self._commands:
print("Unknown command: %s" % parsed_args.name, file=sys.stderr)
print(f"Unknown command: {parsed_args.name}", file=sys.stderr)
sys.exit(1)
cmd = self._commands[parsed_args.name](self._commands)
cmd._parser.prog = "modoboa-admin.py %s" % parsed_args.name
cmd._parser.prog = f"modoboa-admin.py {parsed_args.name}"
cmd._parser.description = cmd.help
cmd._parser.print_help()
+1 -1
View File
@@ -124,7 +124,7 @@ class ExtensionsPool:
root = ""
pattern = "{}.urls_api"
else:
root = r"^{}/".format(ext.get_url())
root = rf"^{ext.get_url()}/"
pattern = "{}.urls"
try:
result.append(re_path(root, include(pattern.format(ext_name))))
+4 -6
View File
@@ -1,7 +1,5 @@
"""Factories for core application."""
import base64
import factory
from django.contrib.auth.models import Group
@@ -12,7 +10,7 @@ from . import models
class PermissionFactory(factory.django.DjangoModelFactory):
"""A base factory to handle permissions."""
class Meta(object):
class Meta:
abstract = True
@factory.post_generation
@@ -25,7 +23,7 @@ class PermissionFactory(factory.django.DjangoModelFactory):
class GroupFactory(factory.django.DjangoModelFactory):
"""A factory to create Group instances."""
class Meta(object):
class Meta:
model = Group
name = "DefaultGroup"
@@ -34,7 +32,7 @@ class GroupFactory(factory.django.DjangoModelFactory):
class UserFactory(PermissionFactory):
"""A factory to create User instances."""
class Meta(object):
class Meta:
model = models.User
django_get_or_create = ("username",)
@@ -54,7 +52,7 @@ class UserFactory(PermissionFactory):
class LogFactory(factory.django.DjangoModelFactory):
"""Factory for Log."""
class Meta(object):
class Meta:
model = models.Log
logger = "modoboa.admin"
+1 -1
View File
@@ -133,7 +133,7 @@ class APIAccessForm(forms.Form):
def __init__(self, *args, **kwargs):
"""Initialize form."""
user = kwargs.pop("user")
super(APIAccessForm, self).__init__(*args, **kwargs)
super().__init__(*args, **kwargs)
self.fields["enable_api_access"].initial = hasattr(user, "auth_token")
@@ -72,7 +72,7 @@ class Command(BaseCommand):
return
action = "delete" if options["delete"] else "disable"
if not options["silent"]:
answer = input("Do you want to {} those accounts? (y/N) ".format(action))
answer = input(f"Do you want to {action} those accounts? (y/N) ")
if not answer.lower().startswith("y"):
return
if action == "disable":
@@ -38,7 +38,7 @@ class Command(BaseCommand):
self.verbose = options["verbose"]
log_maximum_age = param_tools.get_global_parameter("log_maximum_age")
self.__vprint("Deleting logs older than %d days..." % log_maximum_age)
self.__vprint(f"Deleting logs older than {log_maximum_age} days...")
limit = timezone.now() - datetime.timedelta(log_maximum_age)
Log.objects.filter(date_created__lt=limit).delete()
self.__vprint("Done.")
@@ -126,9 +126,7 @@ query = {{ query|safe }}
if os.path.exists(fullpath) and not force_overwrite:
if not self.__check_file(fullpath):
print(
"Cannot upgrade '{}' map because it has been modified.".format(
mapobject.filename
)
f"Cannot upgrade '{mapobject.filename}' map because it has been modified."
)
return self.__checksums[mapobject.filename]
mapcontent = utils.parse_map_file(fullpath)
@@ -91,7 +91,7 @@ class Command(BaseCommand):
extension.load_initial_data()
except Exception as e:
self.stderr.write(
"Unable to load initial data for '{}' ({}).".format(extname, str(e))
f"Unable to load initial data for '{extname}' ({str(e)})."
)
else:
signals.initial_data_loaded.send(sender=self.__class__, extname=extname)
@@ -141,7 +141,7 @@ class Command(BaseCommand):
base_frontend_dir = os.path.join(
os.path.dirname(__file__), "../../../frontend_dist/"
)
frontend_target_dir = "{}/frontend".format(settings.BASE_DIR)
frontend_target_dir = f"{settings.BASE_DIR}/frontend"
if os.path.exists(base_frontend_dir):
shutil.rmtree(frontend_target_dir, ignore_errors=True)
shutil.copytree(base_frontend_dir, frontend_target_dir)
@@ -1,4 +1,4 @@
from django.db import models, migrations
from django.db import migrations
class Migration(migrations.Migration):
+7 -5
View File
@@ -373,15 +373,17 @@ class User(AbstractUser):
if len(row) < 8 or not row[7].strip():
raise BadRequest(
_(
"The simple user '%s' must have a valid email address"
% self.username
"The simple user '{}' must have a valid email address".format(
self.username
)
)
)
if self.username != row[7].strip():
raise BadRequest(
_(
"username and email fields must not differ for '%s'"
% self.username
"username and email fields must not differ for '{}'".format(
self.username
)
)
)
@@ -517,4 +519,4 @@ class ExtensionUpdateHistory(models.Model):
unique_together = [("extension", "version")]
def __str__(self):
return "{}: {}".format(self.extension, self.name)
return f"{self.extension}: {self.name}"
+1 -1
View File
@@ -28,7 +28,7 @@ def get_password_hasher(scheme: str) -> type[PasswordHasher]:
"""
try:
scheme = scheme.replace("-", "")
hasher = globals()["%sHasher" % scheme.upper()]
hasher = globals()[f"{scheme.upper()}Hasher"]
except KeyError:
hasher = PLAINHasher
return hasher
+1 -1
View File
@@ -148,7 +148,7 @@ if argon2_hasher is not None:
def __init__(
self,
):
super(ARGON2IDHasher, self).__init__()
super().__init__()
parameters = dict()
+1 -1
View File
@@ -80,7 +80,7 @@ class PasswordHasher(metaclass=MetaHasher):
:return: encrypted password
"""
pwhash = self._b64encode(self._encrypt(force_str(clearvalue)))
return "%s%s" % (self.scheme, force_str(pwhash))
return f"{self.scheme}{force_str(pwhash)}"
def verify(self, clearvalue, hashed_value):
"""Verify a password against a hashed value.
+1 -1
View File
@@ -58,7 +58,7 @@ def get_dovecot_schemes():
internal_name=DOVEADM_PASS_SCHEME_ALARM,
)
return ["{{{}}}".format(smart_str(scheme)) for scheme in schemes.split()], status
return [f"{{{smart_str(scheme)}}}" for scheme in schemes.split()], status
def cache_available_password_hasher(bypass_cache=False):
+1 -1
View File
@@ -4,7 +4,7 @@ from django.core.exceptions import ValidationError
from django.utils.translation import gettext as _, ngettext
class ComplexityValidator(object):
class ComplexityValidator:
"""Check password contains at least a few things."""
def __init__(self, upper=1, lower=1, digits=1, specials=1):
+2 -2
View File
@@ -24,9 +24,9 @@ class SMSBackend:
def get_backend_class(name):
"""Return class for given backend."""
backend_module = import_module("modoboa.core.sms_backends.{}".format(name))
backend_module = import_module(f"modoboa.core.sms_backends.{name}")
try:
backend_class = getattr(backend_module, "{}Backend".format(name.upper()))
backend_class = getattr(backend_module, f"{name.upper()}Backend")
except AttributeError:
return None
else:
+1 -1
View File
@@ -148,7 +148,7 @@ class OVHBackend(SMSBackend):
"""Send a new SMS to given recipients."""
services = self.client.get("/sms")
result = self.client.post(
"/sms/{}/jobs".format(services[0]),
f"/sms/{services[0]}/jobs",
message=text,
receivers=recipients,
priority="high",
+8 -9
View File
@@ -181,7 +181,7 @@ def colorize_level(level):
}
if level not in classes:
return level
return "<p class='%s'>%s</p>" % (classes[level], level)
return f"<p class='{classes[level]}'>{level}</p>"
@register.filter
@@ -238,7 +238,7 @@ def connected_users(parser, token):
except ValueError:
raise template.TemplateSyntaxError(
"connected_users usage: {% connected_users as users %}"
)
) from None
return ConnectedUsers(varname)
@@ -279,14 +279,13 @@ def display_messages(msgs):
timeout = "undefined"
return mark_safe(
"""
f"""
<script type="text/javascript">
$(document).ready(function() {
$('body').notify('%s', '%s', %s);
});
$(document).ready(function() {{
$('body').notify('{level}', '{text}', {timeout});
}});
</script>
"""
% (level, text, timeout)
)
@@ -295,5 +294,5 @@ def currencyfmt(amount):
"""Simple temp. filter to replace babel."""
lang = get_language()
if lang == "fr":
return "{} €".format(amount)
return "€{}".format(amount)
return f"{amount} €"
return f"€{amount}"
+3 -3
View File
@@ -19,7 +19,7 @@ from modoboa.core.password_hashers.utils import get_dovecot_schemes
from modoboa.lib.tests import NO_SMTP, ModoTestCase
from .. import factories, models
DOVEADM_TEST_PATH = "{}/doveadm".format(os.path.dirname(__file__))
DOVEADM_TEST_PATH = f"{os.path.dirname(__file__)}/doveadm"
class MockedAttestedCredentialData:
@@ -38,7 +38,7 @@ class AuthenticationTestCase(ModoTestCase):
@classmethod
def setUpTestData(cls): # NOQA:N802
"""Create test data."""
super(AuthenticationTestCase, cls).setUpTestData()
super().setUpTestData()
from modoboa.admin.factories import MailboxFactory
mailbox = MailboxFactory(
@@ -322,7 +322,7 @@ class PasswordResetTestCase(ModoTestCase):
@classmethod
def setUpTestData(cls): # NOQA:N802
"""Create test data."""
super(PasswordResetTestCase, cls).setUpTestData()
super().setUpTestData()
cls.account_ok = factories.UserFactory(
username="user@test.com",
secondary_email="test@ext.com",
-1
View File
@@ -6,7 +6,6 @@ from django.test.utils import override_settings
from modoboa.core import checks
from modoboa.lib.tests import SimpleModoTestCase
from modoboa.core.password_hashers.base import CRYPTHasher
class CheckSessionCookieSecureTest(SimpleModoTestCase):
+3 -3
View File
@@ -21,7 +21,7 @@ class AuthenticationTestCase(ModoTestCase):
@classmethod
def setUpTestData(cls): # NOQA:N802
"""Create test data."""
super(AuthenticationTestCase, cls).setUpTestData()
super().setUpTestData()
cls.account = factories.UserFactory(
username="user@test.com", groups=("SimpleUsers",)
)
@@ -115,7 +115,7 @@ class ProfileTestCase(ModoTestCase):
@classmethod
def setUpTestData(cls): # NOQA:N802
"""Create test data."""
super(ProfileTestCase, cls).setUpTestData()
super().setUpTestData()
cls.account = factories.UserFactory(
username="user@test.com", groups=("SimpleUsers",)
)
@@ -231,7 +231,7 @@ class APIAccessFormTestCase(ModoTestCase):
@classmethod
def setUpTestData(cls): # NOQA:N802
"""Create test data."""
super(APIAccessFormTestCase, cls).setUpTestData()
super().setUpTestData()
cls.account = factories.UserFactory(
username="user@test.com", groups=("SimpleUsers",)
)
+3 -3
View File
@@ -13,7 +13,7 @@ from .. import factories, models
@skipIf(NO_LDAP, "No ldap module installed")
class LDAPTestCaseMixin(object):
class LDAPTestCaseMixin:
"""Set of methods used to test LDAP features."""
@cached_property
@@ -84,7 +84,7 @@ class LDAPAuthenticationTestCase(LDAPTestCaseMixin, ModoTestCase):
def setUp(self):
"""Create test data."""
super(LDAPAuthenticationTestCase, self).setUp()
super().setUp()
self.activate_ldap_authentication()
def check_created_user(self, username, group="SimpleUsers", with_mb=True):
@@ -142,7 +142,7 @@ class ProfileTestCase(LDAPTestCaseMixin, ModoTestCase):
@classmethod
def setUpTestData(cls): # NOQA:N802
"""Create test data."""
super(ProfileTestCase, cls).setUpTestData()
super().setUpTestData()
cls.account = factories.UserFactory(
username="user@test.com", groups=("SimpleUsers",)
)
+4 -4
View File
@@ -148,7 +148,7 @@ class DashboardTestCase(ModoTestCase):
@classmethod
def setUpTestData(cls): # NOQA:N802
"""Create some data."""
super(DashboardTestCase, cls).setUpTestData()
super().setUpTestData()
cls.dadmin = factories.UserFactory(
username="admin@test.com", groups=("DomainAdmins",)
)
@@ -236,7 +236,7 @@ class SettingsTestCase(ModoTestCase):
url = reverse("core:parameters")
response = self.ajax_get(url)
for app in ["core", "admin", "limits"]:
self.assertIn('data-app="{}"'.format(app), response["content"])
self.assertIn(f'data-app="{app}"', response["content"])
def test_save_settings(self):
"""Test settings save."""
@@ -317,13 +317,13 @@ class PreferencesTestCase(ModoTestCase):
@classmethod
def setUpClass(cls):
"""Register user form."""
super(PreferencesTestCase, cls).setUpClass()
super().setUpClass()
param_tools.registry.add("user", UserSettings, "Test")
@classmethod
def tearDownClass(cls):
"""Remove user class."""
super(PreferencesTestCase, cls).tearDownClass()
super().tearDownClass()
del param_tools.registry._registry["user"]["core"]
def test_get_user_index(self):
+1 -1
View File
@@ -2,4 +2,4 @@ import os
def get_doveadm_test_path():
return "{}/doveadm".format(os.path.dirname(__file__))
return f"{os.path.dirname(__file__)}/doveadm"
+1 -1
View File
@@ -88,7 +88,7 @@ def get_logs_page(request, page_id=None):
if page_id is None:
return None
return get_listing_page(
Log.objects.all().order_by("%s%s" % (sort_dir, sort_order)), page_id
Log.objects.all().order_by(f"{sort_dir}{sort_order}"), page_id
)
+8 -9
View File
@@ -141,11 +141,10 @@ class LoginView(LoginViewMixin, auth_views.LoginView):
def form_invalid(self, form):
# FIXME: should we return a 401 as before ?
self.logger.warning(
"Failed connection attempt from '%(addr)s' as user '%(user)s'"
% {
"addr": self.request.META["REMOTE_ADDR"],
"user": escape(form.cleaned_data["username"]),
}
"Failed connection attempt from '{addr}' as user '{user}'".format(
addr=self.request.META["REMOTE_ADDR"],
user=escape(form.cleaned_data["username"]),
)
)
return self.render_to_response(self.get_context_data(form=form), status=401)
@@ -220,7 +219,7 @@ class VerifySMSCodeView(generic.FormView):
try:
kwargs.update({"totp_secret": self.request.session["totp_secret"]})
except KeyError:
raise Http404
raise Http404 from None
return kwargs
def form_valid(self, form):
@@ -241,11 +240,11 @@ class ResendSMSCodeView(generic.View):
"sms_password_recovery"
)
if not sms_password_recovery:
raise Http404
raise Http404 from None
try:
user = models.User._default_manager.get(pk=self.request.session["user_pk"])
except KeyError:
raise Http404
raise Http404 from None
backend = sms_backends.get_active_backend(self.request.localconfig.parameters)
secret = cryptutils.random_hex_key(20)
code = oath.totp(secret)
@@ -255,7 +254,7 @@ class ResendSMSCodeView(generic.View):
)
)
if not backend.send(text, [user.phone_number]):
raise Http404
raise Http404 from None
self.request.session["totp_secret"] = secret
return JsonResponse({"status": "ok"})
+2 -2
View File
@@ -37,7 +37,7 @@ class DashboardView(auth_mixins.AccessMixin, generic.TemplateView):
lang = "en"
context.update({"selection": "dashboard"})
feed_url = "{}{}/weblog/feeds/".format(MODOBOA_WEBSITE_URL, lang)
feed_url = f"{MODOBOA_WEBSITE_URL}{lang}/weblog/feeds/"
if self.request.user.role != "SuperAdmins":
custom_feed_url = self.request.localconfig.parameters.get_value(
"rss_feed_url"
@@ -57,7 +57,7 @@ class DashboardView(auth_mixins.AccessMixin, generic.TemplateView):
"hide_features_widget"
)
if self.request.user.is_superuser or not hide_features_widget:
url = "{}{}/api/projects/?featured=true".format(MODOBOA_WEBSITE_URL, lang)
url = f"{MODOBOA_WEBSITE_URL}{lang}/api/projects/?featured=true"
features = []
if not settings.DISABLE_DASHBOARD_EXTERNAL_QUERIES:
try:
-2
View File
@@ -1,3 +1 @@
from django.contrib import admin
# Register your models here.

Some files were not shown because too many files have changed in this diff Show More