Changeset 9094 for main/waeup.kofa/branches/uli-zc-async/src

Timestamp:

8 Aug 2012, 22:57:01 (12 years ago)

Author:

uli

Message:

Add tests for export job containers.

File:

• main/waeup.kofa/branches/uli-zc-async/src/waeup/kofa/utils/tests/test_batching.py (modified) (4 diffs)

main/waeup.kofa/branches/uli-zc-async/src/waeup/kofa/utils/tests/test_batching.py

@@ -23 +23 @@
 import tempfile
 import unittest
-from zc.async.interfaces import IJob
+from zc.async.interfaces import IJob, COMPLETED
 from zope import schema
 from zope.component import provideUtility, getGlobalSiteManager
@@ -31 +31 @@
 from zope.interface import Interface, implements, verify
 from waeup.kofa.app import University
-from waeup.kofa.interfaces import ICSVExporter, IBatchProcessor
+from waeup.kofa.interfaces import (
+    ICSVExporter, IBatchProcessor, IExportJobContainer, IJobManager)
 from waeup.kofa.testing import FunctionalLayer, FunctionalTestCase
 from waeup.kofa.utils.batching import (
-    ExporterBase, BatchProcessor, export_job, AsyncExportJob)
+    ExporterBase, BatchProcessor, export_job, AsyncExportJob,
+    ExportJobContainer)
 
 optionflags = (
@@ -341 +343 @@
     # A minimal fake exporter suitable to be called by export_jobs
     fields = ('name', 'dinoports', 'owner', 'taxpayer')
+    title = u'Dummy cave exporter'
 
     def export_all(self, site, filepath=None):
@@ -376 +379 @@
         verify.verifyObject(IJob, job)
         return
+
+
+class FakeJob(object):
+
+    status = COMPLETED
+    result = None
+
+class FakeJobWithResult(FakeJob):
+
+    def __init__(self):
+        dir_path = tempfile.mkdtemp()
+        self.result = os.path.join(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
+
+class ExportJobContainerTests(unittest.TestCase):
+    # Test ExportJobContainer
+
+    def setUp(self):
+        # register a suitable ICSVExporter as named utility
+        self.exporter = CaveExporter()
+        self.job_manager = FakeJobManager()
+        self.gsm = getGlobalSiteManager()
+        self.gsm.registerUtility(
+            self.exporter, ICSVExporter, name='cave_exporter')
+        self.gsm.registerUtility(
+            self.job_manager, IJobManager)
+
+    def tearDown(self):
+        self.gsm.unregisterUtility(self.exporter)
+        self.gsm.unregisterUtility(self.job_manager, IJobManager)
+
+    def test_export_job_interfaces(self):
+        # the ExportJobContainer implements promised interfaces correctly...
+        container = ExportJobContainer()
+        verify.verifyClass(IExportJobContainer, ExportJobContainer)
+        verify.verifyObject(IExportJobContainer, container)
+        return
+
+    def test_start_export_job(self):
+        # we can start jobs
+        container = ExportJobContainer()
+        container.start_export_job('cave_exporter', 'bob')
+        result = self.job_manager._jobs.values()[0]
+        self.assertTrue(IJob.providedBy(result))
+        self.assertEqual(
+            container.running_exports,
+            [('1', 'cave_exporter', 'bob')]
+            )
+        return
+
+    def test_get_running_export_jobs_all(self):
+        # we can get export jobs of all users
+        container = ExportJobContainer()
+        container.start_export_job('cave_exporter', 'bob')
+        container.start_export_job('cave_exporter', 'alice')
+        result = container.get_running_export_jobs()
+        self.assertEqual(
+            result,
+            [('1', 'cave_exporter', 'bob'),
+             ('2', 'cave_exporter', 'alice')]
+            )
+        return
+
+    def test_get_running_export_jobs_user(self):
+        # we can get the export jobs running for a certain user
+        container = ExportJobContainer()
+        container.start_export_job('cave_exporter', 'bob')
+        container.start_export_job('cave_exporter', 'alice')
+        result1 = container.get_running_export_jobs(user_id='alice')
+        result2 = container.get_running_export_jobs(user_id='foo')
+        self.assertEqual(
+            result1, [('2', 'cave_exporter', 'alice')])
+        self.assertEqual(
+            result2, [])
+        return
+
+    def test_get_running_export_jobs_only_if_exist(self):
+        # we get only jobs that are accessible through the job manager...
+        container = ExportJobContainer()
+        container.start_export_job('cave_exporter', 'bob')
+        container.start_export_job('cave_exporter', 'bob')
+        self.assertTrue(
+            ('2', 'cave_exporter', 'bob') in container.running_exports)
+        # we remove the second entry from job manager
+        del self.job_manager._jobs['2']
+        result = container.get_running_export_jobs(user_id='bob')
+        self.assertEqual(
+            result, [('1', 'cave_exporter', 'bob')])
+        self.assertTrue(
+            ('2', 'cave_exporter', 'bob') not in container.running_exports)
+        return
+
+    def test_get_export_job_status(self):
+        # we can get the stati of jobs...
+        container = ExportJobContainer()
+        container.start_export_job('cave_exporter', 'alice')
+        container.start_export_job('cave_exporter', 'bob')
+        container.start_export_job('cave_exporter', 'bob')
+        result = container.get_export_jobs_status(user_id='bob')
+        # we'll get the raw value, a translation and the title of the
+        # exporter
+        self.assertEqual(
+            result,
+            [('new', u'new', u'Dummy cave exporter'),
+             ('completed', u'completed', u'Dummy cave exporter')]
+            )
+        return
+
+    def test_delete_export_entry(self):
+        # we can remove export entries in local lists and the job
+        # manager as well...
+        container = ExportJobContainer()
+        container.start_export_job('cave_exporter', 'bob')
+        entry = container.running_exports[0]
+        container.delete_export_entry(entry)
+        # both, running_exports list and job manager are empty now
+        self.assertEqual(
+            container.running_exports, [])
+        self.assertEqual(
+            self.job_manager._jobs, {})
+        return
+
+    def test_delete_export_entry_remove_file(self):
+        # any result files of exports are deleted as well
+        container = ExportJobContainer()
+        entry = ('4', 'cave_exporter', 'bob')
+        container.running_exports = [entry]
+        fake_job = FakeJobWithResult()
+        self.job_manager._jobs['4'] = fake_job
+        self.assertTrue(os.path.isfile(fake_job.result))
+        container.delete_export_entry(entry)
+        self.assertTrue(not os.path.exists(fake_job.result))
+        return