trustchain-oc1-orchestral
/
authentik
Archived
10
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

root: improve code style (#4436) 

* cleanup pylint comments

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* remove more

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* fix url name

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* *: use ExtractHour instead of ExtractDay

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

Signed-off-by: Jens Langhammer <jens@goauthentik.io>
This commit is contained in:
Jens L 2023-01-15 17:02:31 +01:00 committed by GitHub
parent 69d4719687
commit 9568f4dbd6
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
65 changed files with 56 additions and 164 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
authentik/admin/api/metrics.py
View File

@ -1,7 +1,7 @@
"""authentik administration metrics""" """authentik administration metrics"""
from datetime import timedelta from datetime import timedelta
from django.db.models.functions import ExtractDay from django.db.models.functions import ExtractHour
from drf_spectacular.utils import extend_schema, extend_schema_field from drf_spectacular.utils import extend_schema, extend_schema_field
from guardian.shortcuts import get_objects_for_user from guardian.shortcuts import get_objects_for_user
from rest_framework.fields import IntegerField, SerializerMethodField from rest_framework.fields import IntegerField, SerializerMethodField
@ -37,7 +37,7 @@ class LoginMetricsSerializer(PassiveSerializer):
action=EventAction.LOGIN action=EventAction.LOGIN
) )
# 3 data points per day, so 8 hour spans # 3 data points per day, so 8 hour spans
.get_events_per(timedelta(days=7), ExtractDay, 7 * 3) .get_events_per(timedelta(days=7), ExtractHour, 7 * 3)
) )
@extend_schema_field(CoordinateSerializer(many=True)) @extend_schema_field(CoordinateSerializer(many=True))
@ -49,7 +49,7 @@ class LoginMetricsSerializer(PassiveSerializer):
action=EventAction.LOGIN_FAILED action=EventAction.LOGIN_FAILED
) )
# 3 data points per day, so 8 hour spans # 3 data points per day, so 8 hour spans
.get_events_per(timedelta(days=7), ExtractDay, 7 * 3) .get_events_per(timedelta(days=7), ExtractHour, 7 * 3)
) )
@extend_schema_field(CoordinateSerializer(many=True)) @extend_schema_field(CoordinateSerializer(many=True))
@ -61,7 +61,7 @@ class LoginMetricsSerializer(PassiveSerializer):
action=EventAction.AUTHORIZE_APPLICATION action=EventAction.AUTHORIZE_APPLICATION
) )
# 3 data points per day, so 8 hour spans # 3 data points per day, so 8 hour spans
.get_events_per(timedelta(days=7), ExtractDay, 7 * 3) .get_events_per(timedelta(days=7), ExtractHour, 7 * 3)
) )

2
authentik/admin/api/tasks.py
View File

@ -79,7 +79,6 @@ class TaskViewSet(ViewSet):
), ),
], ],
) )
# pylint: disable=invalid-name
def retrieve(self, request: Request, pk=None) -> Response: def retrieve(self, request: Request, pk=None) -> Response:
"""Get a single system task""" """Get a single system task"""
task = TaskInfo.by_name(pk) task = TaskInfo.by_name(pk)
@ -110,7 +109,6 @@ class TaskViewSet(ViewSet):
], ],
) )
@action(detail=True, methods=["post"]) @action(detail=True, methods=["post"])
# pylint: disable=invalid-name
def retry(self, request: Request, pk=None) -> Response: def retry(self, request: Request, pk=None) -> Response:
"""Retry task""" """Retry task"""
task = TaskInfo.by_name(pk) task = TaskInfo.by_name(pk)

2
authentik/admin/signals.py
View File

@ -8,7 +8,6 @@ from authentik.root.monitoring import monitoring_set
@receiver(monitoring_set) @receiver(monitoring_set)
# pylint: disable=unused-argument
def monitoring_set_workers(sender, **kwargs): def monitoring_set_workers(sender, **kwargs):
"""Set worker gauge""" """Set worker gauge"""
count = len(CELERY_APP.control.ping(timeout=0.5)) count = len(CELERY_APP.control.ping(timeout=0.5))
@ -16,7 +15,6 @@ def monitoring_set_workers(sender, **kwargs):
@receiver(monitoring_set) @receiver(monitoring_set)
# pylint: disable=unused-argument
def monitoring_set_tasks(sender, **kwargs): def monitoring_set_tasks(sender, **kwargs):
"""Set task gauges""" """Set task gauges"""
for task in TaskInfo.all().values(): for task in TaskInfo.all().values():

10
authentik/blueprints/v1/common.py
View File

@ -66,7 +66,6 @@ class BlueprintEntry:
identifiers: dict[str, Any] = field(default_factory=dict) identifiers: dict[str, Any] = field(default_factory=dict)
attrs: Optional[dict[str, Any]] = field(default_factory=dict) attrs: Optional[dict[str, Any]] = field(default_factory=dict)
# pylint: disable=invalid-name
id: Optional[str] = None id: Optional[str] = None
_state: BlueprintEntryState = field(default_factory=BlueprintEntryState) _state: BlueprintEntryState = field(default_factory=BlueprintEntryState)
@ -192,7 +191,6 @@ class KeyOf(YAMLTag):
id_from: str id_from: str
# pylint: disable=unused-argument
def __init__(self, loader: "BlueprintLoader", node: ScalarNode) -> None: def __init__(self, loader: "BlueprintLoader", node: ScalarNode) -> None:
super().__init__() super().__init__()
self.id_from = node.value self.id_from = node.value
@ -219,7 +217,6 @@ class Env(YAMLTag):
key: str key: str
default: Optional[Any] default: Optional[Any]
# pylint: disable=unused-argument
def __init__(self, loader: "BlueprintLoader", node: ScalarNode | SequenceNode) -> None: def __init__(self, loader: "BlueprintLoader", node: ScalarNode | SequenceNode) -> None:
super().__init__() super().__init__()
self.default = None self.default = None
@ -239,7 +236,6 @@ class Context(YAMLTag):
key: str key: str
default: Optional[Any] default: Optional[Any]
# pylint: disable=unused-argument
def __init__(self, loader: "BlueprintLoader", node: ScalarNode | SequenceNode) -> None: def __init__(self, loader: "BlueprintLoader", node: ScalarNode | SequenceNode) -> None:
super().__init__() super().__init__()
self.default = None self.default = None
@ -262,7 +258,6 @@ class Format(YAMLTag):
format_string: str format_string: str
args: list[Any] args: list[Any]
# pylint: disable=unused-argument
def __init__(self, loader: "BlueprintLoader", node: SequenceNode) -> None: def __init__(self, loader: "BlueprintLoader", node: SequenceNode) -> None:
super().__init__() super().__init__()
self.format_string = node.value[0].value self.format_string = node.value[0].value
@ -342,7 +337,6 @@ class Condition(YAMLTag):
"XNOR": lambda args: not (reduce(ixor, args) if len(args) > 1 else args[0]), "XNOR": lambda args: not (reduce(ixor, args) if len(args) > 1 else args[0]),
} }
# pylint: disable=unused-argument
def __init__(self, loader: "BlueprintLoader", node: SequenceNode) -> None: def __init__(self, loader: "BlueprintLoader", node: SequenceNode) -> None:
super().__init__() super().__init__()
self.mode = node.value[0].value self.mode = node.value[0].value
@ -375,7 +369,6 @@ class If(YAMLTag):
when_true: Any when_true: Any
when_false: Any when_false: Any
# pylint: disable=unused-argument
def __init__(self, loader: "BlueprintLoader", node: SequenceNode) -> None: def __init__(self, loader: "BlueprintLoader", node: SequenceNode) -> None:
super().__init__() super().__init__()
self.condition = loader.construct_object(node.value[0]) self.condition = loader.construct_object(node.value[0])
@ -414,7 +407,6 @@ class Enumerate(YAMLTag, YAMLTagContext):
), ),
} }
# pylint: disable=unused-argument
def __init__(self, loader: "BlueprintLoader", node: SequenceNode) -> None: def __init__(self, loader: "BlueprintLoader", node: SequenceNode) -> None:
super().__init__() super().__init__()
self.iterable = loader.construct_object(node.value[0]) self.iterable = loader.construct_object(node.value[0])
@ -422,7 +414,6 @@ class Enumerate(YAMLTag, YAMLTagContext):
self.item_body = loader.construct_object(node.value[2]) self.item_body = loader.construct_object(node.value[2])
self.__current_context: tuple[Any, Any] = tuple() self.__current_context: tuple[Any, Any] = tuple()
# pylint: disable=unused-argument
def get_context(self, entry: BlueprintEntry, blueprint: Blueprint) -> Any: def get_context(self, entry: BlueprintEntry, blueprint: Blueprint) -> Any:
return self.__current_context return self.__current_context
@ -480,7 +471,6 @@ class EnumeratedItem(YAMLTag):
_SUPPORTED_CONTEXT_TAGS = (Enumerate,) _SUPPORTED_CONTEXT_TAGS = (Enumerate,)
# pylint: disable=unused-argument
def __init__(self, loader: "BlueprintLoader", node: ScalarNode) -> None: def __init__(self, loader: "BlueprintLoader", node: ScalarNode) -> None:
super().__init__() super().__init__()
self.depth = int(node.value) self.depth = int(node.value)

7
authentik/core/api/applications.py
View File

