Ignore:
Timestamp:
5 Apr 2016, 21:04:39 (9 years ago)
Author:
Henrik Bettermann
Message:

Do only allow one running export job.
Add switch to disable all exports.

See r13198 - r13201 and r13211.

File:
1 edited

Legend:

Unmodified
Added
Removed
  • main/waeup.ikoba/trunk/src/waeup/ikoba/browser/tests/test_browser.py

    r12842 r13802  
    6969
    7070
    71 class DataCenterUITests(CompanySetup):
     71class DataCenterUITests(CompanySetup, FunctionalAsyncTestCase):
    7272    # Tests for DataCenter class views and pages
    7373
    7474    layer = FunctionalLayer
     75
     76    def wait_for_export_job_completed(self):
     77        # helper function waiting until the current export job is completed
     78        manager = getUtility(IJobManager)
     79        job_id = self.app['datacenter'].running_exports[0][0]
     80        job = manager.get(job_id)
     81        wait_for_result(job)
     82        return job_id
     83
     84    def cleanup_run_job(self):
     85        # helper to remove any result file of an export
     86        job_id = self.wait_for_export_job_completed()
     87        manager = getUtility(IJobManager)
     88        job = manager.get(job_id)
     89        if os.path.exists(job.result):
     90            shutil.rmtree(os.path.dirname(job.result))
    7591
    7692    def test_user_data_import_permission(self):
     
    96112        self.assertTrue('User Processor' in self.browser.contents)
    97113        return
     114
     115    def test_export_disabled(self):
     116        self.app['configuration'].export_disabled_message = u'Export forbidden'
     117        self.browser.addHeader('Authorization', 'Basic mgr:mgrpw')
     118        self.browser.open(self.datacenter_path)
     119        self.assertEqual(self.browser.headers['Status'], '200 Ok')
     120        self.assertEqual(self.browser.url, self.datacenter_path)
     121        self.browser.getLink("Export data").click()
     122        self.browser.getControl(name="exporter").value = ['products']
     123        self.browser.getControl("Create CSV file").click()
     124        self.assertEqual(self.browser.headers['Status'], '200 Ok')
     125        self.assertTrue('div class="alert alert-danger">Export forbidden</div>'
     126                         in self.browser.contents)
     127        return
     128
     129    def test_export_start(self):
     130        # we can trigger export file creation
     131        self.browser.addHeader('Authorization', 'Basic mgr:mgrpw')
     132        self.browser.open(self.datacenter_path)
     133        self.assertEqual(self.browser.headers['Status'], '200 Ok')
     134        self.assertEqual(self.browser.url, self.datacenter_path)
     135        self.browser.getLink("Export data").click()
     136        self.browser.getControl(name="exporter").value = ['products']
     137        self.browser.getControl("Create CSV file").click()
     138        # we can trigger export file creation only once
     139        self.browser.getControl(name="exporter").value = ['pdfdocuments']
     140        self.browser.getControl("Create CSV file").click()
     141        self.assertTrue('Sorry, another export job is running.'
     142                         in self.browser.contents)
     143        self.wait_for_export_job_completed()
     144        self.assertEqual(len(self.app['datacenter'].running_exports), 1)
     145        self.browser.getControl(name="exporter").value = ['pdfdocuments']
     146        self.browser.getControl("Create CSV file").click()
     147        self.assertEqual(self.browser.headers['Status'], '200 Ok')
     148        self.assertEqual(len(self.app['datacenter'].running_exports), 2)
     149        # The second job is running
     150        self.assertTrue('<img src="/static/img/actionicon_reload.png" />'
     151                         in self.browser.contents)
     152        self.cleanup_run_job() # make sure to remove temp dirs
     153        # Also the second job must be cleaned up
     154        manager = getUtility(IJobManager)
     155        job_id = self.app['datacenter'].running_exports[1][0]
     156        job = manager.get(job_id)
     157        wait_for_result(job)
     158        if os.path.exists(job.result):
     159            shutil.rmtree(os.path.dirname(job.result))
     160        return
     161
Note: See TracChangeset for help on using the changeset viewer.