Fix scheduled task purge failing without user

When a task was dispatched by a task schedule, there is no user attached
to it. Such purge tasks should just look at all tasks.

fixes #5881

CHANGES/5881.bugfix

@@ -0,0 +1 @@
+Fixed task purge to not expect a user, when a run has been scheduled by Pulp itself.

pulpcore/app/tasks/purge.py

@@ -10,7 +10,7 @@
     Task,
 )
 from pulpcore.app.role_util import get_objects_for_user
-from pulpcore.app.util import get_domain, get_current_authenticated_user
+from pulpcore.app.util import current_task, get_domain, get_current_authenticated_user
 from pulpcore.constants import TASK_STATES, TASK_FINAL_STATES
 
 log = getLogger(__name__)
@@ -61,9 +61,10 @@ def purge(finished_before=None, states=None):
     """
     This task purges from the database records of tasks which finished prior to the specified time.
 
-    It will remove only tasks that are 'owned' by the current-user (admin-users own All The Things,
-    so admins can delete all tasks). It will only delete tasks within the domain this task was
-    triggered in.
+    It will remove only tasks that are 'deletable' by the current-user (admin-users own All The
+    Things, so admins can delete all tasks). It will only delete tasks within the domain this task
+    was triggered in.
+    If scheduled, deletion is not limited to a user.
 
     It will not remove tasks that are incomplete (ie, in states running|waiting|cancelling).
 
@@ -81,14 +82,21 @@ def purge(finished_before=None, states=None):
         finished_before = timezone.now() - timezone.timedelta(minutes=settings.TASK_PROTECTION_TIME)
     if states is None:
         states = TASK_FINAL_STATES
-    current_user = get_current_authenticated_user()
     domain = get_domain()
     # Tasks, prior to the specified date, in the specified state, owned by the current-user, in the
     # current domain
-    tasks_qs = Task.objects.filter(
+    candidate_qs = Task.objects.filter(
         finished_at__lt=finished_before, state__in=states, pulp_domain=domain
     )
-    candidate_qs = get_objects_for_user(current_user, "core.delete_task", qs=tasks_qs)
+    # Has this task not been dispatched from a task schedule? Then we assume there was a user doing
+    # that.
+    if not current_task.get().taskschedule_set.exists():
+        current_user = get_current_authenticated_user()
+        assert current_user is not None, (
+            "This task should have been dispatched by a user. Cannot find it though. "
+            "Maybe it got deleted."
+        )
+        candidate_qs = get_objects_for_user(current_user, "core.delete_task", qs=candidate_qs)
     # Progress bar reporting total-units
     totals_pb = ProgressReport(
         message=_("Purged task-related-objects total"),