fix(core): Add explicit user_id and user filters to JobFilterSet

Adds the missing user_id companion filter (by ID) and an explicit
user filter (by username), mirroring the ObjectChangeFilterSet
pattern.

Fixes #22208
This commit is contained in:
Martin Hauser 2026-05-15 21:20:42 +02:00
parent a65afe7eaf
commit 521bc44c40
No known key found for this signature in database
2 changed files with 67 additions and 0 deletions

View File

@ -137,7 +137,24 @@ class JobFilterSet(BaseFilterSet):
distinct=False,
null_value=None
)
notifications = django_filters.MultipleChoiceFilter(
choices=JobNotificationChoices,
distinct=False,
null_value=None
)
queue_name = django_filters.CharFilter()
user_id = django_filters.ModelMultipleChoiceFilter(
queryset=User.objects.all(),
distinct=False,
label=_('User (ID)'),
)
user = django_filters.ModelMultipleChoiceFilter(
field_name='user__username',
queryset=User.objects.all(),
distinct=False,
to_field_name='username',
label=_('User name'),
)
class Meta:
model = Job

View File

@ -244,6 +244,56 @@ class ObjectChangeTestCase(TestCase, BaseFilterSetTests):
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 3)
class JobTestCase(TestCase, BaseFilterSetTests):
queryset = Job.objects.all()
filterset = JobFilterSet
ignore_fields = ('data', 'error', 'log_entries')
@classmethod
def setUpTestData(cls):
users = (
User(username='user1'),
User(username='user2'),
User(username='user3'),
)
User.objects.bulk_create(users)
jobs = (
Job(
name='Job 1', job_id=uuid.uuid4(), user=users[0],
notifications=JobNotificationChoices.NOTIFICATION_ALWAYS,
),
Job(
name='Job 2', job_id=uuid.uuid4(), user=users[0],
notifications=JobNotificationChoices.NOTIFICATION_ALWAYS,
),
Job(
name='Job 3', job_id=uuid.uuid4(), user=users[1],
notifications=JobNotificationChoices.NOTIFICATION_ON_FAILURE,
),
Job(
name='Job 4', job_id=uuid.uuid4(), user=users[2],
notifications=JobNotificationChoices.NOTIFICATION_NEVER,
),
)
Job.objects.bulk_create(jobs)
def test_user(self):
"""Filter Jobs by user (ID and username)."""
params = {'user_id': User.objects.filter(username__in=['user1', 'user2']).values_list('pk', flat=True)}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 3)
params = {'user': ['user1', 'user2']}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 3)
def test_notifications(self):
"""Filter Jobs by notification policy."""
params = {'notifications': [
JobNotificationChoices.NOTIFICATION_ALWAYS,
JobNotificationChoices.NOTIFICATION_ON_FAILURE,
]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 3)
class ObjectTypeTestCase(TestCase, BaseFilterSetTests):
queryset = ObjectType.objects.all()
filterset = ObjectTypeFilterSet