Fixes build for PyPy3 (celery#6635)

* installs packages the same way docker does

* removes couchbase dependency for PyPy

* removes ephem dependency for PyPy

* fixes mongo unit tests for PyPy3

Mocking `datetime.datetime` was causing an issue with
`datetime.utcnow()`.  This mock doesn't appear to be needed.
See https://github.com/celery/celery/pull/6635/checks?check_run_id=1944166896.

* fix: Avoid shadowing `Thread` attributes

Fixes celery#6489

* ci: Install default deps for pypy3 toxenvs

* ci: Run unit tests with `tox`

* ci: Lint source in separate action using `tox`

* ci: Redent codecov action

* test: Rework some mocking in `test_platforms.py`

Also fix some flakes which may have been added by some other
autoformatter in celery#6804. The 4 space non-visual-indentation should keep
most formatters fairly happy.

* style: Fix some flakes

Co-authored-by: maybe-sybr <58414429+maybe-sybr@users.noreply.github.com>

Author: mjhoffman65

.github/workflows/python-package.yml

@@ -48,26 +48,31 @@ jobs:
           ${{ matrix.python-version }}-v1-${{ hashFiles('**/setup.py') }}
         restore-keys: |
           ${{ matrix.python-version }}-v1-
-    - name: Install dependencies
-      run: |
-        python -m pip install --upgrade pip
-        python -m pip install flake8 pytest case pytest-celery pytest-subtests pytest-timeout pytest-cov
-        python -m pip install moto boto3 msgpack PyYAML
-        if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
 
-    - name: Run Unit test with pytest
-      run: |
-        PYTHONPATH=. pytest -xv --cov=celery --cov-report=xml --cov-report term t/unit
+    - name: Install tox
+      run: python -m pip install tox
+    - name: >
+        Run tox for
+        "${{ matrix.python-version }}-unit"
+      timeout-minutes: 15
+      run: >
+        tox --verbose --verbose -e
+        "${{ matrix.python-version }}-unit"
+
+    - uses: codecov/codecov-action@v1
+      with:
+        flags: unittests # optional
+        fail_ci_if_error: true # optional (default = false)
+        verbose: true # optional (default = false)
 
+  lint:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v2
+    - uses: actions/setup-python@v2
+      # Must match the Python version in tox.ini for flake8
+      with: { python-version: 3.9 }
+    - name: Install tox
+      run: python -m pip install tox
     - name: Lint with flake8
-      run: |
-        # stop the build if there are Python syntax errors or undefined names
-        flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
-        # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide
-        flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
-    -   uses: codecov/codecov-action@v1
-        with:
-            token: ${{ secrets.CODECOV_TOKEN }} # not required for public repos
-            flags: unittests # optional
-            fail_ci_if_error: true # optional (default = false)
-            verbose: true # optional (default = false)
+      run: tox --verbose -e flake8

celery/backends/redis.py

@@ -585,6 +585,7 @@ class SentinelManagedSSLConnection(
 
 class SentinelBackend(RedisBackend):
     """Redis sentinel task result store."""
+
     # URL looks like `sentinel://0.0.0.0:26347/3;sentinel://0.0.0.0:26348/3`
     _SERVER_URI_SEPARATOR = ";"
 
@@ -598,9 +599,7 @@ def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
     def as_uri(self, include_password=False):
-        """
-        Return the server addresses as URIs, sanitizing the password or not.
-        """
+        """Return the server addresses as URIs, sanitizing the password or not."""
         # Allow superclass to do work if we don't need to force sanitization
         if include_password:
             return super().as_uri(

celery/bin/base.py

@@ -116,6 +116,7 @@ def say_chat(self, direction, title, body='', show_body=False):
 
 
 def handle_preload_options(f):
+    """Extract preload options and return a wrapped callable."""
     def caller(ctx, *args, **kwargs):
         app = ctx.obj.app
 

celery/canvas.py

@@ -1253,8 +1253,10 @@ def _freeze_group_tasks(self, _id=None, group_id=None, chord=None,
 
     def freeze(self, _id=None, group_id=None, chord=None,
                root_id=None, parent_id=None, group_index=None):
-        return self.app.GroupResult(*self._freeze_group_tasks(_id=_id, group_id=group_id,
-                                                              chord=chord, root_id=root_id, parent_id=parent_id, group_index=group_index))
+        return self.app.GroupResult(*self._freeze_group_tasks(
+            _id=_id, group_id=group_id,
+            chord=chord, root_id=root_id, parent_id=parent_id, group_index=group_index
+        ))
 
     _freeze = freeze
 

celery/concurrency/__init__.py

@@ -29,4 +29,5 @@ def get_implementation(cls):
 
 
 def get_available_pool_names():
+    """Return all available pool type names."""
     return tuple(ALIASES.keys())

celery/exceptions.py

@@ -304,6 +304,7 @@ def __repr__(self):
 
 
 class CeleryCommandException(ClickException):
+    """A general command exception which stores an exit code."""
 
     def __init__(self, message, exit_code):
         super().__init__(message=message)

celery/utils/threads.py

@@ -46,8 +46,8 @@ class bgThread(threading.Thread):
 
     def __init__(self, name=None, **kwargs):
         super().__init__()
-        self._is_shutdown = threading.Event()
-        self._is_stopped = threading.Event()
+        self.__is_shutdown = threading.Event()
+        self.__is_stopped = threading.Event()
         self.daemon = True
         self.name = name or self.__class__.__name__
 
@@ -60,7 +60,7 @@ def on_crash(self, msg, *fmt, **kwargs):
 
     def run(self):
         body = self.body
-        shutdown_set = self._is_shutdown.is_set
+        shutdown_set = self.__is_shutdown.is_set
         try:
             while not shutdown_set():
                 try:
@@ -77,16 +77,16 @@ def run(self):
 
     def _set_stopped(self):
         try:
-            self._is_stopped.set()
+            self.__is_stopped.set()
         except TypeError:  # pragma: no cover
             # we lost the race at interpreter shutdown,
             # so gc collected built-in modules.
             pass
 
     def stop(self):
         """Graceful shutdown."""
-        self._is_shutdown.set()
-        self._is_stopped.wait()
+        self.__is_shutdown.set()
+        self.__is_stopped.wait()
         if self.is_alive():
             self.join(THREAD_TIMEOUT_MAX)
 

celery/utils/timer2.py

@@ -49,8 +49,12 @@ def __init__(self, schedule=None, on_error=None, on_tick=None,
         self.on_start = on_start
         self.on_tick = on_tick or self.on_tick
         threading.Thread.__init__(self)
-        self._is_shutdown = threading.Event()
-        self._is_stopped = threading.Event()
+        # `_is_stopped` is likely to be an attribute on `Thread` objects so we
+        # double underscore these names to avoid shadowing anything and
+        # potentially getting confused by the superclass turning these into
+        # something other than an `Event` instance (e.g. a `bool`)
+        self.__is_shutdown = threading.Event()
+        self.__is_stopped = threading.Event()
         self.mutex = threading.Lock()
         self.not_empty = threading.Condition(self.mutex)
         self.daemon = True
@@ -71,7 +75,7 @@ def run(self):
             self.running = True
             self.scheduler = iter(self.schedule)
 
-            while not self._is_shutdown.isSet():
+            while not self.__is_shutdown.isSet():
                 delay = self._next_entry()
                 if delay:
                     if self.on_tick:
@@ -80,7 +84,7 @@ def run(self):
                         break
                     sleep(delay)
             try:
-                self._is_stopped.set()
+                self.__is_stopped.set()
             except TypeError:  # pragma: no cover
                 # we lost the race at interpreter shutdown,
                 # so gc collected built-in modules.
@@ -91,9 +95,9 @@ def run(self):
             os._exit(1)
 
     def stop(self):
-        self._is_shutdown.set()
+        self.__is_shutdown.set()
         if self.running:
-            self._is_stopped.wait()
+            self.__is_stopped.wait()
             self.join(THREAD_TIMEOUT_MAX)
             self.running = False
 

celery/worker/consumer/consumer.py

@@ -119,7 +119,7 @@
 These tasks cannot be acknowledged as the connection is gone, and the tasks are automatically redelivered back to the queue.
 You can enable this behavior using the worker_cancel_long_running_tasks_on_connection_loss setting.
 In Celery 5.1 it is set to False by default. The setting will be set to True by default in Celery 6.0.
-"""
+"""  # noqa: E501
 
 
 def dump_body(m, body):

requirements/extras/couchbase.txt

@@ -1 +1 @@
-couchbase>=3.0.0
+couchbase>=3.0.0; platform_python_implementation!='PyPy'