Changeset 9166 for main/waeup.kofa/branches/uli-zc-async/src/waeup/kofa/browser/tests

Timestamp:

6 Sep 2012, 16:50:17 (12 years ago)

Author:

uli

Message:

All remaining changes from last weeks. Sorry for the mess.

Location:

main/waeup.kofa/branches/uli-zc-async/src/waeup/kofa/browser/tests

Files:

• test_async.py (added)
• test_browser.py (modified) (5 diffs)

main/waeup.kofa/branches/uli-zc-async/src/waeup/kofa/browser/tests/test_browser.py

@@ -26 +26 @@
 import os
 import grok
+from zc.async.testing import wait_for_result
 from zope.event import notify
-from zope.component import createObject, queryUtility
+from zope.component import createObject, queryUtility, getUtility
 from zope.component.hooks import setSite, clearSite
 from zope.catalog.interfaces import ICatalog
@@ -36 +37 @@
 from waeup.kofa.testing import FunctionalLayer, FunctionalTestCase
 from waeup.kofa.app import University
+from waeup.kofa.interfaces import IJobManager
+from waeup.kofa.tests.test_async import FunctionalAsyncTestCase
 from waeup.kofa.university.faculty import Faculty
 from waeup.kofa.university.department import Department
+
+
+
 
 SAMPLE_FILE = os.path.join(os.path.dirname(__file__), 'test_file.csv')
@@ -106 +112 @@
         shutil.rmtree(self.dc_root)
 
-
 class DataCenterUITests(UniversitySetup):
     # Tests for DataCenter class views and pages
@@ -149 +154 @@
         return
 
-    def test_export(self):
+
+class DataCenterUIExportTests(UniversitySetup, FunctionalAsyncTestCase):
+    # Tests for DataCenter class views and pages
+
+    layer = FunctionalLayer
+
+    def wait_for_export_job_completed(self):
+        # helper function waiting until the current export job is completed
+        manager = getUtility(IJobManager)
+        job_id = self.app['datacenter'].running_exports[0][0]
+        job = manager.get(job_id)
+        wait_for_result(job)
+        return job_id
+
+    def stored_in_datacenter(self, job_id):
+        # tell whether job_id is stored in datacenter's running jobs list
+        for entry in list(self.app['datacenter'].running_exports):
+            if entry[0] == job_id:
+                return True
+        return False
+
+    def test_export_start(self):
+        # we can trigger export file creation
         self.browser.addHeader('Authorization', 'Basic mgr:mgrpw')
         self.browser.open(self.datacenter_path)
@@ -156 +183 @@
         self.browser.getLink("Export CSV file").click()
         self.browser.getControl(name="exporter").value = ['faculties']
+        self.browser.getControl("Create CSV file").click()
+        self.assertEqual(self.browser.headers['Status'], '200 Ok')
+        return
+
+    def test_export_download(self):
+        # we can download a generated export result
+        self.test_export_start()
+        # while the export file is created, we get a reload button
+        # (or a loading bar if javascript is enabled)...
+        self.browser.getControl("Reload").click()
+        # ...which is displayed as long as the job is not finished.
+        # When the job is finished and we reload the page...
+        job_id = self.wait_for_export_job_completed()
+        try:
+            self.browser.getControl("Reload").click()
+        except LookupError:
+            # if the job completed very fast, we will get the download
+            # link immediately
+            pass
+        # ...we can download the result
         self.browser.getControl("Download").click()
-        self.assertEqual(self.browser.headers['Status'], '200 Ok')
         self.assertEqual(self.browser.headers['Content-Type'],
                          'text/csv; charset=UTF-8')
         self.assertEqual(self.browser.contents,
             'code,title,title_prefix\r\nfac1,Unnamed Faculty,faculty\r\n')
-        logfile = os.path.join(
-            self.app['datacenter'].storage, 'logs', 'datacenter.log')
-        logcontent = open(logfile).read()
-        self.assertTrue('zope.mgr - browser.pages.ExportCSVView - '
-                        'exported: faculties' in logcontent)
-        return
+
+        # after download, the job and the result file are removed
+        manager = getUtility(IJobManager)
+        result = manager.get(job_id)
+        self.assertEqual(result, None)
+        self.assertEqual(self.stored_in_datacenter(job_id), False)
+        #logfile = os.path.join(
+        #    self.app['datacenter'].storage, 'logs', 'datacenter.log')
+        #logcontent = open(logfile).read()
+        #self.assertTrue('zope.mgr - browser.pages.ExportCSVView - '
+        #                'exported: faculties' in logcontent)
+        return
+
+    def test_export_discard(self):
+        # we can discard a generated export result
+        self.test_export_start()
+        self.wait_for_export_job_completed()
+        self.browser.open(self.datacenter_path + '/@@export')
+        self.browser.getControl("Discard").click()
+        self.assertTrue('Discarded export result' in self.browser.contents)
+        return