Changeset 13806

Timestamp:

6 Apr 2016, 10:27:11 (9 years ago)

Author:

Henrik Bettermann

Message:

Add portal maintenance mode.

See r13394, r13396, r13468.

Location:

main/waeup.ikoba/trunk

Files:

• CHANGES.txt (modified) (1 diff)
• layout/static/img/actionicon_stop.png (added)
• src/waeup/ikoba/authentication.py (modified) (1 diff)
• src/waeup/ikoba/browser/pages.py (modified) (9 diffs)
• src/waeup/ikoba/browser/tests/test_browser.py (modified) (2 diffs)
• src/waeup/ikoba/browser/viewlets.py (modified) (1 diff)
• src/waeup/ikoba/customers/authentication.py (modified) (1 diff)
• src/waeup/ikoba/customers/tests/test_browser.py (modified) (1 diff)
• src/waeup/ikoba/doctests/batchprocessing_browser.txt (modified) (3 diffs)
• src/waeup/ikoba/interfaces.py (modified) (1 diff)
• src/waeup/ikoba/products/tests/test_batching.py (modified) (3 diffs)

main/waeup.ikoba/trunk/CHANGES.txt

@@ -4 +4 @@
 0.2.dev0 (unreleased)
 =====================
+
+* Add portal maintenance mode.
+
+* Forbid style and script elements in html fields.
 
 * Rearrange datacenter upload page. Provide modal windows to view all

main/waeup.ikoba/trunk/src/waeup/ikoba/authentication.py

@@ -223 +223 @@
 
     def checkPassword(self, password):
+        try:
+            blocker = grok.getSite()['configuration'].maintmode_enabled_by
+            if blocker and blocker != self.name:
+                return False
+        except (TypeError, KeyError):  # in unit tests
+            pass
         if not isinstance(password, basestring):
             return False

main/waeup.ikoba/trunk/src/waeup/ikoba/browser/pages.py

@@ -308 +308 @@
 
     def update(self, SUBMIT=None, camefrom=None):
+        blocker = grok.getSite()['configuration'].maintmode_enabled_by
+        if blocker:
+            self.flash(_('The portal is in maintenance mode '
+                        'and authentication (login) temporarily disabled.'),
+                       type='warning')
         self.camefrom = camefrom
         if SUBMIT is not None:
@@ -335 +340 @@
                     return
                 self.redirect(self.camefrom)
+                return
+            # Display second warning message and remind user that
+            # s/he can't login.
+            if blocker:
+                self.flash(_('The portal is in maintenance mode. '
+                            'You can\'t login!'),
+                           type='danger')
                 return
             # Display appropriate flash message if credentials are correct
@@ -638 +650 @@
 
     def render(self):
-        #ob_class = self.__implemented__.__name__.replace('waeup.kofa.','')
+        #ob_class = self.__implemented__.__name__.replace('waeup.ikoba.','')
         #self.context.logger.info(
         #    '%s - skeleton downloaded: %s' % (ob_class, self.filename))
@@ -921 +933 @@
     form_fields = grok.AutoFields(IConfigurationContainer).omit(
         'frontpage_dict')
+    form_fields['maintmode_enabled_by'].for_display = True
 
     @action(_('Save'), style='primary')
@@ -1193 +1206 @@
 
     def update(self, filename=None, select=None, cancel=None):
+        if not grok.getSite()['configuration'].maintmode_enabled_by and \
+            not self.request.principal.id == 'admin':
+            self.flash(
+                _('Portal must be in maintenance mode for data import.'),
+                type='warning')
+            self.redirect(self.url(self.context))
+            return
         if cancel is not None:
             self.redirect(self.url(self.context))
@@ -1201 +1221 @@
             session['import_filename'] = select
             self.redirect(self.url(self.context, '@@import2'))
+        return
 
 class DatacenterImportStep2(IkobaPage):
@@ -1440 +1461 @@
             self.redirect(self.url(self.context, '@@import1'))
             return
+        if not grok.getSite()['configuration'].maintmode_enabled_by and \
+            not self.request.principal.id == 'admin':
+            self.flash(
+                _('File has not been imported. '
+                  'Portal must be in maintenance mode for data import.'),
+                type='danger')
+            self.redirect(self.url(self.context))
+            return
         session = ISession(self.request)['waeup.ikoba']
         self.filename = session.get('import_filename', None)
@@ -1476 +1505 @@
             self.warn_num == 0, self.fullpath, fin_path, pending_path,
             self.mode)
+        # Disable maintenance mode if set.
+        if grok.getSite()['configuration'].maintmode_enabled_by:
+            grok.getSite()['configuration'].maintmode_enabled_by = None
+            self.flash(_('Maintenance mode disabled.'))
 
         if self.warn_num:
@@ -1659 +1692 @@
         return result
 
