feat: Allauth 어드민 API들에 필터 추가

Author: MU-Software

app/admin_api/filtersets/socialaccount.py

@@ -0,0 +1,64 @@
+from allauth.account.models import EmailAddress
+from allauth.socialaccount.models import SocialAccount, SocialApp
+from core.filter.multi_field import MultiFieldOrCharInFilter
+from django_filters import rest_framework as filters
+
+
+def _social_app_provider_choices() -> list[tuple[str, str]]:
+    """SocialApp 에 등록된 provider 만 허용. callable 로 두어 매 요청마다 최신 DB 상태를 반영."""
+    return [(p, p) for p in SocialApp.objects.values_list("provider", flat=True).distinct().order_by("provider")]
+
+
+class SocialAppProviderInFilter(filters.BaseInFilter, filters.ChoiceFilter):
+    """CSV 다중값 입력 + SocialApp.provider 화이트리스트 검증 + `__in` 매칭. 각 값을 단일 choice 로 검증.
+
+    `expose_callable_choices_in_schema` 는 `core.openapi.filter_extension.DjangoFilterExtension` 가
+    스키마 생성 시 callable choices 를 한 번 호출하도록 opt-in.
+    """
+
+    expose_callable_choices_in_schema = True
+
+
+class SocialAccountAdminFilterSet(filters.FilterSet):
+    """admin 운영자 검색. provider 는 SocialApp 에 등록된 값만 허용 (`?provider=google,kakao`)."""
+
+    provider = SocialAppProviderInFilter(field_name="provider", choices=_social_app_provider_choices)
+    uid = filters.CharFilter(field_name="uid", lookup_expr="icontains")
+    user_email = filters.CharFilter(field_name="user__email", lookup_expr="icontains")
+    user_username = filters.CharFilter(field_name="user__username", lookup_expr="icontains")
+
+    date_joined_after = filters.DateTimeFilter(field_name="date_joined", lookup_expr="gte")
+    date_joined_before = filters.DateTimeFilter(field_name="date_joined", lookup_expr="lte")
+    last_login_after = filters.DateTimeFilter(field_name="last_login", lookup_expr="gte")
+    last_login_before = filters.DateTimeFilter(field_name="last_login", lookup_expr="lte")
+
+    class Meta:
+        model = SocialAccount
+        fields = [
+            "user",
+            "provider",
+            "uid",
+            "user_email",
+            "user_username",
+            "date_joined_after",
+            "date_joined_before",
+            "last_login_after",
+            "last_login_before",
+        ]
+
+
+class EmailAddressAdminFilterSet(filters.FilterSet):
+    """`email` 은 EmailAddress.email 과 User.email 양쪽을 OR 매칭. CSV 다중값 지원."""
+
+    email = MultiFieldOrCharInFilter(field_names=["email", "user__email"], lookup_expr="icontains")
+    user_username = filters.CharFilter(field_name="user__username", lookup_expr="icontains")
+
+    class Meta:
+        model = EmailAddress
+        fields = [
+            "user",
+            "email",
+            "verified",
+            "primary",
+            "user_username",
+        ]

app/admin_api/test/socialaccount_test.py

@@ -1,6 +1,7 @@
 import http
 
 import pytest
+import yaml
 from allauth.account.models import EmailAddress
 from allauth.socialaccount.models import SocialAccount, SocialApp
 from django.urls import reverse
