Changeset 9001 for main/waeup.kofa

Timestamp:

13 Jul 2012, 16:17:44 (12 years ago)

Author:

Henrik Bettermann

Message:

Update local roles by import. Test if ignore and deltion marker are doing what they are supposed to do.

Location:

main/waeup.kofa/trunk/src/waeup/kofa

Files:

• students/batching.py (modified) (1 diff)
• university/batching.py (modified) (3 diffs)
• university/tests/sample_faculty_data_update.csv (added)
• university/tests/test_batching.py (modified) (4 diffs)

main/waeup.kofa/trunk/src/waeup/kofa/students/batching.py

@@ -184 +184 @@
 
         # Update password
+        # XXX: Tale DELETION_MARKER into consideration
         if row.has_key('password'):
             passwd = row.get('password', IGNORE_MARKER)

main/waeup.kofa/trunk/src/waeup/kofa/university/batching.py

@@ -30 +30 @@
 from zope.catalog.interfaces import ICatalog
 from zope.event import notify
-from zope.securitypolicy.interfaces import IPrincipalRoleManager
+from zope.securitypolicy.interfaces import (
+    IPrincipalRoleManager, IPrincipalRoleMap)
 from waeup.kofa.authentication import LocalRoleSetEvent
-from waeup.kofa.interfaces import IBatchProcessor, IGNORE_MARKER, FatalCSVError
+from waeup.kofa.interfaces import (
+    IBatchProcessor, IGNORE_MARKER, DELETION_MARKER, FatalCSVError)
 from waeup.kofa.university.interfaces import (
     IFacultiesContainer, IFaculty, ICourse, IDepartment, ICertificate,
@@ -95 +97 @@
         items_changed = ''
 
-        if row.has_key('local_roles'):
-            local_roles = eval(row['local_roles'])
-            for rolemap in local_roles:
-                user = rolemap['user_name']
-                local_role = rolemap['local_role']
-                role_manager = IPrincipalRoleManager(obj)
-                role_manager.assignRoleToPrincipal(local_role, user)
-                notify(LocalRoleSetEvent(obj, local_role, user, granted=True))
-                items_changed += (
-                    '%s=%s, ' % ('local_roles', '%s|%s' % (user,local_role)))
+        if row.has_key('local_roles') and row['local_roles'] not in (
+            None, IGNORE_MARKER):
+            role_manager = IPrincipalRoleManager(obj)
+            role_map = IPrincipalRoleMap(obj)
+            # Remove all existing local roles.
+            for local_role, user_name, setting in role_map.getPrincipalsAndRoles():
+                role_manager.unsetRoleForPrincipal(local_role, user_name)
+                notify(LocalRoleSetEvent(
+                        obj, local_role, user_name, granted=False))
+            # Add new local roles.
+            if row['local_roles'] != DELETION_MARKER:
+                local_roles = eval(row['local_roles'])
+                for rolemap in local_roles:
+                    user = rolemap['user_name']
+                    local_role = rolemap['local_role']
+                    role_manager.assignRoleToPrincipal(local_role, user)
+                    notify(LocalRoleSetEvent(obj, local_role, user, granted=True))
+                    items_changed += (
+                        '%s=%s, ' % ('local_roles', '%s|%s' % (user,local_role)))
             row.pop('local_roles')
 
@@ -122 +133 @@
             FacultyProcessor, self).checkConversion(row, mode=mode)
         if row.has_key('local_roles'):
+            if row['local_roles'] in (None, DELETION_MARKER, IGNORE_MARKER):
+                return errs, inv_errs, conv_dict
             try:
                 local_roles = eval(row['local_roles'])

main/waeup.kofa/trunk/src/waeup/kofa/university/tests/test_batching.py

@@ -24 +24 @@
 from zope.component.hooks import setSite, clearSite
 from zope.component import createObject
-from zope.securitypolicy.interfaces import IPrincipalRoleMap
+from zope.securitypolicy.interfaces import (
+    IPrincipalRoleMap, IPrincipalRoleManager)
 from zope.interface.verify import verifyClass, verifyObject
 from waeup.kofa.interfaces import IBatchProcessor
@@ -42 +43 @@
 
 FACULTY_HEADER_FIELDS = FACULTY_SAMPLE_DATA.split(
+    '\n')[0].split(',')
+
+FACULTY_SAMPLE_DATA_UPDATE = open(
+    os.path.join(os.path.dirname(__file__), 'sample_faculty_data_update.csv'),
+    'rb').read()
+
+FACULTY_HEADER_FIELDS_UPDATE = FACULTY_SAMPLE_DATA_UPDATE.split(
     '\n')[0].split(',')
 