@ -4,7 +4,7 @@ from typing import Optional
from django.core.cache import cache from django.core.cache import cache
from django.db.models import QuerySet from django.db.models import QuerySet
from django.db.models.functions import ExtractDay from django.db.models.functions import ExtractHour
from django.http.response import HttpResponseBadRequest from django.http.response import HttpResponseBadRequest
from django.shortcuts import get_object_or_404 from django.shortcuts import get_object_or_404
from drf_spectacular.types import OpenApiTypes from drf_spectacular.types import OpenApiTypes
@ -227,7 +227,6 @@ class ApplicationViewSet(UsedByMixin, ModelViewSet):
methods=["POST"], methods=["POST"],
parser_classes=(MultiPartParser,), parser_classes=(MultiPartParser,),
) )
# pylint: disable=unused-argument
def set_icon(self, request: Request, slug: str): def set_icon(self, request: Request, slug: str):
"""Set application icon""" """Set application icon"""
app: Application = self.get_object() app: Application = self.get_object()
@ -247,7 +246,6 @@ class ApplicationViewSet(UsedByMixin, ModelViewSet):
filter_backends=[], filter_backends=[],
methods=["POST"], methods=["POST"],
) )
# pylint: disable=unused-argument
def set_icon_url(self, request: Request, slug: str): def set_icon_url(self, request: Request, slug: str):
"""Set application icon (as URL)""" """Set application icon (as URL)"""
app: Application = self.get_object() app: Application = self.get_object()
@ -256,7 +254,6 @@ class ApplicationViewSet(UsedByMixin, ModelViewSet):
@permission_required("authentik_core.view_application", ["authentik_events.view_event"]) @permission_required("authentik_core.view_application", ["authentik_events.view_event"])
@extend_schema(responses={200: CoordinateSerializer(many=True)}) @extend_schema(responses={200: CoordinateSerializer(many=True)})
@action(detail=True, pagination_class=None, filter_backends=[]) @action(detail=True, pagination_class=None, filter_backends=[])
# pylint: disable=unused-argument
def metrics(self, request: Request, slug: str): def metrics(self, request: Request, slug: str):
"""Metrics for application logins""" """Metrics for application logins"""
app = self.get_object() app = self.get_object()
@ -266,5 +263,5 @@ class ApplicationViewSet(UsedByMixin, ModelViewSet):
context__authorized_application__pk=app.pk.hex, context__authorized_application__pk=app.pk.hex,
) )
# 3 data points per day, so 8 hour spans # 3 data points per day, so 8 hour spans
.get_events_per(timedelta(days=7), ExtractDay, 7 * 3) .get_events_per(timedelta(days=7), ExtractHour, 7 * 3)
) )

3
authentik/core/api/groups.py
View File

@ -96,7 +96,6 @@ class GroupFilter(FilterSet):
queryset=User.objects.all(), queryset=User.objects.all(),
) )
# pylint: disable=unused-argument
def filter_attributes(self, queryset, name, value): def filter_attributes(self, queryset, name, value):
"""Filter attributes by query args""" """Filter attributes by query args"""
try: try:
@ -157,7 +156,6 @@ class GroupViewSet(UsedByMixin, ModelViewSet):
}, },
) )
@action(detail=True, methods=["POST"], pagination_class=None, filter_backends=[]) @action(detail=True, methods=["POST"], pagination_class=None, filter_backends=[])
# pylint: disable=unused-argument, invalid-name
def add_user(self, request: Request, pk: str) -> Response: def add_user(self, request: Request, pk: str) -> Response:
"""Add user to group""" """Add user to group"""
group: Group = self.get_object() group: Group = self.get_object()
@ -182,7 +180,6 @@ class GroupViewSet(UsedByMixin, ModelViewSet):
}, },
) )
@action(detail=True, methods=["POST"], pagination_class=None, filter_backends=[]) @action(detail=True, methods=["POST"], pagination_class=None, filter_backends=[])
# pylint: disable=unused-argument, invalid-name
def remove_user(self, request: Request, pk: str) -> Response: def remove_user(self, request: Request, pk: str) -> Response:
"""Add user to group""" """Add user to group"""
group: Group = self.get_object() group: Group = self.get_object()

1
authentik/core/api/propertymappings.py
View File

@ -117,7 +117,6 @@ class PropertyMappingViewSet(
], ],
) )
@action(detail=True, pagination_class=None, filter_backends=[], methods=["POST"]) @action(detail=True, pagination_class=None, filter_backends=[], methods=["POST"])
# pylint: disable=unused-argument, invalid-name
def test(self, request: Request, pk: str) -> Response: def test(self, request: Request, pk: str) -> Response:
"""Test Property Mapping""" """Test Property Mapping"""
mapping: PropertyMapping = self.get_object() mapping: PropertyMapping = self.get_object()

2
authentik/core/api/sources.py
View File

@ -102,7 +102,6 @@ class SourceViewSet(
methods=["POST"], methods=["POST"],
parser_classes=(MultiPartParser,), parser_classes=(MultiPartParser,),
) )
# pylint: disable=unused-argument
def set_icon(self, request: Request, slug: str): def set_icon(self, request: Request, slug: str):
"""Set source icon""" """Set source icon"""
source: Source = self.get_object() source: Source = self.get_object()
@ -122,7 +121,6 @@ class SourceViewSet(
filter_backends=[], filter_backends=[],
methods=["POST"], methods=["POST"],
) )
# pylint: disable=unused-argument
def set_icon_url(self, request: Request, slug: str): def set_icon_url(self, request: Request, slug: str):
"""Set source icon (as URL)""" """Set source icon (as URL)"""
source: Source = self.get_object() source: Source = self.get_object()

2
authentik/core/api/tokens.py
View File

@ -112,7 +112,6 @@ class TokenViewSet(UsedByMixin, ModelViewSet):
} }
) )
@action(detail=True, pagination_class=None, filter_backends=[], methods=["GET"]) @action(detail=True, pagination_class=None, filter_backends=[], methods=["GET"])
# pylint: disable=unused-argument
def view_key(self, request: Request, identifier: str) -> Response: def view_key(self, request: Request, identifier: str) -> Response:
"""Return token key and log access""" """Return token key and log access"""
token: Token = self.get_object() token: Token = self.get_object()
@ -134,7 +133,6 @@ class TokenViewSet(UsedByMixin, ModelViewSet):
}, },
) )
@action(detail=True, pagination_class=None, filter_backends=[], methods=["POST"]) @action(detail=True, pagination_class=None, filter_backends=[], methods=["POST"])
# pylint: disable=unused-argument
def set_key(self, request: Request, identifier: str) -> Response: def set_key(self, request: Request, identifier: str) -> Response:
"""Return token key and log access""" """Return token key and log access"""
token: Token = self.get_object() token: Token = self.get_object()

2
authentik/core/api/used_by.py
View File

@ -53,7 +53,7 @@ class UsedByMixin:
responses={200: UsedBySerializer(many=True)}, responses={200: UsedBySerializer(many=True)},
) )
@action(detail=True, pagination_class=None, filter_backends=[]) @action(detail=True, pagination_class=None, filter_backends=[])
# pylint: disable=invalid-name, unused-argument, too-many-locals # pylint: disable=too-many-locals
def used_by(self, request: Request, *args, **kwargs) -> Response: def used_by(self, request: Request, *args, **kwargs) -> Response:
"""Get a list of all objects that use this object""" """Get a list of all objects that use this object"""
# pyright: reportGeneralTypeIssues=false # pyright: reportGeneralTypeIssues=false

18
authentik/core/api/users.py
View File

