test:workers cleanup

Author: cunla

scheduler/redis_models/base.py

@@ -24,7 +24,7 @@ def as_str(v: Union[bytes, str]) -> Optional[str]:
         return v
     if isinstance(v, bytes):
         return v.decode("utf-8")
-    raise ValueError("Unknown type %r" % type(v))
+    raise ValueError(f"Unknown type {type(v)} for `{v}`.")
 
 
 def decode_dict(d: Dict[bytes, bytes], exclude_keys: Set[str]) -> Dict[str, str]:

scheduler/redis_models/result.py

@@ -3,9 +3,9 @@
 from enum import Enum
 from typing import Optional, Any, Self, ClassVar, List
 
-from scheduler.types import ConnectionType
 from scheduler.helpers.utils import utcnow
 from scheduler.redis_models.base import StreamModel, decode_dict
+from scheduler.types import ConnectionType
 
 
 class ResultType(Enum):
@@ -31,14 +31,14 @@ class Result(StreamModel):
 
     @classmethod
     def create(
-        cls,
-        connection: ConnectionType,
-        job_name: str,
-        worker_name: str,
-        _type: ResultType,
-        ttl: int,
-        return_value: Any = None,
-        exc_string: Optional[str] = None,
+            cls,
+            connection: ConnectionType,
+            job_name: str,
+            worker_name: str,
+            _type: ResultType,
+            ttl: int,
+            return_value: Any = None,
+            exc_string: Optional[str] = None,
     ) -> Self:
         result = cls(
             parent=job_name,
@@ -65,15 +65,3 @@ def fetch_latest(cls, connection: ConnectionType, job_name: str) -> Optional["Re
         result_id, payload = response[0]
         res = cls.deserialize(decode_dict(payload, set()))
         return res
-
-    def __repr__(self):
-        return f"Result(name={self.name}, type={self.type.name})"
-
-    def __eq__(self, other: Self) -> bool:
-        try:
-            return self.name == other.name
-        except AttributeError:
-            return False
-
-    def __bool__(self) -> bool:
-        return bool(self.name)

scheduler/redis_models/worker.py

@@ -98,9 +98,9 @@ def cleanup(cls, connection: ConnectionType, queue_name: Optional[str] = None):
                 pipeline.exists(worker_key)
             worker_exist = pipeline.execute()
             invalid_workers = list()
-            for i, worker_key in enumerate(worker_keys):
+            for i, worker_name in enumerate(worker_names):
                 if not worker_exist[i]:
-                    invalid_workers.append(worker_key)
+                    invalid_workers.append(worker_name)
             if len(invalid_workers) == 0:
                 return
             for invalid_subset in _split_list(invalid_workers, MAX_KEYS):

scheduler/tests/test_worker/test_worker_creation.py

@@ -2,6 +2,7 @@
 import uuid
 
 from scheduler import settings
+from scheduler.redis_models import WorkerModel
 from scheduler.worker import create_worker
 from scheduler.tests import test_settings  # noqa
 from scheduler.tests.testtools import SchedulerBaseCase
@@ -41,3 +42,15 @@ def test_create_worker__scheduler_interval(self):
         self.assertEqual(worker.scheduler.interval, 1)
         settings.SCHEDULER_CONFIG.SCHEDULER_INTERVAL = prev
         worker.teardown()
+
+    def test_create_worker__cleanup(self):
+        worker = create_worker("default", name="test", burst=True, with_scheduler=False)
+        worker.bootstrap()
+        worker.connection.delete(WorkerModel.key_for(worker.name))
+        all_names = WorkerModel.all_names(worker.connection)
+        self.assertIn(worker.name, all_names)
+        # act
+        WorkerModel.cleanup(worker.connection, "default")
+        # assert
+        all_names = WorkerModel.all_names(worker.connection)
+        self.assertNotIn(worker.name, all_names)