@@ -108 +116 @@
         self.csv_file_faculty = os.path.join(self.workdir, 'sample_faculty_data.csv')
         open(self.csv_file_faculty, 'wb').write(FACULTY_SAMPLE_DATA)
+        self.csv_file_faculty_update = os.path.join(self.workdir, 'sample_faculty_data_update.csv')
+        open(self.csv_file_faculty_update, 'wb').write(FACULTY_SAMPLE_DATA_UPDATE)
         return
 
@@ -214 +224 @@
         self.assertEqual(user_name, 'bob')
         self.assertEqual(local_role, 'waeup.local.DepartmentManager')
+        shutil.rmtree(os.path.dirname(fin_file))
+        return
+
+    def test_import_update(self):
+        self.app['faculties']['FAC2'] = Faculty(code='FAC2')
+        self.app['faculties']['FAC3'] = Faculty(code='FAC3')
+        self.app['faculties']['FAC4'] = Faculty(code='FAC4')
+
+        role_manager1 = IPrincipalRoleManager(self.app['faculties']['FAC1'])
+        role_manager1.assignRoleToPrincipal('alfonsrole', 'alfons')
+        role_map1 = IPrincipalRoleMap(self.app['faculties']['FAC1'])
+        self.assertEqual(len(role_map1.getPrincipalsAndRoles()), 1)
+
+        role_manager2 = IPrincipalRoleManager(self.app['faculties']['FAC2'])
+        role_manager2.assignRoleToPrincipal('alfonsrole', 'alfons')
+        role_map2 = IPrincipalRoleMap(self.app['faculties']['FAC2'])
+        self.assertEqual(len(role_map2.getPrincipalsAndRoles()), 1)
+
+        role_manager3 = IPrincipalRoleManager(self.app['faculties']['FAC3'])
+        role_manager3.assignRoleToPrincipal('alfonsrole', 'alfons')
+        role_map3 = IPrincipalRoleMap(self.app['faculties']['FAC3'])
+        self.assertEqual(len(role_map3.getPrincipalsAndRoles()), 1)
+
+        role_manager4 = IPrincipalRoleManager(self.app['faculties']['FAC4'])
+        role_manager4.assignRoleToPrincipal('alfonsrole', 'alfons')
+        role_map4 = IPrincipalRoleMap(self.app['faculties']['FAC4'])
+        self.assertEqual(len(role_map4.getPrincipalsAndRoles()), 1)
+
+        local_role, user_name, setting = role_map2.getPrincipalsAndRoles()[0]
+        self.assertEqual(user_name, 'alfons')
+        self.assertEqual(local_role, 'alfonsrole')
+
+        num, num_warns, fin_file, fail_file = self.proc.doImport(
+            self.csv_file_faculty_update, FACULTY_HEADER_FIELDS_UPDATE, 'update')
+        self.assertEqual(num_warns,0)
+        # Local roles have been removed in FAC1 due to deletion marker.
+        self.assertEqual(len(role_map1.getPrincipalsAndRoles()), 0)
+        # Old local roles have been removed and new roles have been added in FAC2.
+        self.assertEqual(len(role_map2.getPrincipalsAndRoles()), 1)
+        local_role, user_name, setting = role_map2.getPrincipalsAndRoles()[0]
+        self.assertEqual(user_name, 'bob')
+        self.assertEqual(local_role, 'waeup.local.DepartmentManager')
+        # Local roles are not touched in FAC3 due to ignore marker.
+        self.assertEqual(len(role_map3.getPrincipalsAndRoles()), 1)
+        local_role, user_name, setting = role_map3.getPrincipalsAndRoles()[0]
+        self.assertEqual(user_name, 'alfons')
+        self.assertEqual(local_role, 'alfonsrole')
+        # Local roles are not touched in FAC4 due to empty cell.
+        self.assertEqual(len(role_map4.getPrincipalsAndRoles()), 1)
+        local_role, user_name, setting = role_map4.getPrincipalsAndRoles()[0]
+        self.assertEqual(user_name, 'alfons')
+        self.assertEqual(local_role, 'alfonsrole')
         shutil.rmtree(os.path.dirname(fin_file))
         return