@ -4,7 +4,7 @@ from json import loads
from typing import Any, Optional from typing import Any, Optional
from django.contrib.auth import update_session_auth_hash from django.contrib.auth import update_session_auth_hash
from django.db.models.functions import ExtractDay from django.db.models.functions import ExtractHour
from django.db.models.query import QuerySet from django.db.models.query import QuerySet
from django.db.transaction import atomic from django.db.transaction import atomic
from django.db.utils import IntegrityError from django.db.utils import IntegrityError
@ -213,7 +213,7 @@ class UserMetricsSerializer(PassiveSerializer):
action=EventAction.LOGIN, user__pk=user.pk action=EventAction.LOGIN, user__pk=user.pk
) )
# 3 data points per day, so 8 hour spans # 3 data points per day, so 8 hour spans
.get_events_per(timedelta(days=7), ExtractDay, 7 * 3) .get_events_per(timedelta(days=7), ExtractHour, 7 * 3)
) )
@extend_schema_field(CoordinateSerializer(many=True)) @extend_schema_field(CoordinateSerializer(many=True))
@ -225,7 +225,7 @@ class UserMetricsSerializer(PassiveSerializer):
action=EventAction.LOGIN_FAILED, context__username=user.username action=EventAction.LOGIN_FAILED, context__username=user.username
) )
# 3 data points per day, so 8 hour spans # 3 data points per day, so 8 hour spans
.get_events_per(timedelta(days=7), ExtractDay, 7 * 3) .get_events_per(timedelta(days=7), ExtractHour, 7 * 3)
) )
@extend_schema_field(CoordinateSerializer(many=True)) @extend_schema_field(CoordinateSerializer(many=True))
@ -237,7 +237,7 @@ class UserMetricsSerializer(PassiveSerializer):
action=EventAction.AUTHORIZE_APPLICATION, user__pk=user.pk action=EventAction.AUTHORIZE_APPLICATION, user__pk=user.pk
) )
# 3 data points per day, so 8 hour spans # 3 data points per day, so 8 hour spans
.get_events_per(timedelta(days=7), ExtractDay, 7 * 3) .get_events_per(timedelta(days=7), ExtractHour, 7 * 3)
) )
@ -269,7 +269,6 @@ class UsersFilter(FilterSet):
queryset=Group.objects.all(), queryset=Group.objects.all(),
) )
# pylint: disable=unused-argument
def filter_attributes(self, queryset, name, value): def filter_attributes(self, queryset, name, value):
"""Filter attributes by query args""" """Filter attributes by query args"""
try: try:
@ -404,9 +403,8 @@ class UserViewSet(UsedByMixin, ModelViewSet):
return Response(data={"non_field_errors": [str(exc)]}, status=400) return Response(data={"non_field_errors": [str(exc)]}, status=400)
@extend_schema(responses={200: SessionUserSerializer(many=False)}) @extend_schema(responses={200: SessionUserSerializer(many=False)})
@action(detail=False, pagination_class=None, filter_backends=[]) @action(url_path="me", url_name="me", detail=False, pagination_class=None, filter_backends=[])
# pylint: disable=invalid-name def user_me(self, request: Request) -> Response:
def me(self, request: Request) -> Response:
"""Get information about current user""" """Get information about current user"""
context = {"request": request} context = {"request": request}
serializer = SessionUserSerializer( serializer = SessionUserSerializer(
@ -434,7 +432,6 @@ class UserViewSet(UsedByMixin, ModelViewSet):
}, },
) )
@action(detail=True, methods=["POST"]) @action(detail=True, methods=["POST"])
# pylint: disable=invalid-name, unused-argument
def set_password(self, request: Request, pk: int) -> Response: def set_password(self, request: Request, pk: int) -> Response:
"""Set password for user""" """Set password for user"""
user: User = self.get_object() user: User = self.get_object()
@ -452,7 +449,6 @@ class UserViewSet(UsedByMixin, ModelViewSet):
@permission_required("authentik_core.view_user", ["authentik_events.view_event"]) @permission_required("authentik_core.view_user", ["authentik_events.view_event"])
@extend_schema(responses={200: UserMetricsSerializer(many=False)}) @extend_schema(responses={200: UserMetricsSerializer(many=False)})
@action(detail=True, pagination_class=None, filter_backends=[]) @action(detail=True, pagination_class=None, filter_backends=[])
# pylint: disable=invalid-name, unused-argument
def metrics(self, request: Request, pk: int) -> Response: def metrics(self, request: Request, pk: int) -> Response:
"""User metrics per 1h""" """User metrics per 1h"""
user: User = self.get_object() user: User = self.get_object()
@ -468,7 +464,6 @@ class UserViewSet(UsedByMixin, ModelViewSet):
}, },
) )
@action(detail=True, pagination_class=None, filter_backends=[]) @action(detail=True, pagination_class=None, filter_backends=[])
# pylint: disable=invalid-name, unused-argument
def recovery(self, request: Request, pk: int) -> Response: def recovery(self, request: Request, pk: int) -> Response:
"""Create a temporary link that a user can use to recover their accounts""" """Create a temporary link that a user can use to recover their accounts"""
link, _ = self._create_recovery_link() link, _ = self._create_recovery_link()
@ -493,7 +488,6 @@ class UserViewSet(UsedByMixin, ModelViewSet):
}, },
) )
@action(detail=True, pagination_class=None, filter_backends=[]) @action(detail=True, pagination_class=None, filter_backends=[])
# pylint: disable=invalid-name, unused-argument
def recovery_email(self, request: Request, pk: int) -> Response: def recovery_email(self, request: Request, pk: int) -> Response:
"""Create a temporary link that a user can use to recover their accounts""" """Create a temporary link that a user can use to recover their accounts"""
for_user: User = self.get_object() for_user: User = self.get_object()

2
authentik/core/management/commands/shell.py
View File

@ -49,7 +49,6 @@ class Command(BaseCommand):
return namespace return namespace
@staticmethod @staticmethod
# pylint: disable=unused-argument
def post_save_handler(sender, instance: Model, created: bool, **_): def post_save_handler(sender, instance: Model, created: bool, **_):
"""Signal handler for all object's post_save""" """Signal handler for all object's post_save"""
if not should_log_model(instance): if not should_log_model(instance):
@ -65,7 +64,6 @@ class Command(BaseCommand):
).save() ).save()
@staticmethod @staticmethod
# pylint: disable=unused-argument
def pre_delete_handler(sender, instance: Model, **_): def pre_delete_handler(sender, instance: Model, **_):
"""Signal handler for all object's pre_delete""" """Signal handler for all object's pre_delete"""
if not should_log_model(instance): # pragma: no cover if not should_log_model(instance): # pragma: no cover

3
authentik/core/signals.py
View File

@ -20,7 +20,6 @@ if TYPE_CHECKING:
@receiver(post_save) @receiver(post_save)
# pylint: disable=unused-argument
def post_save_application(sender: type[Model], instance, created: bool, **_): def post_save_application(sender: type[Model], instance, created: bool, **_):
"""Clear user's application cache upon application creation""" """Clear user's application cache upon application creation"""
from authentik.core.api.applications import user_app_cache_key from authentik.core.api.applications import user_app_cache_key
@ -36,7 +35,6 @@ def post_save_application(sender: type[Model], instance, created: bool, **_):
@receiver(user_logged_in) @receiver(user_logged_in)
# pylint: disable=unused-argument
def user_logged_in_session(sender, request: HttpRequest, user: "User", **_): def user_logged_in_session(sender, request: HttpRequest, user: "User", **_):
"""Create an AuthenticatedSession from request""" """Create an AuthenticatedSession from request"""
from authentik.core.models import AuthenticatedSession from authentik.core.models import AuthenticatedSession
@ -47,7 +45,6 @@ def user_logged_in_session(sender, request: HttpRequest, user: "User", **_):
@receiver(user_logged_out) @receiver(user_logged_out)
# pylint: disable=unused-argument
def user_logged_out_session(sender, request: HttpRequest, user: "User", **_): def user_logged_out_session(sender, request: HttpRequest, user: "User", **_):
"""Delete AuthenticatedSession if it exists""" """Delete AuthenticatedSession if it exists"""
from authentik.core.models import AuthenticatedSession from authentik.core.models import AuthenticatedSession

3
authentik/core/sources/flow_manager.py
View File

@ -48,7 +48,6 @@ class Action(Enum):
class MessageStage(StageView): class MessageStage(StageView):
"""Show a pre-configured message after the flow is done""" """Show a pre-configured message after the flow is done"""
# pylint: disable=unused-argument
def get(self, request: HttpRequest, *args, **kwargs) -> HttpResponse: def get(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
"""Show a pre-configured message after the flow is done""" """Show a pre-configured message after the flow is done"""
message = getattr(self.executor.current_stage, "message", "") message = getattr(self.executor.current_stage, "message", "")
@ -209,7 +208,6 @@ class SourceFlowManager:
response.error_message = error.messages response.error_message = error.messages
return response return response
# pylint: disable=unused-argument
def get_stages_to_append(self, flow: Flow) -> list[Stage]: def get_stages_to_append(self, flow: Flow) -> list[Stage]:
"""Hook to override stages which are appended to the flow""" """Hook to override stages which are appended to the flow"""
if not self.source.enrollment_flow: if not self.source.enrollment_flow:
@ -264,7 +262,6 @@ class SourceFlowManager:
flow_slug=flow.slug, flow_slug=flow.slug,
) )
# pylint: disable=unused-argument
def handle_auth( def handle_auth(
self, self,
connection: UserSourceConnection, connection: UserSourceConnection,

1
authentik/core/sources/stage.py
View File

@ -13,7 +13,6 @@ class PostUserEnrollmentStage(StageView):
"""Dynamically injected stage which saves the Connection after """Dynamically injected stage which saves the Connection after
the user has been enrolled.""" the user has been enrolled."""
# pylint: disable=unused-argument
def get(self, request: HttpRequest, *args, **kwargs) -> HttpResponse: def get(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
"""Stage used after the user has been enrolled""" """Stage used after the user has been enrolled"""
connection: UserSourceConnection = self.executor.plan.context[ connection: UserSourceConnection = self.executor.plan.context[

3
authentik/crypto/api.py
View File

@ -187,7 +187,6 @@ class CertificateKeyPairFilter(FilterSet):
label="Only return certificate-key pairs with keys", method="filter_has_key" label="Only return certificate-key pairs with keys", method="filter_has_key"
) )
# pylint: disable=unused-argument
def filter_has_key(self, queryset, name, value): # pragma: no cover def filter_has_key(self, queryset, name, value): # pragma: no cover
"""Only return certificate-key pairs with keys""" """Only return certificate-key pairs with keys"""
return queryset.exclude(key_data__exact="") return queryset.exclude(key_data__exact="")
@ -256,7 +255,6 @@ class CertificateKeyPairViewSet(UsedByMixin, ModelViewSet):
responses={200: CertificateDataSerializer(many=False)}, responses={200: CertificateDataSerializer(many=False)},
) )
@action(detail=True, pagination_class=None, filter_backends=[]) @action(detail=True, pagination_class=None, filter_backends=[])
# pylint: disable=invalid-name, unused-argument
def view_certificate(self, request: Request, pk: str) -> Response: def view_certificate(self, request: Request, pk: str) -> Response:
"""Return certificate-key pairs certificate and log access""" """Return certificate-key pairs certificate and log access"""
certificate: CertificateKeyPair = self.get_object() certificate: CertificateKeyPair = self.get_object()
@ -287,7 +285,6 @@ class CertificateKeyPairViewSet(UsedByMixin, ModelViewSet):
responses={200: CertificateDataSerializer(many=False)}, responses={200: CertificateDataSerializer(many=False)},
) )
@action(detail=True, pagination_class=None, filter_backends=[]) @action(detail=True, pagination_class=None, filter_backends=[])
# pylint: disable=invalid-name, unused-argument
def view_private_key(self, request: Request, pk: str) -> Response: def view_private_key(self, request: Request, pk: str) -> Response:
"""Return certificate-key pairs private key and log access""" """Return certificate-key pairs private key and log access"""
certificate: CertificateKeyPair = self.get_object() certificate: CertificateKeyPair = self.get_object()

1
authentik/events/api/events.py
View File

@ -83,7 +83,6 @@ class EventsFilter(django_filters.FilterSet):
label="Tenant name", label="Tenant name",
) )
# pylint: disable=unused-argument
def filter_context_model_pk(self, queryset, name, value): def filter_context_model_pk(self, queryset, name, value):
"""Because we store the PK as UUID.hex, """Because we store the PK as UUID.hex,
we need to remove the dashes that a client may send. We can't use a we need to remove the dashes that a client may send. We can't use a

