Changeset 9342 for main/waeup.kofa/trunk/src/waeup/kofa

Timestamp:

16 Oct 2012, 09:42:18 (12 years ago)

Author:

uli

Message:

Move fake job components to a more general location.

Location:

main/waeup.kofa/trunk/src/waeup/kofa

Files:

• testing.py (modified) (2 diffs)
• utils/tests/test_batching.py (modified) (4 diffs)

main/waeup.kofa/trunk/src/waeup/kofa/testing.py

@@ -29 +29 @@
 import zope.component
 import waeup.kofa
+from zc.async.interfaces import COMPLETED
 from zope.app.testing.functional import (
     ZCMLLayer, FunctionalTestSetup, getRootFolder, sync, FunctionalTestCase)
@@ -430 +431 @@
         suite.addTest(test)
     return suite
+
+class FakeJob(object):
+    # A job usable for simple async tests
+    status = COMPLETED
+    result = None
+
+class FakeJobManager(object):
+    # A fake job manager for testing async functionality
+    _jobs = dict()
+    _curr_num = 1
+
+    def get(self, job_id):
+        if job_id == '3':
+            return FakeJob()
+        return self._jobs.get(job_id, None)
+
+    def put(self, job):
+        num = str(self._curr_num)
+        self._jobs[num] = job
+        self._curr_num += 1
+        return num
+
+    def remove(self, job_id, site):
+        if job_id in self._jobs:
+            del self._jobs[job_id]
+        return

main/waeup.kofa/trunk/src/waeup/kofa/utils/tests/test_batching.py

@@ -23 +23 @@
 import tempfile
 import unittest
-from zc.async.interfaces import IJob, COMPLETED
+from zc.async.interfaces import IJob
 from zope import schema
 from zope.component import provideUtility, getGlobalSiteManager
@@ -34 +34 @@
     ICSVExporter, IBatchProcessor, IExportJobContainer, IJobManager,
     IExportJob)
-from waeup.kofa.testing import FunctionalLayer, FunctionalTestCase
+from waeup.kofa.testing import (
+    FunctionalLayer, FunctionalTestCase, FakeJob, FakeJobManager)
 from waeup.kofa.utils.batching import (
     ExporterBase, BatchProcessor, export_job, AsyncExportJob,
@@ -384 +385 @@
 
 
-class FakeJob(object):
-
-    status = COMPLETED
-    result = None
-
 class FakeJobWithResult(FakeJob):
 
@@ -395 +391 @@
         self.result = os.path.join(self.dir_path, 'fake.csv')
         open(self.result, 'wb').write('a fake result')
-        return
-
-class FakeJobManager(object):
-
-    _jobs = dict()
-    _curr_num = 1
-
-    def get(self, job_id):
-        if job_id == '3':
-            return FakeJob()
-        return self._jobs.get(job_id, None)
-
-    def put(self, job):
-        num = str(self._curr_num)
-        self._jobs[num] = job
-        self._curr_num += 1
-        return num
-
-    def remove(self, job_id, site):
-        if job_id in self._jobs:
-            del self._jobs[job_id]
         return