+class SwitchMaintModePage(UtilityView, grok.View):
+    """Import managers must disable authentication for all other users
+    before starting an import. This pages switches maintenance mode
+    on and off.
+    """
+    grok.context(IDataCenter)
+    grok.name('switchmaintmode')
+    grok.require('waeup.importData')
+
+    def update(self):
+        ob_class = self.__implemented__.__name__.replace('waeup.ikoba.','')
+        if grok.getSite()['configuration'].maintmode_enabled_by:
+            grok.getSite()['configuration'].maintmode_enabled_by = None
+            self.context.logger.info('%s - maintmode disabled' % ob_class)
+            self.flash(_('Maintenance mode disabled.'))
+            self.redirect(self.url(self.context))
+        else:
+            grok.getSite()['configuration'].maintmode_enabled_by = unicode(
+                self.request.principal.id)
+            self.context.logger.info('%s - maintmode enabled' % ob_class)
+            self.flash(_('Maintenance mode enabled.'), type='warning')
+            self.redirect(self.url(self.context, 'import1'))
+        return
+
+    def render(self):
+        return
+
+
 class ChangePasswordRequestPage(IkobaFormPage):
     """Captcha'd page for all kind of users to request a password change.

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

@@ -22 +22 @@
 import tempfile
 import os
+import grok
 from zc.async.testing import wait_for_result
 from zope.component import createObject, getUtility
@@ -215 +216 @@
         return
 
+    def test_maintenance_mode(self):
+        config = grok.getSite()['configuration']
+        self.app['users'].addUser(
+            'officer', 'secret', title='Bob Officer', email='bob@abcd.ng')
+        self.browser.open('http://localhost/app/login')
+        self.browser.getControl(name="form.login").value = 'officer'
+        self.browser.getControl(name="form.password").value = 'secret'
+        self.browser.getControl("Login").click()
+        # Officer logged in.
+        self.assertMatches('...You logged in...', self.browser.contents)
+        self.assertTrue("Bob Officer" in self.browser.contents)
+        # If maintenance mode is enabled, officer is immediately logged out.
+        config.maintmode_enabled_by = u'any_user'
+        self.assertRaises(
+            Unauthorized, self.browser.open, 'http://localhost/app/customers')
+        self.browser.open('http://localhost/app/login')
+        self.assertMatches(
+            '...The portal is in maintenance mode...',
+            self.browser.contents)
+        # Officers really can't login if maintenance mode is enabled.
+        self.browser.getControl(name="form.login").value = 'officer'
+        self.browser.getControl(name="form.password").value = 'secret'
+        self.browser.getControl("Login").click()
+        # A second warning is raised.
+        self.assertMatches(
+            '...The portal is in maintenance mode. You can\'t login!...',
+            self.browser.contents)
+        # Offficer can login if s/he is the blocker.
+        config.maintmode_enabled_by = u'officer'
+        self.browser.getControl(name="form.login").value = 'officer'
+        self.browser.getControl(name="form.password").value = 'secret'
+        self.browser.getControl("Login").click()
+        self.assertTrue('You logged in' in self.browser.contents)
+        self.assertTrue('Logout' in self.browser.contents)
+        return
+
     def test_sources_overview(self):
         self.browser.open('http://localhost/app/sources')

main/waeup.ikoba/trunk/src/waeup/ikoba/browser/viewlets.py

@@ -454 +454 @@
     text = _('View processed files')
 
+class SwitchMaintMode(ActionButton):
+    grok.context(IDataCenter)
+    grok.require('waeup.importData')
+    #grok.view(DatacenterImportStep1)
+    grok.order(7)
+    icon = 'actionicon_stop.png'
+    target = 'switchmaintmode'
+    text = _('Switch maintenance mode')
+
+    @property
+    def onclick(self):
+        if grok.getSite()['configuration'].maintmode_enabled_by:
+            return
+        return "return window.confirm(%s);" % _(
+            "'In maintenance mode no other user can login, and "
+            "already logged-in users will be automatically logged out. "
+            "You will be the only logged-in user and you can safely start "
+            "any import. Please wait a few seconds before starting the import "
+            "so that all running processes are finished. "
+            "If the import is done, maintenance mode will "
+            "be automatically disabled. \\n\\n"
+            "You really want to enable maintenance mode?'")
+
 #
 # Primary navigation tabs (in upper left navigation bar)...

main/waeup.ikoba/trunk/src/waeup/ikoba/customers/authentication.py

@@ -116 +116 @@
         customers or the temporary password set by the system.
 
-        We additionally check if customer account has been suspended.
-        """
+        We additionally check if customer account has been suspended
+        or maintenance mode is on.
+        """
+        try:
+            blocker = grok.getSite()['configuration'].maintmode_enabled_by
+            if blocker and blocker != self.name:
+                return False
+        except (TypeError, KeyError):  # in unit tests
+            pass
         if not isinstance(password, basestring):
             return False

main/waeup.ikoba/trunk/src/waeup/ikoba/customers/tests/test_browser.py

@@ -492 +492 @@
         self.browser.getControl(name='SUBMIT').click()
         self.browser.getLink('Process data').click()
+        self.browser.getLink("Switch maintenance mode").click()
         button = lookup_submit_value(
             'select', 'customers_zope.mgr.csv', self.browser)

main/waeup.ikoba/trunk/src/waeup/ikoba/doctests/batchprocessing_browser.txt

@@ -84 +84 @@
 
     >>> browser.getLink('Process data').click()
+    >>> browser.getLink('Switch maintenance mode').click()
     >>> button = lookup_submit_value(
     ...   'select', 'users_zope.mgr.csv', browser)
@@ -170 +171 @@
 
     >>> browser.getLink('Process data').click()
+    >>> browser.getLink('Switch maintenance mode').click()
     >>> button = lookup_submit_value(
     ...   'select', 'newusers_zope.mgr.csv', browser)
@@ -232 +234 @@
     >>> browser.open('http://localhost/app/datacenter')
     >>> browser.getLink('Process data').click()
+    >>> browser.getLink('Switch maintenance mode').click()
     >>> button = lookup_submit_value(
     ...   'select', 'newusers_zope.mgr.create.pending.csv', browser)

main/waeup.ikoba/trunk/src/waeup/ikoba/interfaces.py

@@ -650 +650 @@
         )
 
+    maintmode_enabled_by = schema.TextLine(
+        title = _(u'Maintenance Mode enabled by'),
+        default = None,
+        required = False,
+        )
+
 
 class IDataCenter(IIkobaObject):

main/waeup.ikoba/trunk/src/waeup/ikoba/products/tests/test_batching.py

@@ -33 +33 @@
 from zope.interface.verify import verifyClass, verifyObject
 from zope.securitypolicy.interfaces import IPrincipalRoleManager
+from zope.testbrowser.testing import Browser
 
 from waeup.ikoba.app import Company
@@ -64 +65 @@
         self.workdir = tempfile.mkdtemp()
         app = Company()
+        self.browser = Browser()
+        self.browser.handleErrors = False
+        self.datacenter_path = 'http://localhost/app/datacenter'
         app['datacenter'].setStoragePath(self.dc_root)
         self.getRootFolder()['app'] = app
@@ -202 +206 @@
             in failcontent)
         shutil.rmtree(os.path.dirname(fin_file))
+
+
+    def test_upload_import_reupload(self):
+        self.browser.addHeader('Authorization', 'Basic mgr:mgrpw')
+        self.browser.open(self.datacenter_path)
+        self.assertEqual(self.browser.headers['Status'], '200 Ok')
+        self.assertEqual(self.browser.url, self.datacenter_path)
+        self.browser.getLink("Upload data").click()
+        file = open(self.csv_file)
+        ctrl = self.browser.getControl(name='uploadfile:file')
+        file_ctrl = ctrl.mech_control
+        file_ctrl.add_file(file, filename='sample_product_data.csv')
+        self.browser.getControl('Upload').click()
+        self.browser.getLink('Process data').click()
+        self.assertTrue('Portal must be in maintenance mode for data import.'
+            in self.browser.contents)
+        self.browser.getLink("Switch").click()
+        self.assertTrue('Maintenance mode enabled' in self.browser.contents)
+        # Maintenance mode is now set.
+        self.assertEqual(
+            self.app['configuration'].maintmode_enabled_by, 'zope.mgr')
+        self.browser.getControl(name="select").click()
+        importerselect = self.browser.getControl(name='importer')
+        importerselect.getControl('Product Processor').selected = True
+        modeselect = self.browser.getControl(name='mode')
+        modeselect.getControl(value='create').selected = True
+        self.browser.getControl('Proceed to step 3').click()
+        self.assertTrue('Header fields OK' in self.browser.contents)
+        self.browser.getControl('Perform import').click()
+        self.assertTrue('Successfully processed 3 rows' in self.browser.contents)
+        self.assertTrue('Maintenance mode disabled' in self.browser.contents)
+        # Maintenance mode is really disabled.
+        self.assertEqual(
+            self.app['configuration'].maintmode_enabled_by, None)
+        # We import the same file a second time.
+        self.browser.open(self.datacenter_path)
+        self.browser.getLink("Upload data").click()
+        file = open(self.csv_file)
+        ctrl = self.browser.getControl(name='uploadfile:file')
+        file_ctrl = ctrl.mech_control
+        file_ctrl.add_file(file, filename='sample_product_data.csv')
+        self.browser.getControl('Upload').click()
+        self.assertTrue(
+            'File with same name was uploaded earlier' in self.browser.contents)
+        return