1
authentik/events/api/notification_transports.py
View File

@ -80,7 +80,6 @@ class NotificationTransportViewSet(UsedByMixin, ModelViewSet):
request=OpenApiTypes.NONE, request=OpenApiTypes.NONE,
) )
@action(detail=True, pagination_class=None, filter_backends=[], methods=["post"]) @action(detail=True, pagination_class=None, filter_backends=[], methods=["post"])
# pylint: disable=invalid-name, unused-argument
def test(self, request: Request, pk=None) -> Response: def test(self, request: Request, pk=None) -> Response:
"""Send example notification using selected transport. Requires """Send example notification using selected transport. Requires
Modify permissions.""" Modify permissions."""

3
authentik/events/middleware.py
View File

@ -101,7 +101,6 @@ class AuditMiddleware:
self.disconnect(request) self.disconnect(request)
return response return response
# pylint: disable=unused-argument
def process_exception(self, request: HttpRequest, exception: Exception): def process_exception(self, request: HttpRequest, exception: Exception):
"""Disconnect handlers in case of exception""" """Disconnect handlers in case of exception"""
self.disconnect(request) self.disconnect(request)
@ -125,7 +124,6 @@ class AuditMiddleware:
thread.run() thread.run()
@staticmethod @staticmethod
# pylint: disable=unused-argument
def post_save_handler( def post_save_handler(
user: User, request: HttpRequest, sender, instance: Model, created: bool, **_ user: User, request: HttpRequest, sender, instance: Model, created: bool, **_
): ):
@ -137,7 +135,6 @@ class AuditMiddleware:
EventNewThread(action, request, user=user, model=model_to_dict(instance)).run() EventNewThread(action, request, user=user, model=model_to_dict(instance)).run()
@staticmethod @staticmethod
# pylint: disable=unused-argument
def pre_delete_handler(user: User, request: HttpRequest, sender, instance: Model, **_): def pre_delete_handler(user: User, request: HttpRequest, sender, instance: Model, **_):
"""Signal handler for all object's pre_delete""" """Signal handler for all object's pre_delete"""
if not should_log_model(instance): # pragma: no cover if not should_log_model(instance): # pragma: no cover

5
authentik/events/models.py
View File

@ -133,12 +133,11 @@ class EventQuerySet(QuerySet):
) )
data = Counter({int(d["age_interval"]): d["count"] for d in result}) data = Counter({int(d["age_interval"]): d["count"] for d in result})
results = [] results = []
interval_timdelta = time_since / data_points interval_delta = time_since / data_points
for interval in range(1, -data_points, -1): for interval in range(1, -data_points, -1):
results.append( results.append(
{ {
"x_cord": time.mktime((_now + (interval_timdelta * interval)).timetuple()) "x_cord": time.mktime((_now + (interval_delta * interval)).timetuple()) * 1000,
* 1000,
"y_cord": data[interval * -1], "y_cord": data[interval * -1],
} }
) )

8
authentik/events/signals.py
View File

@ -22,7 +22,6 @@ SESSION_LOGIN_EVENT = "login_event"
@receiver(user_logged_in) @receiver(user_logged_in)
# pylint: disable=unused-argument
def on_user_logged_in(sender, request: HttpRequest, user: User, **_): def on_user_logged_in(sender, request: HttpRequest, user: User, **_):
"""Log successful login""" """Log successful login"""
kwargs = {} kwargs = {}
@ -45,14 +44,12 @@ def get_login_event(request: HttpRequest) -> Optional[Event]:
@receiver(user_logged_out) @receiver(user_logged_out)
# pylint: disable=unused-argument
def on_user_logged_out(sender, request: HttpRequest, user: User, **_): def on_user_logged_out(sender, request: HttpRequest, user: User, **_):
"""Log successfully logout""" """Log successfully logout"""
Event.new(EventAction.LOGOUT).from_http(request, user=user) Event.new(EventAction.LOGOUT).from_http(request, user=user)
@receiver(user_write) @receiver(user_write)
# pylint: disable=unused-argument
def on_user_write(sender, request: HttpRequest, user: User, data: dict[str, Any], **kwargs): def on_user_write(sender, request: HttpRequest, user: User, data: dict[str, Any], **kwargs):
"""Log User write""" """Log User write"""
data["created"] = kwargs.get("created", False) data["created"] = kwargs.get("created", False)
@ -60,7 +57,6 @@ def on_user_write(sender, request: HttpRequest, user: User, data: dict[str, Any]
@receiver(login_failed) @receiver(login_failed)
# pylint: disable=unused-argument
def on_login_failed( def on_login_failed(
signal, signal,
sender, sender,
@ -74,7 +70,6 @@ def on_login_failed(
@receiver(invitation_used) @receiver(invitation_used)
# pylint: disable=unused-argument
def on_invitation_used(sender, request: HttpRequest, invitation: Invitation, **_): def on_invitation_used(sender, request: HttpRequest, invitation: Invitation, **_):
"""Log Invitation usage""" """Log Invitation usage"""
Event.new(EventAction.INVITE_USED, invitation_uuid=invitation.invite_uuid.hex).from_http( Event.new(EventAction.INVITE_USED, invitation_uuid=invitation.invite_uuid.hex).from_http(
@ -83,21 +78,18 @@ def on_invitation_used(sender, request: HttpRequest, invitation: Invitation, **_
@receiver(password_changed) @receiver(password_changed)
# pylint: disable=unused-argument
def on_password_changed(sender, user: User, password: str, **_): def on_password_changed(sender, user: User, password: str, **_):
"""Log password change""" """Log password change"""
Event.new(EventAction.PASSWORD_SET).from_http(None, user=user) Event.new(EventAction.PASSWORD_SET).from_http(None, user=user)
@receiver(post_save, sender=Event) @receiver(post_save, sender=Event)
# pylint: disable=unused-argument
def event_post_save_notification(sender, instance: Event, **_): def event_post_save_notification(sender, instance: Event, **_):
"""Start task to check if any policies trigger an notification on this event""" """Start task to check if any policies trigger an notification on this event"""
event_notification_handler.delay(instance.event_uuid.hex) event_notification_handler.delay(instance.event_uuid.hex)
@receiver(pre_delete, sender=User) @receiver(pre_delete, sender=User)
# pylint: disable=unused-argument
def event_user_pre_delete_cleanup(sender, instance: User, **_): def event_user_pre_delete_cleanup(sender, instance: User, **_):
"""If gdpr_compliance is enabled, remove all the user's events""" """If gdpr_compliance is enabled, remove all the user's events"""
gdpr_cleanup.delay(instance.pk) gdpr_cleanup.delay(instance.pk)

5
authentik/flows/api/flows.py
View File

@ -210,7 +210,6 @@ class FlowViewSet(UsedByMixin, ModelViewSet):
}, },
) )
@action(detail=True, pagination_class=None, filter_backends=[]) @action(detail=True, pagination_class=None, filter_backends=[])
# pylint: disable=unused-argument
def export(self, request: Request, slug: str) -> Response: def export(self, request: Request, slug: str) -> Response:
"""Export flow to .yaml file""" """Export flow to .yaml file"""
flow = self.get_object() flow = self.get_object()
@ -221,7 +220,6 @@ class FlowViewSet(UsedByMixin, ModelViewSet):
@extend_schema(responses={200: FlowDiagramSerializer()}) @extend_schema(responses={200: FlowDiagramSerializer()})
@action(detail=True, pagination_class=None, filter_backends=[], methods=["get"]) @action(detail=True, pagination_class=None, filter_backends=[], methods=["get"])
# pylint: disable=unused-argument
def diagram(self, request: Request, slug: str) -> Response: def diagram(self, request: Request, slug: str) -> Response:
"""Return diagram for flow with slug `slug`, in the format used by flowchart.js""" """Return diagram for flow with slug `slug`, in the format used by flowchart.js"""
diagram = FlowDiagram(self.get_object(), request.user) diagram = FlowDiagram(self.get_object(), request.user)
@ -245,7 +243,6 @@ class FlowViewSet(UsedByMixin, ModelViewSet):
methods=["POST"], methods=["POST"],
parser_classes=(MultiPartParser,), parser_classes=(MultiPartParser,),
) )
# pylint: disable=unused-argument
def set_background(self, request: Request, slug: str): def set_background(self, request: Request, slug: str):
"""Set Flow background""" """Set Flow background"""
flow: Flow = self.get_object() flow: Flow = self.get_object()
@ -265,7 +262,6 @@ class FlowViewSet(UsedByMixin, ModelViewSet):
filter_backends=[], filter_backends=[],
methods=["POST"], methods=["POST"],
) )
# pylint: disable=unused-argument
def set_background_url(self, request: Request, slug: str): def set_background_url(self, request: Request, slug: str):
"""Set Flow background (as URL)""" """Set Flow background (as URL)"""
flow: Flow = self.get_object() flow: Flow = self.get_object()
@ -278,7 +274,6 @@ class FlowViewSet(UsedByMixin, ModelViewSet):
}, },
) )
@action(detail=True, pagination_class=None, filter_backends=[]) @action(detail=True, pagination_class=None, filter_backends=[])
# pylint: disable=unused-argument
def execute(self, request: Request, slug: str): def execute(self, request: Request, slug: str):
"""Execute flow for current user""" """Execute flow for current user"""
# Because we pre-plan the flow here, and not in the planner, we need to manually clear # Because we pre-plan the flow here, and not in the planner, we need to manually clear

1
authentik/flows/markers.py
View File

