Minor refactors, found by static analysis (#7587)

* Remove deprecated methods in `celery.local.Proxy`

* Collapse conditionals for readability

* Remove unused parameter `uuid`

* Remove unused import `ClusterOptions`

* Remove dangerous mutable default argument

Continues work from #5478

* Remove always `None` and unused global variable

* Remove unreachable `elif` block

* Consolidate import statements

* Add missing parameter to `os._exit()`

* Add missing assert statement

* Remove unused global `WindowsError`

* Use `mkstemp` instead of deprecated `mktemp`

* No need for `for..else` constructs in loops that don't break

In these cases where the loop returns or raises instead of breaking, it
is simpler to just put the code that runs after the loop completes right
after the loop instead.

* Use the previously unused parameter `compat_modules`

Previously this parameter was always overwritten by the value of
`COMPAT_MODULES.get(name, ())`, which was very likely unintentional.

* Remove unused local variable `tz`

* Make `assert_received` actually check for `is_received`

Previously, it called `is_accepted`, which was likely a copy-paste
mistake from the `assert_accepted` method.

* Use previously unused `args` and `kwargs` params

Unlike other backends' `__reduce__` methods, the one from `RedisBackend`
simply overwrites `args` and `kwargs` instead of adding to them. This
change makes it more in line with other backends.

* Update celery/backends/filesystem.py

Co-authored-by: Gabriel Soldani <1268700+gabrielsoldani@users.noreply.github.com>

Co-authored-by: Asif Saif Uddin <auvipy@gmail.com>

celery/app/log.py

@@ -64,9 +64,8 @@ def setup(self, loglevel=None, logfile=None, redirect_stdouts=False,
         handled = self.setup_logging_subsystem(
             loglevel, logfile, colorize=colorize, hostname=hostname,
         )
-        if not handled:
-            if redirect_stdouts:
-                self.redirect_stdouts(redirect_level)
+        if not handled and redirect_stdouts:
+            self.redirect_stdouts(redirect_level)
         os.environ.update(
             CELERY_LOG_LEVEL=str(loglevel) if loglevel else '',
             CELERY_LOG_FILE=str(logfile) if logfile else '',

celery/app/trace.py

@@ -369,7 +369,7 @@ def build_tracer(name, task, loader=None, hostname=None, store_errors=True,
     from celery import canvas
     signature = canvas.maybe_signature  # maybe_ does not clone if already
 
-    def on_error(request, exc, uuid, state=FAILURE, call_errbacks=True):
+    def on_error(request, exc, state=FAILURE, call_errbacks=True):
         if propagate:
             raise
         I = Info(state, exc)
@@ -459,10 +459,10 @@ def trace_task(uuid, args, kwargs, request=None):
                     traceback_clear(exc)
                 except Retry as exc:
                     I, R, state, retval = on_error(
-                        task_request, exc, uuid, RETRY, call_errbacks=False)
+                        task_request, exc, RETRY, call_errbacks=False)
                     traceback_clear(exc)
                 except Exception as exc:
-                    I, R, state, retval = on_error(task_request, exc, uuid)
+                    I, R, state, retval = on_error(task_request, exc)
                     traceback_clear(exc)
                 except BaseException:
                     raise
@@ -516,7 +516,7 @@ def trace_task(uuid, args, kwargs, request=None):
                             uuid, retval, task_request, publish_result,
                         )
                     except EncodeError as exc:
-                        I, R, state, retval = on_error(task_request, exc, uuid)
+                        I, R, state, retval = on_error(task_request, exc)
                     else:
                         Rstr = saferepr(R, resultrepr_maxsize)
                         T = monotonic() - time_start
@@ -566,7 +566,7 @@ def trace_task(uuid, args, kwargs, request=None):
                 raise
             R = report_internal_error(task, exc)
             if task_request is not None:
-                I, _, _, _ = on_error(task_request, exc, uuid)
+                I, _, _, _ = on_error(task_request, exc)
         return trace_ok_t(R, I, T, Rstr)
 
     return trace_task

celery/backends/couchbase.py

@@ -8,9 +8,9 @@
 
 try:
     from couchbase.auth import PasswordAuthenticator
-    from couchbase.cluster import Cluster, ClusterOptions
+    from couchbase.cluster import Cluster
 except ImportError:
-    Cluster = PasswordAuthenticator = ClusterOptions = None
+    Cluster = PasswordAuthenticator = None
 
 try:
     from couchbase_core._libcouchbase import FMT_AUTO

celery/backends/filesystem.py

@@ -53,10 +53,9 @@ def __init__(self, url=None, open=open, unlink=os.unlink, sep=os.sep,
         # Lets verify that we've everything setup right
         self._do_directory_test(b'.fs-backend-' + uuid().encode(encoding))
 
-    def __reduce__(self, args=(), kwargs={}):
-        kwargs.update(
-            dict(url=self.url))
-        return super().__reduce__(args, kwargs)
+    def __reduce__(self, args=(), kwargs=None):
+        kwargs = {} if not kwargs else kwargs
+        return super().__reduce__(args, {**kwargs, 'url': self.url})
 
     def _find_path(self, url):
         if not url:

celery/backends/redis.py

@@ -576,8 +576,7 @@ def client(self):
     def __reduce__(self, args=(), kwargs=None):
         kwargs = {} if not kwargs else kwargs
         return super().__reduce__(
-            (self.url,), {'expires': self.expires},
-        )
+            args, dict(kwargs, expires=self.expires, url=self.url))
 
 
 if getattr(redis, "sentinel", None):

celery/bin/amqp.py

@@ -309,4 +309,4 @@ def basic_ack(amqp_context, delivery_tag):
             amqp_context.echo_ok()
 
 
-repl = register_repl(amqp)
+register_repl(amqp)

celery/canvas.py

@@ -1469,10 +1469,9 @@ def _descend(cls, sig_obj):
                 child_size = cls._descend(child_sig)
                 if child_size > 0:
                     return child_size
-            else:
-                # We have to just hope this chain is part of some encapsulating
-                # signature which is valid and can fire the chord body
-                return 0
+            # We have to just hope this chain is part of some encapsulating
+            # signature which is valid and can fire the chord body
+            return 0
         elif isinstance(sig_obj, chord):
             # The child chord's body counts toward this chord
             return cls._descend(sig_obj.body)

celery/contrib/rdb.py

@@ -132,8 +132,7 @@ def get_avail_port(self, host, port, search_limit=100, skew=+0):
                 raise
             else:
                 return _sock, this_port
-        else:
-            raise Exception(NO_AVAILABLE_PORT.format(self=self))
+        raise Exception(NO_AVAILABLE_PORT.format(self=self))
 
     def say(self, m):
         print(m, file=self.out)

celery/contrib/testing/manager.py

@@ -153,7 +153,7 @@ def assert_accepted(self, ids, interval=0.5,
     def assert_received(self, ids, interval=0.5,
                         desc='waiting for tasks to be received', **policy):
         return self.assert_task_worker_state(
-            self.is_accepted, ids, interval=interval, desc=desc, **policy
+            self.is_received, ids, interval=interval, desc=desc, **policy
         )
 
     def assert_result_tasks_in_progress_or_completed(

celery/events/cursesmon.py

@@ -273,8 +273,6 @@ def alert_callback(mx, my, xs):
                         nexty = next(y)
                         if nexty >= my - 1:
                             subline = ' ' * 4 + '[...]'
-                        elif nexty >= my:
-                            break
                         self.win.addstr(
                             nexty, 3,
                             abbr(' ' * 4 + subline, self.screen_width - 4),

celery/local.py

@@ -148,12 +148,6 @@ def __setitem__(self, key, value):
     def __delitem__(self, key):
         del self._get_current_object()[key]
 
-    def __setslice__(self, i, j, seq):
-        self._get_current_object()[i:j] = seq
-
-    def __delslice__(self, i, j):
-        del self._get_current_object()[i:j]
-
     def __setattr__(self, name, value):
         setattr(self._get_current_object(), name, value)
 
@@ -199,9 +193,6 @@ def __iter__(self):
     def __contains__(self, i):
         return i in self._get_current_object()
 
-    def __getslice__(self, i, j):
-        return self._get_current_object()[i:j]
-
     def __add__(self, other):
         return self._get_current_object() + other
 
@@ -506,12 +497,11 @@ def create_module(name, attrs, cls_attrs=None, pkg=None,
 
 def recreate_module(name, compat_modules=None, by_module=None, direct=None,
                     base=LazyModule, **attrs):
-    compat_modules = compat_modules or ()
+    compat_modules = compat_modules or COMPAT_MODULES.get(name, ())
     by_module = by_module or {}
     direct = direct or {}
     old_module = sys.modules[name]
     origins = get_origins(by_module)
-    compat_modules = COMPAT_MODULES.get(name, ())
 
     _all = tuple(set(reduce(
         operator.add,

celery/schedules.py

@@ -539,9 +539,7 @@ def __setstate__(self, state):
         super().__init__(**state)
 
     def remaining_delta(self, last_run_at, tz=None, ffwd=ffwd):
-        # pylint: disable=redefined-outer-name
         # caching global ffwd
-        tz = tz or self.tz
         last_run_at = self.maybe_make_aware(last_run_at)
         now = self.maybe_make_aware(self.now())
         dow_num = last_run_at.isoweekday() % 7  # Sunday is day 0, not day 7

celery/utils/imports.py

@@ -1,10 +1,9 @@
 """Utilities related to importing modules and symbols by name."""
-import importlib
 import os
 import sys
 import warnings
 from contextlib import contextmanager
-from importlib import reload
+from importlib import import_module, reload
 
 try:
     from importlib.metadata import entry_points
@@ -69,7 +68,7 @@ def cwd_in_path():
 def find_module(module, path=None, imp=None):
     """Version of :func:`imp.find_module` supporting dots."""
     if imp is None:
-        imp = importlib.import_module
+        imp = import_module
     with cwd_in_path():
         try:
             return imp(module)
@@ -100,7 +99,7 @@ def import_from_cwd(module, imp=None, package=None):
     precedence over modules located in `sys.path`.
     """
     if imp is None:
-        imp = importlib.import_module
+        imp = import_module
     with cwd_in_path():
         return imp(module, package=package)
 

t/benchmarks/bench_worker.py

@@ -60,7 +60,7 @@ def it(_, n):
             n, total, n / (total + .0),
         ))
         import os
-        os._exit()
+        os._exit(0)
     it.cur += 1
 
 

t/integration/test_inspect.py

@@ -51,7 +51,7 @@ def test_registered(self, inspect):
         # TODO: We can check also the exact values of the registered methods
         ret = inspect.registered()
         assert len(ret) == 1
-        len(ret[NODENAME]) > 0
+        assert len(ret[NODENAME]) > 0
         for task_name in ret[NODENAME]:
             assert isinstance(task_name, str)
 

t/unit/app/test_log.py

@@ -2,7 +2,7 @@
 import sys
 from collections import defaultdict
 from io import StringIO
-from tempfile import mktemp
+from tempfile import mkstemp
 from unittest.mock import Mock, patch
 
 import pytest
@@ -210,7 +210,7 @@ def test_setup_logger_no_handlers_stream(self, restore_logging):
 
     @patch('os.fstat')
     def test_setup_logger_no_handlers_file(self, *args):
-        tempfile = mktemp(suffix='unittest', prefix='celery')
+        _, tempfile = mkstemp(suffix='unittest', prefix='celery')
         with patch('builtins.open') as osopen:
             with conftest.restore_logging_context_manager():
                 files = defaultdict(StringIO)

t/unit/conftest.py

@@ -30,12 +30,6 @@
     'celery_parameters'
 )
 
-try:
-    WindowsError = WindowsError
-except NameError:
-
-    class WindowsError(Exception):
-        pass
 
 PYPY3 = getattr(sys, 'pypy_version_info', None) and sys.version_info[0] > 3