unittests were actually creating the celerybeat schedule database file, which it is not allowed to

celery/apps/beat.py

@@ -94,13 +94,14 @@ class Beat(object):
         self.app.loader.init_worker()
 
     def startup_info(self, beat):
+        scheduler = beat.get_scheduler(lazy=True)
         return STARTUP_INFO_FMT % {
             "conninfo": self.app.amqp.format_broker_info(),
             "logfile": self.logfile or "[stderr]",
             "loglevel": LOG_LEVELS[self.loglevel],
             "loader": get_full_cls_name(self.app.loader.__class__),
-            "scheduler": get_full_cls_name(beat.scheduler.__class__),
-            "scheduler_info": beat.scheduler.info,
+            "scheduler": get_full_cls_name(scheduler.__class__),
+            "scheduler_info": scheduler.info,
             "hmax_interval": humanize_seconds(beat.max_interval),
             "max_interval": beat.max_interval,
         }

celery/beat.py

@@ -132,7 +132,7 @@ class Scheduler(UserDict):
     Entry = ScheduleEntry
 
     def __init__(self, schedule=None, logger=None, max_interval=None,
-            app=None, Publisher=None, **kwargs):
+            app=None, Publisher=None, lazy=False, **kwargs):
         UserDict.__init__(self)
         if schedule is None:
             schedule = {}
@@ -142,8 +142,9 @@ class Scheduler(UserDict):
         self.logger = logger or self.app.log.get_default_logger(
                                                 name="celery.beat")
         self.max_interval = max_interval or conf.CELERYBEAT_MAX_LOOP_INTERVAL
-        self.setup_schedule()
         self.Publisher = Publisher or self.app.amqp.TaskPublisher
+        if not lazy:
+            self.setup_schedule()
 
     def maybe_due(self, entry, publisher=None):
         is_due, next_time_to_run = entry.is_due()
@@ -343,16 +344,22 @@ class Service(object):
         self._shutdown.set()
         wait and self._stopped.wait()           # block until shutdown done.
 
+    def get_scheduler(self, lazy=False):
+        filename = self.schedule_filename
+        scheduler = instantiate(self.scheduler_cls,
+                                app=self.app,
+                                schedule_filename=filename,
+                                logger=self.logger,
+                                max_interval=self.max_interval,
+                                lazy=lazy)
+        if not lazy:
+            scheduler.update_from_dict(self.schedule)
+        return scheduler
+
     @property
     def scheduler(self):
         if self._scheduler is None:
-            filename = self.schedule_filename
-            self._scheduler = instantiate(self.scheduler_cls,
-                                          app=self.app,
-                                          schedule_filename=filename,
-                                          logger=self.logger,
-                                          max_interval=self.max_interval)
-            self._scheduler.update_from_dict(self.schedule)
+            self._scheduler = self.get_scheduler()
         return self._scheduler
 
 

celery/concurrency/processes/__init__.py

@@ -5,8 +5,6 @@ Process Pools.
 """
 import traceback
 
-from time import sleep, time
-
 from celery import log
 from celery.datastructures import ExceptionInfo
 from celery.utils.functional import partial

celery/tests/test_bin/test_celerybeat.py

@@ -7,11 +7,21 @@ from celery import platforms
 from celery.app import app_or_default
 from celery.bin import celerybeat as celerybeat_bin
 from celery.apps import beat as beatapp
+from celery.utils.compat import defaultdict
+
+
+class MockedShelveModule(object):
+    shelves = defaultdict(lambda: {})
+
+    def open(self, filename, *args, **kwargs):
+        return self.shelves[filename]
+mocked_shelve = MockedShelveModule()
 
 
 class MockService(beat.Service):
     started = False
     in_sync = False
+    persistence = mocked_shelve
 
     def start(self):
         self.__class__.started = True

celery/tests/test_concurrency_processes.py

@@ -7,8 +7,6 @@ from celery.concurrency import processes as mp
 from celery.datastructures import ExceptionInfo
 from celery.utils import noop
 
-from celery.tests.utils import skip_if_quick
-
 
 class Object(object):   # for writeable attributes.