@ -19,7 +19,6 @@ LOGGER = get_logger()
class StageMarker: class StageMarker:
"""Base stage marker class, no extra attributes, and has no special handler.""" """Base stage marker class, no extra attributes, and has no special handler."""
# pylint: disable=unused-argument
def process( def process(
self, self,
plan: "FlowPlan", plan: "FlowPlan",

2
authentik/flows/signals.py
View File

@ -19,7 +19,6 @@ def delete_cache_prefix(prefix: str) -> int:
@receiver(monitoring_set) @receiver(monitoring_set)
# pylint: disable=unused-argument
def monitoring_set_flows(sender, **kwargs): def monitoring_set_flows(sender, **kwargs):
"""set flow gauges""" """set flow gauges"""
GAUGE_FLOWS_CACHED.set(len(cache.keys(f"{CACHE_PREFIX}*") or [])) GAUGE_FLOWS_CACHED.set(len(cache.keys(f"{CACHE_PREFIX}*") or []))
@ -27,7 +26,6 @@ def monitoring_set_flows(sender, **kwargs):
@receiver(post_save) @receiver(post_save)
@receiver(pre_delete) @receiver(pre_delete)
# pylint: disable=unused-argument
def invalidate_flow_cache(sender, instance, **_): def invalidate_flow_cache(sender, instance, **_):
"""Invalidate flow cache when flow is updated""" """Invalidate flow cache when flow is updated"""
from authentik.flows.models import Flow, FlowStageBinding, Stage from authentik.flows.models import Flow, FlowStageBinding, Stage

1
authentik/flows/stage.py
View File

@ -91,7 +91,6 @@ class ChallengeStageView(StageView):
) )
return HttpChallengeResponse(challenge) return HttpChallengeResponse(challenge)
# pylint: disable=unused-argument
def post(self, request: Request, *args, **kwargs) -> HttpResponse: def post(self, request: Request, *args, **kwargs) -> HttpResponse:
"""Handle challenge response""" """Handle challenge response"""
challenge: ChallengeResponse = self.get_response_instance(data=request.data) challenge: ChallengeResponse = self.get_response_instance(data=request.data)

2
authentik/flows/views/executor.py
View File

@ -166,7 +166,7 @@ class FlowExecutorView(APIView):
self._logger.debug("f(exec): restored flow plan from token", plan=plan) self._logger.debug("f(exec): restored flow plan from token", plan=plan)
return plan return plan
# pylint: disable=unused-argument, too-many-return-statements # pylint: disable=too-many-return-statements
def dispatch(self, request: HttpRequest, flow_slug: str) -> HttpResponse: def dispatch(self, request: HttpRequest, flow_slug: str) -> HttpResponse:
with Hub.current.start_span( with Hub.current.start_span(
op="authentik.flow.executor.dispatch", description=self.flow.slug op="authentik.flow.executor.dispatch", description=self.flow.slug

1
authentik/flows/views/inspector.py
View File

@ -47,7 +47,6 @@ class FlowInspectorPlanSerializer(PassiveSerializer):
"""Get the plan's context, sanitized""" """Get the plan's context, sanitized"""
return sanitize_dict(plan.context) return sanitize_dict(plan.context)
# pylint: disable=unused-argument
def get_session_id(self, plan: FlowPlan) -> str: def get_session_id(self, plan: FlowPlan) -> str:
"""Get a unique session ID""" """Get a unique session ID"""
request: Request = self.context["request"] request: Request = self.context["request"]

1
authentik/lib/expression/evaluator.py
View File

@ -159,7 +159,6 @@ class BaseEvaluator:
raise exc raise exc
return result return result
# pylint: disable=unused-argument
def handle_error(self, exc: Exception, expression_source: str): # pragma: no cover def handle_error(self, exc: Exception, expression_source: str): # pragma: no cover
"""Exception Handler""" """Exception Handler"""
LOGGER.warning("Expression error", exc=exc) LOGGER.warning("Expression error", exc=exc)

1
authentik/lib/logging.py
View File

@ -3,7 +3,6 @@ from logging import Logger
from os import getpid from os import getpid
# pylint: disable=unused-argument
def add_process_id(logger: Logger, method_name: str, event_dict): def add_process_id(logger: Logger, method_name: str, event_dict):
"""Add the current process ID""" """Add the current process ID"""
event_dict["pid"] = getpid() event_dict["pid"] = getpid()

1
authentik/outposts/api/outposts.py
View File

@ -148,7 +148,6 @@ class OutpostViewSet(UsedByMixin, ModelViewSet):
@extend_schema(responses={200: OutpostHealthSerializer(many=True)}) @extend_schema(responses={200: OutpostHealthSerializer(many=True)})
@action(methods=["GET"], detail=True, pagination_class=None) @action(methods=["GET"], detail=True, pagination_class=None)
# pylint: disable=invalid-name, unused-argument
def health(self, request: Request, pk: int) -> Response: def health(self, request: Request, pk: int) -> Response:
"""Get outposts current health""" """Get outposts current health"""
outpost: Outpost = self.get_object() outpost: Outpost = self.get_object()

1
authentik/outposts/api/service_connections.py
View File

@ -91,7 +91,6 @@ class ServiceConnectionViewSet(
@extend_schema(responses={200: ServiceConnectionStateSerializer(many=False)}) @extend_schema(responses={200: ServiceConnectionStateSerializer(many=False)})
@action(detail=True, pagination_class=None, filter_backends=[]) @action(detail=True, pagination_class=None, filter_backends=[])
# pylint: disable=unused-argument, invalid-name
def state(self, request: Request, pk: str) -> Response: def state(self, request: Request, pk: str) -> Response:
"""Get the service connection's state""" """Get the service connection's state"""
connection = self.get_object() connection = self.get_object()

2
authentik/outposts/channels.py
View File

@ -69,7 +69,6 @@ class OutpostConsumer(AuthJsonConsumer):
self.outpost = outpost self.outpost = outpost
self.last_uid = self.channel_name self.last_uid = self.channel_name
# pylint: disable=unused-argument
def disconnect(self, code): def disconnect(self, code):
if self.outpost and self.last_uid: if self.outpost and self.last_uid:
state = OutpostState.for_instance_uid(self.outpost, self.last_uid) state = OutpostState.for_instance_uid(self.outpost, self.last_uid)
@ -127,7 +126,6 @@ class OutpostConsumer(AuthJsonConsumer):
response = WebsocketMessage(instruction=WebsocketMessageInstruction.ACK) response = WebsocketMessage(instruction=WebsocketMessageInstruction.ACK)
self.send_json(asdict(response)) self.send_json(asdict(response))
# pylint: disable=unused-argument
def event_update(self, event): # pragma: no cover def event_update(self, event): # pragma: no cover
"""Event handler which is called by post_save signals, Send update instruction""" """Event handler which is called by post_save signals, Send update instruction"""
self.send_json( self.send_json(

4
authentik/outposts/signals.py
View File

@ -23,7 +23,6 @@ UPDATE_TRIGGERING_MODELS = (
@receiver(pre_save, sender=Outpost) @receiver(pre_save, sender=Outpost)
# pylint: disable=unused-argument
def pre_save_outpost(sender, instance: Outpost, **_): def pre_save_outpost(sender, instance: Outpost, **_):
"""Pre-save checks for an outpost, if the name or config.kubernetes_namespace changes, """Pre-save checks for an outpost, if the name or config.kubernetes_namespace changes,
we call down and then wait for the up after save""" we call down and then wait for the up after save"""
@ -43,7 +42,6 @@ def pre_save_outpost(sender, instance: Outpost, **_):
@receiver(m2m_changed, sender=Outpost.providers.through) @receiver(m2m_changed, sender=Outpost.providers.through)
# pylint: disable=unused-argument
def m2m_changed_update(sender, instance: Model, action: str, **_): def m2m_changed_update(sender, instance: Model, action: str, **_):
"""Update outpost on m2m change, when providers are added or removed""" """Update outpost on m2m change, when providers are added or removed"""
if action in ["post_add", "post_remove", "post_clear"]: if action in ["post_add", "post_remove", "post_clear"]:
@ -51,7 +49,6 @@ def m2m_changed_update(sender, instance: Model, action: str, **_):
@receiver(post_save) @receiver(post_save)
# pylint: disable=unused-argument
def post_save_update(sender, instance: Model, created: bool, **_): def post_save_update(sender, instance: Model, created: bool, **_):
"""If an Outpost is saved, Ensure that token is created/updated """If an Outpost is saved, Ensure that token is created/updated
@ -70,7 +67,6 @@ def post_save_update(sender, instance: Model, created: bool, **_):
@receiver(pre_delete, sender=Outpost) @receiver(pre_delete, sender=Outpost)
# pylint: disable=unused-argument
def pre_delete_cleanup(sender, instance: Outpost, **_): def pre_delete_cleanup(sender, instance: Outpost, **_):
"""Ensure that Outpost's user is deleted (which will delete the token through cascade)""" """Ensure that Outpost's user is deleted (which will delete the token through cascade)"""
instance.user.delete() instance.user.delete()

1
authentik/policies/api/policies.py
View File

@ -144,7 +144,6 @@ class PolicyViewSet(
}, },
) )
@action(detail=True, pagination_class=None, filter_backends=[], methods=["POST"]) @action(detail=True, pagination_class=None, filter_backends=[], methods=["POST"])
# pylint: disable=unused-argument, invalid-name
def test(self, request: Request, pk: str) -> Response: def test(self, request: Request, pk: str) -> Response:
"""Test policy""" """Test policy"""
policy = self.get_object() policy = self.get_object()

3
authentik/policies/reputation/signals.py
View File

@ -37,7 +37,6 @@ def update_score(request: HttpRequest, identifier: str, amount: int):
@receiver(login_failed) @receiver(login_failed)
# pylint: disable=unused-argument
def handle_failed_login(sender, request, credentials, **_): def handle_failed_login(sender, request, credentials, **_):
"""Lower Score for failed login attempts""" """Lower Score for failed login attempts"""
if "username" in credentials: if "username" in credentials:
@ -45,14 +44,12 @@ def handle_failed_login(sender, request, credentials, **_):
@receiver(identification_failed) @receiver(identification_failed)
# pylint: disable=unused-argument
def handle_identification_failed(sender, request, uid_field: str, **_): def handle_identification_failed(sender, request, uid_field: str, **_):
"""Lower Score for failed identification attempts""" """Lower Score for failed identification attempts"""
update_score(request, uid_field, -1) update_score(request, uid_field, -1)
@receiver(user_logged_in) @receiver(user_logged_in)
# pylint: disable=unused-argument
def handle_successful_login(sender, request, user, **_): def handle_successful_login(sender, request, user, **_):
"""Raise score for successful attempts""" """Raise score for successful attempts"""
update_score(request, user.username, 1) update_score(request, user.username, 1)

2
authentik/policies/signals.py
View File

@ -13,14 +13,12 @@ LOGGER = get_logger()
@receiver(monitoring_set) @receiver(monitoring_set)
# pylint: disable=unused-argument
def monitoring_set_policies(sender, **kwargs): def monitoring_set_policies(sender, **kwargs):
"""set policy gauges""" """set policy gauges"""
GAUGE_POLICIES_CACHED.set(len(cache.keys(f"{CACHE_PREFIX}_*") or [])) GAUGE_POLICIES_CACHED.set(len(cache.keys(f"{CACHE_PREFIX}_*") or []))
@receiver(post_save) @receiver(post_save)
# pylint: disable=unused-argument
def invalidate_policy_cache(sender, instance, **_): def invalidate_policy_cache(sender, instance, **_):
"""Invalidate Policy cache when policy is updated""" """Invalidate Policy cache when policy is updated"""
from authentik.policies.models import Policy, PolicyBinding from authentik.policies.models import Policy, PolicyBinding

2
authentik/providers/oauth2/api/providers.py
View File

@ -83,7 +83,6 @@ class OAuth2ProviderViewSet(UsedByMixin, ModelViewSet):
} }
) )
@action(methods=["GET"], detail=True) @action(methods=["GET"], detail=True)
# pylint: disable=invalid-name
def setup_urls(self, request: Request, pk: int) -> str: def setup_urls(self, request: Request, pk: int) -> str:
"""Get Providers setup URLs""" """Get Providers setup URLs"""
provider = get_object_or_404(OAuth2Provider, pk=pk) provider = get_object_or_404(OAuth2Provider, pk=pk)
@ -140,7 +139,6 @@ class OAuth2ProviderViewSet(UsedByMixin, ModelViewSet):
}, },
) )
@action(detail=True, methods=["GET"]) @action(detail=True, methods=["GET"])
# pylint: disable=invalid-name, unused-argument
def preview_user(self, request: Request, pk: int) -> Response: def preview_user(self, request: Request, pk: int) -> Response:
"""Preview user data for provider""" """Preview user data for provider"""
provider: OAuth2Provider = self.get_object() provider: OAuth2Provider = self.get_object()

