Fixup failed test during singleton

src/aiotaskq/constants.py

@@ -1,17 +1,34 @@
-"""Module to define and store all constants used across the library."""
+"""
+Module to define and store all constants used across the library.
+
+The public object from this module is `Config`. This object wraps
+all the constants, which include:
+- Variables
+- Environment variables
+- Static methods that return constant values
+"""
 
 import logging
 from os import environ
 
 from .interfaces import SerializationType
 
+# TODO @imranariffin: Wrap these inside `Config`
+# TODO @imranariffin: Rename REDIS_URL to BROKER_URL
 REDIS_URL = "redis://127.0.0.1:6379"
 TASKS_CHANNEL = "channel:tasks"
 RESULTS_CHANNEL_TEMPLATE = "channel:results:{task_id}"
 
 
 class Config:
-    """Provide configuration values."""
+    """
+    Provide configuration values.
+
+    These include:
+    - Variables
+    - Environment variables
+    - Static methods that return constant values
+    """
 
     @staticmethod
     def serialization_type() -> SerializationType:
@@ -21,6 +38,6 @@ def serialization_type() -> SerializationType:
 
     @staticmethod
     def log_level() -> int:
-        """Return the LOG_LEVEL environment variable."""
+        """Return the log level as provided via env var LOG_LEVEL."""
         level: int = int(environ.get("AIOTASKQ_LOG_LEVEL", logging.DEBUG))
         return level

src/tests/conftest.py

@@ -7,6 +7,7 @@
 import pytest
 
 from aiotaskq.interfaces import ConcurrencyType
+from aiotaskq.concurrency_manager import ConcurrencyManagerSingleton
 from aiotaskq.worker import Defaults, run_worker_forever
 
 
@@ -21,6 +22,9 @@ async def start(
         worker_rate_limit: int = Defaults.worker_rate_limit(),
         poll_interval_s: t.Optional[float] = Defaults.poll_interval_s(),
     ) -> None:
+        # Reset singleton so each test is isolated
+        ConcurrencyManagerSingleton.reset()
+
         proc = multiprocessing.Process(
             target=lambda: run_worker_forever(
                 app_import_path=app,

src/tests/test_cli.py

@@ -8,6 +8,7 @@
 
 from aiotaskq import __version__
 from aiotaskq.__main__ import _version_callback
+from aiotaskq.worker import Defaults
 
 
 def test_root_show_proper_help_message():
@@ -59,6 +60,8 @@ def test_version():
 
 def test_worker_show_proper_help_message():
     bash_command = "aiotaskq worker --help"
+    default_cpu_count: int = multiprocessing.cpu_count()
+    default_poll_interval_s: float = Defaults.poll_interval_s()
     with os.popen(bash_command) as pipe:
         output = pipe.read()
         output_expected = (
@@ -70,8 +73,8 @@ def test_worker_show_proper_help_message():
             "  APP  [required]\n"
             "\n"
             "Options:\n"
-            f"  --concurrency INTEGER           [default: {multiprocessing.cpu_count()}]\n"
-            "  --poll-interval-s FLOAT         [default: 0.01]\n"
+            f"  --concurrency INTEGER           [default: {default_cpu_count}]\n"
+            f"  --poll-interval-s FLOAT         [default: {default_poll_interval_s}]\n"
             "  --concurrency-type [multiprocessing]\n"
             "                                  [default: multiprocessing]\n"
             "  --worker-rate-limit INTEGER     [default: -1]\n"

src/tests/test_serde.py

@@ -77,4 +77,4 @@ def test_serialize_task_to_json__with_retry_param():
     }
     # And the deserialized task should function the same as the original
     task_deserialized = import_module(task_serialized_dict["func"]["module"]).some_task_2
-    task_deserialized(2, 3) == some_task_2(2, 3)
+    assert task_deserialized(2, 3) == some_task_2(2, 3)