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

Timestamp:

8 Aug 2012, 13:27:57 (12 years ago)

Author:

uli

Message:

Add tests for export_job function.

File:

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

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

@@ -24 +24 @@
 import unittest
 from zope import schema
-from zope.component import provideUtility
+from zope.component import provideUtility, getGlobalSiteManager
 from zope.component.factory import Factory
 from zope.component.hooks import clearSite
@@ -32 +32 @@
 from waeup.kofa.interfaces import ICSVExporter, IBatchProcessor
 from waeup.kofa.testing import FunctionalLayer, FunctionalTestCase
-from waeup.kofa.utils.batching import ExporterBase
+from waeup.kofa.utils.batching import (
+    ExporterBase, BatchProcessor, export_job)
 
 optionflags = (
@@ -71 +72 @@
         self.owner = owner
         self.taxpayer = taxpayer
-#Cave = attrs_to_fields(Cave)
 
 stoneville = dict
 
-from waeup.kofa.utils.batching import BatchProcessor
+SAMPLE_DATA = """name,dinoports,owner,taxpayer
+Barneys Home,2,Barney,1
+Wilmas Asylum,1,Wilma,1
+Freds Dinoburgers,10,Fred,0
+Joeys Drive-in,110,Joey,0
+"""
+
 class CaveProcessor(BatchProcessor):
     util_name = 'caveprocessor'
-    #grok.name(util_name)
     name = 'Cave Processor'
     iface = ICave
@@ -136 +141 @@
         # Provide sample data
         self.newcomers_csv = os.path.join(self.workdir, 'newcomers.csv')
-        open(self.newcomers_csv, 'wb').write(
-            """name,dinoports,owner,taxpayer
-Barneys Home,2,Barney,1
-Wilmas Asylum,1,Wilma,1
-Freds Dinoburgers,10,Fred,0
-Joeys Drive-in,110,Joey,0
-""")
+        open(self.newcomers_csv, 'wb').write(SAMPLE_DATA)
         self.setupLogger()
         self.stoneville = stoneville
@@ -336 +335 @@
         self.assertEqual(result, None)
         return
+
+
+class CaveExporter(ExporterBase):
+    # A minimal fake exporter suitable to be called by export_jobs
+    fields = ('name', 'dinoports', 'owner', 'taxpayer')
+
+    def export_all(self, site, filepath=None):
+        if filepath is None:
+            return SAMPLE_DATA
+        open(filepath, 'wb').write(SAMPLE_DATA)
+        return
+
+class ExportJobTests(unittest.TestCase):
+    # Test asynchronous export functionality (simple cases)
+
+    def setUp(self):
+        # register a suitable ICSVExporter as named utility
+        self.exporter = CaveExporter()
+        self.gsm = getGlobalSiteManager()
+        self.gsm.registerUtility(
+            self.exporter, ICSVExporter, name='cave_exporter')
+
+    def tearDown(self):
+        self.gsm.unregisterUtility(self.exporter)
+
+    def test_export_job_func(self):
+        # the export_job func does really export data...
+        result_path = export_job(None, 'cave_exporter')
+        self.assertTrue(os.path.isfile(result_path))
+        contents = open(result_path, 'rb').read()
+        shutil.rmtree(os.path.dirname(result_path))
+        self.assertEqual(contents, SAMPLE_DATA)
+        return