2
authentik/providers/oauth2/views/authorize.py
View File

@ -318,7 +318,6 @@ class AuthorizationFlowInitView(PolicyAccessView):
request.context["oauth_response_type"] = self.params.response_type request.context["oauth_response_type"] = self.params.response_type
return request return request
# pylint: disable=unused-argument
def get(self, request: HttpRequest, *args, **kwargs) -> HttpResponse: def get(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
"""Start FlowPLanner, return to flow executor shell""" """Start FlowPLanner, return to flow executor shell"""
# After we've checked permissions, and the user has access, check if we need # After we've checked permissions, and the user has access, check if we need
@ -429,7 +428,6 @@ class OAuthFulfillmentStage(StageView):
"""Wrapper when this stage gets hit with a post request""" """Wrapper when this stage gets hit with a post request"""
return self.get(request, *args, **kwargs) return self.get(request, *args, **kwargs)
# pylint: disable=unused-argument
def get(self, request: HttpRequest, *args, **kwargs) -> HttpResponse: def get(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
"""final Stage of an OAuth2 Flow""" """final Stage of an OAuth2 Flow"""
if PLAN_CONTEXT_PARAMS not in self.executor.plan.context: if PLAN_CONTEXT_PARAMS not in self.executor.plan.context:

1
authentik/providers/oauth2/views/github.py
View File

@ -73,7 +73,6 @@ class GitHubUserView(View):
class GitHubUserTeamsView(View): class GitHubUserTeamsView(View):
"""Emulate GitHub's /user/teams API Endpoint""" """Emulate GitHub's /user/teams API Endpoint"""
# pylint: disable=unused-argument
def get(self, request: HttpRequest, token: RefreshToken) -> HttpResponse: def get(self, request: HttpRequest, token: RefreshToken) -> HttpResponse:
"""Emulate GitHub's /user/teams API Endpoint""" """Emulate GitHub's /user/teams API Endpoint"""
user = token.user user = token.user

1
authentik/providers/oauth2/views/provider.py
View File

@ -144,7 +144,6 @@ class ProviderInfoView(View):
default_claims.extend(value.keys()) default_claims.extend(value.keys())
return default_claims return default_claims
# pylint: disable=unused-argument
def get(self, request: HttpRequest, *args, **kwargs) -> HttpResponse: def get(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
"""OpenID-compliant Provider Info""" """OpenID-compliant Provider Info"""
return JsonResponse(self.get_info(self.provider), json_dumps_params={"indent": 2}) return JsonResponse(self.get_info(self.provider), json_dumps_params={"indent": 2})

2
authentik/providers/saml/api/providers.py
View File

@ -209,7 +209,6 @@ class SAMLProviderViewSet(UsedByMixin, ModelViewSet):
], ],
) )
@action(methods=["GET"], detail=True, permission_classes=[AllowAny]) @action(methods=["GET"], detail=True, permission_classes=[AllowAny])
# pylint: disable=invalid-name, unused-argument
def metadata(self, request: Request, pk: int) -> Response: def metadata(self, request: Request, pk: int) -> Response:
"""Return metadata as XML string""" """Return metadata as XML string"""
# We don't use self.get_object() on purpose as this view is un-authenticated # We don't use self.get_object() on purpose as this view is un-authenticated
@ -282,7 +281,6 @@ class SAMLProviderViewSet(UsedByMixin, ModelViewSet):
}, },
) )
@action(detail=True, methods=["GET"]) @action(detail=True, methods=["GET"])
# pylint: disable=invalid-name, unused-argument
def preview_user(self, request: Request, pk: int) -> Response: def preview_user(self, request: Request, pk: int) -> Response:
"""Preview user data for provider""" """Preview user data for provider"""
provider: SAMLProvider = self.get_object() provider: SAMLProvider = self.get_object()

1
authentik/providers/saml/processors/authn_request_parser.py
View File

@ -35,7 +35,6 @@ ERROR_FAILED_TO_VERIFY = "Failed to verify signature"
class AuthNRequest: class AuthNRequest:
"""AuthNRequest Dataclass""" """AuthNRequest Dataclass"""
# pylint: disable=invalid-name
id: Optional[str] = None id: Optional[str] = None
relay_state: Optional[str] = None relay_state: Optional[str] = None

1
authentik/providers/saml/processors/logout_request_parser.py
View File

@ -16,7 +16,6 @@ from authentik.sources.saml.processors.constants import NS_SAML_PROTOCOL
class LogoutRequest: class LogoutRequest:
"""Logout Request""" """Logout Request"""
# pylint: disable=invalid-name
id: Optional[str] = None id: Optional[str] = None
issuer: Optional[str] = None issuer: Optional[str] = None

1
authentik/providers/saml/views/slo.py
View File

@ -39,7 +39,6 @@ class SAMLSLOView(PolicyAccessView):
"""Handler to verify the SAML Request. Must be implemented by a subclass""" """Handler to verify the SAML Request. Must be implemented by a subclass"""
raise NotImplementedError raise NotImplementedError
# pylint: disable=unused-argument
def get(self, request: HttpRequest, application_slug: str) -> HttpResponse: def get(self, request: HttpRequest, application_slug: str) -> HttpResponse:
"""Verify the SAML Request, and if valid initiate the FlowPlanner for the application""" """Verify the SAML Request, and if valid initiate the FlowPlanner for the application"""
# Call the method handler, which checks the SAML # Call the method handler, which checks the SAML

1
authentik/providers/saml/views/sso.py
View File

@ -50,7 +50,6 @@ class SAMLSSOView(PolicyAccessView):
"""Handler to verify the SAML Request. Must be implemented by a subclass""" """Handler to verify the SAML Request. Must be implemented by a subclass"""
raise NotImplementedError raise NotImplementedError
# pylint: disable=unused-argument
def get(self, request: HttpRequest, application_slug: str) -> HttpResponse: def get(self, request: HttpRequest, application_slug: str) -> HttpResponse:
"""Verify the SAML Request, and if valid initiate the FlowPlanner for the application""" """Verify the SAML Request, and if valid initiate the FlowPlanner for the application"""
# Call the method handler, which checks the SAML # Call the method handler, which checks the SAML

5
authentik/root/celery.py
View File

