Changeset 9217 for main/waeup.kofa/trunk/src/waeup/kofa/browser/tests

Timestamp:

21 Sep 2012, 11:21:05 (12 years ago)

Author:

uli

Message:

Merge changes from uli-async-update back into trunk.

Location:

main/waeup.kofa/trunk/src/waeup/kofa/browser/tests

Files:

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

main/waeup.kofa/trunk/src/waeup/kofa/browser/tests/test_browser.py

@@ -21 +21 @@
 import shutil
 import tempfile
-import pytz
-from datetime import datetime, timedelta
-from StringIO import StringIO
 import os
-import grok
-from zope.event import notify
-from zope.component import createObject, queryUtility
+from zc.async.testing import wait_for_result
+from zope.component import createObject, getUtility
 from zope.component.hooks import setSite, clearSite
-from zope.catalog.interfaces import ICatalog
 from zope.security.interfaces import Unauthorized
-from zope.securitypolicy.interfaces import IPrincipalRoleManager
 from zope.testbrowser.testing import Browser
-from hurry.workflow.interfaces import IWorkflowInfo, IWorkflowState
 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
@@ -106 +101 @@
         shutil.rmtree(self.dc_root)
 
-
 class DataCenterUITests(UniversitySetup):
     # Tests for DataCenter class views and pages
@@ -149 +143 @@
         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 +172 @@
         self.browser.getLink("Export data").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'],
+        self.assertEqual(self.browser.headers['content-type'],
                          'text/csv; charset=UTF-8')
-        self.assertTrue ('WAeUP.KofaFaculties.csv' in
-            self.browser.headers['content-disposition'])
+        self.assertEqual(self.browser.headers['content-disposition'],
+                         'attachment; filename="WAeUP.Kofa_faculties.csv')
         self.assertEqual(self.browser.contents,
             'code,title,title_prefix,users_with_local_roles\r\n'
             'fac1,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)
+
+        # 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