@@ -134,6 +135,74 @@ def test_social_account_list_filter_by_user(api_client, regular_user, multi_soci
     assert {row["uid"] for row in rows} == {"alice-google-1"}
 
 
+@pytest.mark.django_db
+def test_social_account_list_filter_by_provider_csv(api_client, regular_user, multi_social_user):
+    # provider choices 는 SocialApp 등록값 기반 — alice=google, bob=google+kakao 모두 매치되려면
+    # google/kakao 두 SocialApp 모두 등록되어 있어야 함.
+    SocialApp.objects.get_or_create(provider="google", defaults={"name": "Google", "client_id": "g", "secret": "g"})
+    SocialApp.objects.get_or_create(provider="kakao", defaults={"name": "Kakao", "client_id": "k", "secret": "k"})
+
+    response = api_client.get(reverse("v1:admin-social-account-list"), {"provider": "google,kakao"})
+    assert response.status_code == http.HTTPStatus.OK
+    rows = response.json()["results"]
+    assert {row["uid"] for row in rows} == {"alice-google-1", "bob-google-1", "bob-kakao-1"}
+
+
+@pytest.mark.django_db
+def test_social_account_list_filter_by_provider_unknown_rejected(api_client, regular_user, social_app):
+    # social_app fixture 가 google 만 등록 — kakao 는 SocialApp 에 없으므로 거부.
+    response = api_client.get(reverse("v1:admin-social-account-list"), {"provider": "kakao"})
+    assert response.status_code == http.HTTPStatus.BAD_REQUEST
+
+
+@pytest.mark.django_db
+def test_social_account_provider_filter_enum_exposed_in_openapi(api_client):
+    # callable choices 를 OpenAPI 스키마에 enum 으로 노출 (core.openapi.filter_extension).
+    SocialApp.objects.get_or_create(provider="google", defaults={"name": "G", "client_id": "g", "secret": "g"})
+    SocialApp.objects.get_or_create(provider="kakao", defaults={"name": "K", "client_id": "k", "secret": "k"})
+
+    response = APIClient().get("/api/schema/v1/")
+    assert response.status_code == http.HTTPStatus.OK
+    schema = yaml.safe_load(response.content)
+
+    list_path = next(p for p in schema["paths"] if p.endswith("/allauth/social-account/"))
+    params = schema["paths"][list_path]["get"]["parameters"]
+    provider = next(p for p in params if p["name"] == "provider")
+    assert provider["schema"]["items"]["enum"] == ["google", "kakao"]
+
+
+@pytest.mark.django_db
+def test_social_account_list_filter_by_uid_icontains(api_client, regular_user, multi_social_user):
+    response = api_client.get(reverse("v1:admin-social-account-list"), {"uid": "kakao"})
+    assert response.status_code == http.HTTPStatus.OK
+    rows = response.json()["results"]
+    assert {row["uid"] for row in rows} == {"bob-kakao-1"}
+
+
+@pytest.mark.django_db
+def test_social_account_list_filter_by_user_email_and_username(api_client, regular_user, multi_social_user):
+    response = api_client.get(reverse("v1:admin-social-account-list"), {"user_email": "alice@"})
+    assert response.status_code == http.HTTPStatus.OK
+    rows = response.json()["results"]
+    assert {row["uid"] for row in rows} == {"alice-google-1"}
+
+    response = api_client.get(reverse("v1:admin-social-account-list"), {"user_username": "bob"})
+    assert response.status_code == http.HTTPStatus.OK
+    rows = response.json()["results"]
+    assert {row["uid"] for row in rows} == {"bob-google-1", "bob-kakao-1"}
+
+
+@pytest.mark.django_db
+def test_social_account_list_filter_by_date_joined_range(api_client, regular_user, multi_social_user):
+    import datetime as _dt
+
+    # 미래 시점 _after 는 결과 0 건.
+    future = (_dt.datetime.now(_dt.timezone.utc) + _dt.timedelta(days=1)).isoformat()
+    response = api_client.get(reverse("v1:admin-social-account-list"), {"date_joined_after": future})
+    assert response.status_code == http.HTTPStatus.OK
+    assert response.json()["results"] == []
+
+
 @pytest.mark.django_db
 def test_social_account_no_create_endpoint(api_client, regular_user):
     response = api_client.post(
@@ -175,6 +244,55 @@ def test_email_address_list_filter_by_user(api_client, regular_user):
     assert {row["email"] for row in rows} == {"alice@example.com"}
 
 
+@pytest.mark.django_db
+def test_email_address_list_filter_by_email_matches_emailaddress(api_client, regular_user, multi_social_user):
+    # EmailAddress.email substring 매칭.
+    response = api_client.get(reverse("v1:admin-email-address-list"), {"email": "alice@"})
+    assert response.status_code == http.HTTPStatus.OK
+    rows = response.json()["results"]
+    assert {row["email"] for row in rows} == {"alice@example.com"}
+
+
+@pytest.mark.django_db
+def test_email_address_list_filter_by_email_matches_user_email(api_client, regular_user):
+    # EA.email 은 alt 이지만 user.email 은 alice 라서 ?email=alice 로도 잡혀야 한다.
+    EmailAddress.objects.create(user=regular_user, email="alt@example.com", verified=False, primary=False)
+    response = api_client.get(reverse("v1:admin-email-address-list"), {"email": "alice@"})
+    assert response.status_code == http.HTTPStatus.OK
+    rows = response.json()["results"]
+    # alice 의 primary EA + alt EA (User.email join 으로 매칭) 모두 포함.
+    assert {row["email"] for row in rows} == {"alice@example.com", "alt@example.com"}
+
+
+@pytest.mark.django_db
+def test_email_address_list_filter_by_email_csv(api_client, regular_user, multi_social_user):
+    response = api_client.get(reverse("v1:admin-email-address-list"), {"email": "alice@,bob@"})
+    assert response.status_code == http.HTTPStatus.OK
+    rows = response.json()["results"]
+    assert {row["email"] for row in rows} == {"alice@example.com", "bob@example.com"}
+
+
+@pytest.mark.django_db
+def test_email_address_list_filter_by_verified_and_primary(api_client, regular_user):
+    EmailAddress.objects.create(user=regular_user, email="alt@example.com", verified=False, primary=False)
+    # verified=false
+    response = api_client.get(reverse("v1:admin-email-address-list"), {"verified": "false"})
+    assert response.status_code == http.HTTPStatus.OK
+    assert {row["email"] for row in response.json()["results"]} == {"alt@example.com"}
+    # primary=true
+    response = api_client.get(reverse("v1:admin-email-address-list"), {"primary": "true"})
+    assert response.status_code == http.HTTPStatus.OK
+    assert {row["email"] for row in response.json()["results"]} == {"alice@example.com"}
+
+
+@pytest.mark.django_db
+def test_email_address_list_filter_by_user_username(api_client, regular_user, multi_social_user):
+    response = api_client.get(reverse("v1:admin-email-address-list"), {"user_username": "bob"})
+    assert response.status_code == http.HTTPStatus.OK
+    rows = response.json()["results"]
+    assert {row["email"] for row in rows} == {"bob@example.com"}
+
+
 @pytest.mark.django_db
 def test_email_address_create_lowercases_email(api_client, regular_user):
     response = api_client.post(

app/admin_api/views/socialaccount.py

@@ -1,3 +1,4 @@
+from admin_api.filtersets.socialaccount import EmailAddressAdminFilterSet, SocialAccountAdminFilterSet
 from admin_api.serializers.socialaccount import (
     EmailAddressAdminSerializer,
     SocialAccountAdminSerializer,
@@ -39,7 +40,7 @@ class SocialAccountAdminViewSet(
     pagination_class = AdminPagination
     serializer_class = SocialAccountAdminSerializer
     queryset = SocialAccount.objects.all().select_related("user").order_by("-date_joined", "-id")
-    filterset_fields = ["user"]
+    filterset_class = SocialAccountAdminFilterSet
 
     def perform_destroy(self, instance: SocialAccount) -> None:
         delete_social_accounts_and_cleanup_user_emails(SocialAccount.objects.filter(pk=instance.pk))
@@ -52,4 +53,4 @@ class EmailAddressAdminViewSet(JsonSchemaViewSet, viewsets.ModelViewSet):
     permission_classes = [IsSuperUser]
     pagination_class = AdminPagination
     queryset = EmailAddress.objects.all().select_related("user").order_by("-id")
-    filterset_fields = ["user"]
+    filterset_class = EmailAddressAdminFilterSet

app/core/openapi/filter_extension.py

@@ -0,0 +1,22 @@
+from drf_spectacular.contrib.django_filters import DjangoFilterExtension as _BaseDjangoFilterExtension
+
+
+class DjangoFilterExtension(_BaseDjangoFilterExtension):
+    """기본 확장은 DB 비용 우려로 callable `choices` 를 무시한다.
+
+    Filter 클래스에 `expose_callable_choices_in_schema = True` 를 붙이면 스키마 생성 시
+    한 번만 호출해 enum 으로 노출. 런타임 dynamism 은 유지하면서 OpenAPI 문서에도
+    선택지를 드러내고 싶을 때 사용.
+    """
+
+    priority = 1
+
+    def _get_explicit_filter_choices(self, filter_field):  # type: ignore[no-untyped-def]
+        choices = filter_field.extra.get("choices")
+        if callable(choices) and getattr(filter_field, "expose_callable_choices_in_schema", False):
+            try:
+                resolved = list(choices())
+            except Exception:
+                return []
+            return [c for c, _ in resolved]
+        return super()._get_explicit_filter_choices(filter_field)

app/core/openapi/schemas.py

@@ -1,3 +1,4 @@
+from core.openapi.filter_extension import DjangoFilterExtension  # noqa: F401  drf-spectacular 확장 등록
 from drf_spectacular.openapi import AutoSchema, OpenApiExample, OpenApiResponse
 from drf_spectacular.utils import OpenApiParameter
 from rest_framework import status