@ -30,14 +30,12 @@ CELERY_APP = Celery("authentik")
CTX_TASK_ID = ContextVar(STRUCTLOG_KEY_PREFIX + "task_id", default=Ellipsis) CTX_TASK_ID = ContextVar(STRUCTLOG_KEY_PREFIX + "task_id", default=Ellipsis)
# pylint: disable=unused-argument
@setup_logging.connect @setup_logging.connect
def config_loggers(*args, **kwargs): def config_loggers(*args, **kwargs):
"""Apply logging settings from settings.py to celery""" """Apply logging settings from settings.py to celery"""
dictConfig(settings.LOGGING) dictConfig(settings.LOGGING)
# pylint: disable=unused-argument
@after_task_publish.connect @after_task_publish.connect
def after_task_publish_hook(sender=None, headers=None, body=None, **kwargs): def after_task_publish_hook(sender=None, headers=None, body=None, **kwargs):
"""Log task_id after it was published""" """Log task_id after it was published"""
@ -45,7 +43,6 @@ def after_task_publish_hook(sender=None, headers=None, body=None, **kwargs):
LOGGER.info("Task published", task_id=info.get("id", ""), task_name=info.get("task", "")) LOGGER.info("Task published", task_id=info.get("id", ""), task_name=info.get("task", ""))
# pylint: disable=unused-argument
@task_prerun.connect @task_prerun.connect
def task_prerun_hook(task_id: str, task, *args, **kwargs): def task_prerun_hook(task_id: str, task, *args, **kwargs):
"""Log task_id on worker""" """Log task_id on worker"""
@ -54,7 +51,6 @@ def task_prerun_hook(task_id: str, task, *args, **kwargs):
LOGGER.info("Task started", task_id=task_id, task_name=task.__name__) LOGGER.info("Task started", task_id=task_id, task_name=task.__name__)
# pylint: disable=unused-argument
@task_postrun.connect @task_postrun.connect
def task_postrun_hook(task_id, task, *args, retval=None, state=None, **kwargs): def task_postrun_hook(task_id, task, *args, retval=None, state=None, **kwargs):
"""Log task_id on worker""" """Log task_id on worker"""
@ -62,7 +58,6 @@ def task_postrun_hook(task_id, task, *args, retval=None, state=None, **kwargs):
LOGGER.info("Task finished", task_id=task_id, task_name=task.__name__, state=state) LOGGER.info("Task finished", task_id=task_id, task_name=task.__name__, state=state)
# pylint: disable=unused-argument
@task_failure.connect @task_failure.connect
@task_internal_error.connect @task_internal_error.connect
def task_error_hook(task_id, exception: Exception, traceback, *args, **kwargs): def task_error_hook(task_id, exception: Exception, traceback, *args, **kwargs):

1
authentik/root/messages/consumer.py
View File

@ -18,7 +18,6 @@ class MessageConsumer(JsonWebsocketConsumer):
return return
cache.set(f"{CACHE_PREFIX}{self.session_key}_messages_{self.channel_name}", True, None) cache.set(f"{CACHE_PREFIX}{self.session_key}_messages_{self.channel_name}", True, None)
# pylint: disable=unused-argument
def disconnect(self, code): def disconnect(self, code):
cache.delete(f"{CACHE_PREFIX}{self.session_key}_messages_{self.channel_name}") cache.delete(f"{CACHE_PREFIX}{self.session_key}_messages_{self.channel_name}")

1
authentik/sources/ldap/api.py
View File

@ -100,7 +100,6 @@ class LDAPSourceViewSet(UsedByMixin, ModelViewSet):
} }
) )
@action(methods=["GET"], detail=True, pagination_class=None, filter_backends=[]) @action(methods=["GET"], detail=True, pagination_class=None, filter_backends=[])
# pylint: disable=unused-argument
def sync_status(self, request: Request, slug: str) -> Response: def sync_status(self, request: Request, slug: str) -> Response:
"""Get source's sync status""" """Get source's sync status"""
source = self.get_object() source = self.get_object()

3
authentik/sources/ldap/signals.py
View File

@ -22,7 +22,6 @@ from authentik.stages.prompt.signals import password_validate
@receiver(post_save, sender=LDAPSource) @receiver(post_save, sender=LDAPSource)
# pylint: disable=unused-argument
def sync_ldap_source_on_save(sender, instance: LDAPSource, **_): def sync_ldap_source_on_save(sender, instance: LDAPSource, **_):
"""Ensure that source is synced on save (if enabled)""" """Ensure that source is synced on save (if enabled)"""
if not instance.enabled: if not instance.enabled:
@ -42,7 +41,6 @@ def sync_ldap_source_on_save(sender, instance: LDAPSource, **_):
@receiver(password_validate) @receiver(password_validate)
# pylint: disable=unused-argument
def ldap_password_validate(sender, password: str, plan_context: dict[str, Any], **__): def ldap_password_validate(sender, password: str, plan_context: dict[str, Any], **__):
"""if there's an LDAP Source with enabled password sync, check the password""" """if there's an LDAP Source with enabled password sync, check the password"""
sources = LDAPSource.objects.filter(sync_users_password=True) sources = LDAPSource.objects.filter(sync_users_password=True)
@ -59,7 +57,6 @@ def ldap_password_validate(sender, password: str, plan_context: dict[str, Any],
@receiver(password_changed) @receiver(password_changed)
# pylint: disable=unused-argument
def ldap_sync_password(sender, user: User, password: str, **_): def ldap_sync_password(sender, user: User, password: str, **_):
"""Connect to ldap and update password.""" """Connect to ldap and update password."""
sources = LDAPSource.objects.filter(sync_users_password=True) sources = LDAPSource.objects.filter(sync_users_password=True)

1
authentik/sources/oauth/api/source.py
View File

@ -118,7 +118,6 @@ class OAuthSourceFilter(FilterSet):
has_jwks = BooleanFilter(label="Only return sources with JWKS data", method="filter_has_jwks") has_jwks = BooleanFilter(label="Only return sources with JWKS data", method="filter_has_jwks")
# pylint: disable=unused-argument
def filter_has_jwks(self, queryset, name, value): # pragma: no cover def filter_has_jwks(self, queryset, name, value): # pragma: no cover
"""Only return sources with JWKS data""" """Only return sources with JWKS data"""
return queryset.exclude(oidc_jwks__iexact="{}") return queryset.exclude(oidc_jwks__iexact="{}")

1
authentik/sources/oauth/types/registry.py
View File

@ -41,7 +41,6 @@ class SourceType:
"""Get Icon URL for login""" """Get Icon URL for login"""
return static(f"authentik/sources/{self.slug}.svg") return static(f"authentik/sources/{self.slug}.svg")
# pylint: disable=unused-argument
def login_challenge(self, source: OAuthSource, request: HttpRequest) -> Challenge: def login_challenge(self, source: OAuthSource, request: HttpRequest) -> Challenge:
"""Allow types to return custom challenges""" """Allow types to return custom challenges"""
return RedirectChallenge( return RedirectChallenge(

3
authentik/sources/oauth/views/callback.py
View File

@ -70,12 +70,10 @@ class OAuthCallback(OAuthClientMixin, View):
access_token=self.token.get("access_token"), access_token=self.token.get("access_token"),
) )
# pylint: disable=unused-argument
def get_callback_url(self, source: OAuthSource) -> str: def get_callback_url(self, source: OAuthSource) -> str:
"Return callback url if different than the current url." "Return callback url if different than the current url."
return "" return ""
# pylint: disable=unused-argument
def get_error_redirect(self, source: OAuthSource, reason: str) -> str: def get_error_redirect(self, source: OAuthSource, reason: str) -> str:
"Return url to redirect on login failure." "Return url to redirect on login failure."
return settings.LOGIN_URL return settings.LOGIN_URL
@ -87,7 +85,6 @@ class OAuthCallback(OAuthClientMixin, View):
"""Create a dict of User data""" """Create a dict of User data"""
raise NotImplementedError() raise NotImplementedError()
# pylint: disable=unused-argument
def get_user_id(self, info: dict[str, Any]) -> Optional[str]: def get_user_id(self, info: dict[str, Any]) -> Optional[str]:
"""Return unique identifier from the profile info.""" """Return unique identifier from the profile info."""
if "id" in info: if "id" in info:

1
authentik/sources/oauth/views/redirect.py
View File

@ -18,7 +18,6 @@ class OAuthRedirect(OAuthClientMixin, RedirectView):
permanent = False permanent = False
params = None params = None
# pylint: disable=unused-argument
def get_additional_parameters(self, source: OAuthSource) -> dict[str, Any]: def get_additional_parameters(self, source: OAuthSource) -> dict[str, Any]:
"Return additional redirect parameters for this source." "Return additional redirect parameters for this source."
return self.params or {} return self.params or {}

1
authentik/sources/saml/api/source.py
View File

@ -66,7 +66,6 @@ class SAMLSourceViewSet(UsedByMixin, ModelViewSet):
@extend_schema(responses={200: SAMLMetadataSerializer(many=False)}) @extend_schema(responses={200: SAMLMetadataSerializer(many=False)})
@action(methods=["GET"], detail=True) @action(methods=["GET"], detail=True)
# pylint: disable=unused-argument
def metadata(self, request: Request, slug: str) -> Response: def metadata(self, request: Request, slug: str) -> Response:
"""Return metadata as XML string""" """Return metadata as XML string"""
source = self.get_object() source = self.get_object()

1
authentik/sources/saml/signals.py
View File

@ -10,7 +10,6 @@ LOGGER = get_logger()
@receiver(user_logged_out) @receiver(user_logged_out)
# pylint: disable=unused-argument
def on_user_logged_out(sender, request: HttpRequest, user: User, **_): def on_user_logged_out(sender, request: HttpRequest, user: User, **_):
"""Delete temporary user if the `delete_on_logout` flag is enabled""" """Delete temporary user if the `delete_on_logout` flag is enabled"""
if not user: if not user:

3
authentik/stages/authenticator_duo/api.py
View File

@ -78,7 +78,6 @@ class AuthenticatorDuoStageViewSet(UsedByMixin, ModelViewSet):
}, },
) )
@action(methods=["POST"], detail=True, permission_classes=[]) @action(methods=["POST"], detail=True, permission_classes=[])
# pylint: disable=invalid-name,unused-argument
def enrollment_status(self, request: Request, pk: str) -> Response: def enrollment_status(self, request: Request, pk: str) -> Response:
"""Check enrollment status of user details in current session""" """Check enrollment status of user details in current session"""
stage: AuthenticatorDuoStage = AuthenticatorDuoStage.objects.filter(pk=pk).first() stage: AuthenticatorDuoStage = AuthenticatorDuoStage.objects.filter(pk=pk).first()
@ -108,7 +107,6 @@ class AuthenticatorDuoStageViewSet(UsedByMixin, ModelViewSet):
}, },
) )
@action(methods=["POST"], detail=True) @action(methods=["POST"], detail=True)
# pylint: disable=invalid-name,unused-argument
def import_device_manual(self, request: Request, pk: str) -> Response: def import_device_manual(self, request: Request, pk: str) -> Response:
"""Import duo devices into authentik""" """Import duo devices into authentik"""
stage: AuthenticatorDuoStage = self.get_object() stage: AuthenticatorDuoStage = self.get_object()
@ -150,7 +148,6 @@ class AuthenticatorDuoStageViewSet(UsedByMixin, ModelViewSet):
}, },
) )
@action(methods=["POST"], detail=True) @action(methods=["POST"], detail=True)
# pylint: disable=invalid-name,unused-argument
def import_devices_automatic(self, request: Request, pk: str) -> Response: def import_devices_automatic(self, request: Request, pk: str) -> Response:
"""Import duo devices into authentik""" """Import duo devices into authentik"""
stage: AuthenticatorDuoStage = self.get_object() stage: AuthenticatorDuoStage = self.get_object()

1
authentik/stages/authenticator_static/signals.py
View File

@ -7,7 +7,6 @@ from authentik.events.models import Event
@receiver(pre_delete, sender=StaticDevice) @receiver(pre_delete, sender=StaticDevice)
# pylint: disable=unused-argument
def pre_delete_event(sender, instance: StaticDevice, **_): def pre_delete_event(sender, instance: StaticDevice, **_):
"""Create event before deleting Static Devices""" """Create event before deleting Static Devices"""
# Create event with email notification # Create event with email notification

1
authentik/stages/authenticator_validate/challenge.py
View File

@ -125,7 +125,6 @@ def validate_challenge_code(code: str, stage_view: StageView, user: User) -> Dev
return device return device
# pylint: disable=unused-argument
def validate_challenge_webauthn(data: dict, stage_view: StageView, user: User) -> Device: def validate_challenge_webauthn(data: dict, stage_view: StageView, user: User) -> Device:
"""Validate WebAuthn Challenge""" """Validate WebAuthn Challenge"""
request = stage_view.request request = stage_view.request

1
authentik/stages/authenticator_validate/stage.py
View File

@ -374,7 +374,6 @@ class AuthenticatorValidateStageView(ChallengeStageView):
) )
return response return response
# pylint: disable=unused-argument
def challenge_valid(self, response: AuthenticatorValidationChallengeResponse) -> HttpResponse: def challenge_valid(self, response: AuthenticatorValidationChallengeResponse) -> HttpResponse:
# All validation is done by the serializer # All validation is done by the serializer
user = self.executor.plan.context.get(PLAN_CONTEXT_PENDING_USER) user = self.executor.plan.context.get(PLAN_CONTEXT_PENDING_USER)

1
authentik/stages/prompt/stage.py
View File

@ -134,7 +134,6 @@ class PromptChallengeResponse(ChallengeResponse):
def username_field_validator_factory() -> Callable[[PromptChallenge, str], Any]: def username_field_validator_factory() -> Callable[[PromptChallenge, str], Any]:
"""Return a `clean_` method for `field`. Clean method checks if username is taken already.""" """Return a `clean_` method for `field`. Clean method checks if username is taken already."""
# pylint: disable=unused-argument
def username_field_validator(self: PromptChallenge, value: str) -> Any: def username_field_validator(self: PromptChallenge, value: str) -> Any:
"""Check for duplicate usernames""" """Check for duplicate usernames"""
if User.objects.filter(username=value).exists(): if User.objects.filter(username=value).exists():

1
authentik/tenants/api.py
View File

@ -117,7 +117,6 @@ class TenantViewSet(UsedByMixin, ModelViewSet):
responses=CurrentTenantSerializer(many=False), responses=CurrentTenantSerializer(many=False),
) )
@action(methods=["GET"], detail=False, permission_classes=[AllowAny]) @action(methods=["GET"], detail=False, permission_classes=[AllowAny])
# pylint: disable=invalid-name, unused-argument
def current(self, request: Request) -> Response: def current(self, request: Request) -> Response:
"""Get current tenant""" """Get current tenant"""
tenant: Tenant = request._request.tenant tenant: Tenant = request._request.tenant

3
lifecycle/gunicorn.conf.py
View File

@ -82,7 +82,7 @@ else:
workers = int(CONFIG.y("web.workers", default_workers)) workers = int(CONFIG.y("web.workers", default_workers))
threads = int(CONFIG.y("web.threads", 4)) threads = int(CONFIG.y("web.threads", 4))
# pylint: disable=unused-argument
def post_fork(server: "Arbiter", worker: DjangoUvicornWorker): def post_fork(server: "Arbiter", worker: DjangoUvicornWorker):
"""Tell prometheus to use worker number instead of process ID for multiprocess""" """Tell prometheus to use worker number instead of process ID for multiprocess"""
from prometheus_client import values from prometheus_client import values
@ -90,7 +90,6 @@ def post_fork(server: "Arbiter", worker: DjangoUvicornWorker):
values.ValueClass = MultiProcessValue(lambda: worker._worker_id) values.ValueClass = MultiProcessValue(lambda: worker._worker_id)
# pylint: disable=unused-argument
def worker_exit(server: "Arbiter", worker: DjangoUvicornWorker): def worker_exit(server: "Arbiter", worker: DjangoUvicornWorker):
"""Remove pid dbs when worker is shutdown""" """Remove pid dbs when worker is shutdown"""
from prometheus_client import multiprocess from prometheus_client import multiprocess

12
pyproject.toml
View File

@ -62,10 +62,19 @@ exclude_lines = [
] ]
show_missing = true show_missing = true
[tool.pylint.basic]
good-names = [
"pk",
"id",
"i",
"j",
"k",
"_",
]
[tool.pylint.master] [tool.pylint.master]
disable = [ disable = [
"arguments-differ", "arguments-differ",
"fixme",
"locally-disabled", "locally-disabled",
"too-many-ancestors", "too-many-ancestors",
"too-few-public-methods", "too-few-public-methods",
@ -74,6 +83,7 @@ disable = [
"similarities", "similarities",
"cyclic-import", "cyclic-import",
"protected-access", "protected-access",
"unused-argument",
"raise-missing-from", "raise-missing-from",
# To preserve django's translation function we need to use %-formatting # To preserve django's translation function we need to use %-formatting
"consider-using-f-string", "consider-using-f-string",

26
tests/integration/test_outpost_docker.py
View File

@ -65,18 +65,20 @@ class OutpostDockerTests(ChannelsLiveServerTestCase):
external_host="http://localhost", external_host="http://localhost",
authorization_flow=create_test_flow(), authorization_flow=create_test_flow(),
) )
authentication_kp = CertificateKeyPair.objects.create( with (
name="docker-authentication", open(f"{self.ssl_folder}/client/cert.pem", encoding="utf8") as cert,
# pylint: disable=consider-using-with open(f"{self.ssl_folder}/client/key.pem", encoding="utf8") as key,
certificate_data=open(f"{self.ssl_folder}/client/cert.pem", encoding="utf8").read(), ):
# pylint: disable=consider-using-with authentication_kp = CertificateKeyPair.objects.create(
key_data=open(f"{self.ssl_folder}/client/key.pem", encoding="utf8").read(), name="docker-authentication",
) certificate_data=cert.read(),
verification_kp = CertificateKeyPair.objects.create( key_data=key.read(),
name="docker-verification", )
# pylint: disable=consider-using-with with open(f"{self.ssl_folder}/client/ca.pem", encoding="utf8") as authority:
certificate_data=open(f"{self.ssl_folder}/client/ca.pem", encoding="utf8").read(), verification_kp = CertificateKeyPair.objects.create(
) name="docker-verification",
certificate_data=authority.read(),
)
self.service_connection = DockerServiceConnection.objects.create( self.service_connection = DockerServiceConnection.objects.create(
url="https://localhost:2376", url="https://localhost:2376",
tls_verification=verification_kp, tls_verification=verification_kp,

26
tests/integration/test_proxy_docker.py
View File

@ -65,18 +65,20 @@ class TestProxyDocker(ChannelsLiveServerTestCase):
external_host="http://localhost", external_host="http://localhost",
authorization_flow=create_test_flow(), authorization_flow=create_test_flow(),
) )
authentication_kp = CertificateKeyPair.objects.create( with (
name="docker-authentication", open(f"{self.ssl_folder}/client/cert.pem", encoding="utf8") as cert,
# pylint: disable=consider-using-with open(f"{self.ssl_folder}/client/key.pem", encoding="utf8") as key,
certificate_data=open(f"{self.ssl_folder}/client/cert.pem", encoding="utf8").read(), ):
# pylint: disable=consider-using-with authentication_kp = CertificateKeyPair.objects.create(
key_data=open(f"{self.ssl_folder}/client/key.pem", encoding="utf8").read(), name="docker-authentication",
) certificate_data=cert.read(),
verification_kp = CertificateKeyPair.objects.create( key_data=key.read(),
name="docker-verification", )
# pylint: disable=consider-using-with with open(f"{self.ssl_folder}/client/ca.pem", encoding="utf8") as authority:
certificate_data=open(f"{self.ssl_folder}/client/ca.pem", encoding="utf8").read(), verification_kp = CertificateKeyPair.objects.create(
) name="docker-verification",
certificate_data=authority.read(),
)
self.service_connection = DockerServiceConnection.objects.create( self.service_connection = DockerServiceConnection.objects.create(
url="https://localhost:2376", url="https://localhost:2376",
tls_verification=verification_kp, tls_verification=verification_kp,