Changeset 8995 for main/waeup.kofa

Timestamp:

13 Jul 2012, 08:18:02 (12 years ago)

Author:

Henrik Bettermann

Message:

Customize updateEntry method of FacultyProcessor? to import local_roles.

Location:

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

Files:

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

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

@@ -223 +223 @@
         parent = self.getParent(row, site)
         if hasattr(obj,'student_id'):
-            # Update mode: the student exists and we can get the student_id
+            # Update mode: the student exists and we can get the student_id.
+            # Create mode: the record contains the student_id
             parent.logger.info(
                 '%s - Student record updated: %s'

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

@@ -29 +29 @@
 from zope.schema import getFields
 from zope.catalog.interfaces import ICatalog
+from zope.event import notify
+from zope.securitypolicy.interfaces import IPrincipalRoleManager
+from waeup.kofa.authentication import LocalRoleSetEvent
 from waeup.kofa.interfaces import IBatchProcessor, IGNORE_MARKER, FatalCSVError
 from waeup.kofa.university.interfaces import (
@@ -84 +87 @@
         del parent[row['code']]
         pass
+
+    def updateEntry(self, obj, row, site):
+        """Update obj to the values given in row.
+        """
+        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)))
+            row.pop('local_roles')
+
+        # apply other values...
+        items_changed += super(FacultyProcessor, self).updateEntry(
+            obj, row, site)
+
+        # Log actions...
+        grok.getSite().logger.info('%s - %s - Record updated: %s'
+            % (self.name, self.location_fields[0], items_changed))
+        return items_changed
 
     def checkConversion(self, row, mode='create'):

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

@@ -21 +21 @@
 import tempfile
 import shutil
+import os
 from zope.component.hooks import setSite, clearSite
+from zope.securitypolicy.interfaces import IPrincipalRoleMap
 from zope.interface.verify import verifyClass, verifyObject
 from waeup.kofa.interfaces import IBatchProcessor
@@ -34 +36 @@
 from waeup.kofa.university.batching import FacultyProcessor
 
+FACULTY_SAMPLE_DATA = open(
+    os.path.join(os.path.dirname(__file__), 'sample_faculty_data.csv'),
+    'rb').read()
+
+FACULTY_HEADER_FIELDS = FACULTY_SAMPLE_DATA.split(
+    '\n')[0].split(',')
 
 class TestFacultyProcessor(FunctionalTestCase):
@@ -54 +62 @@
         self.site2 = dict(faculties=dict(FAC='pseudo faculty'))
         self.row = dict(code='FAC')
+
         self.processor = FacultyProcessor()
+        self.csv_file_faculty = os.path.join(self.workdir, 'sample_faculty_data.csv')
+        open(self.csv_file_faculty, 'wb').write(FACULTY_SAMPLE_DATA)
         return
 
@@ -144 +155 @@
         self.assertEqual(len(errs),0)
         return
+
+    def test_import(self):
+        num, num_warns, fin_file, fail_file = self.processor.doImport(
+            self.csv_file_faculty, FACULTY_HEADER_FIELDS)
+        content = open(fail_file).read()
+        self.assertEqual(num_warns,5)
+        self.assertEqual(
+            content,
+            'code,local_roles,--ERRORS--\r\n'
+            'CDE,"[{\'user_name\':\'alice\',\'local_role\':\'waeup.local.DepartmentManager\'}]",'
+            'local_roles: alice does not exist\r\n'
+            'DEF,"[{\'user_name\':\'bob\',\'local_role\':\'waeup.local.Boss\'}]",'
+            'local_roles: waeup.local.Boss not allowed\r\n'
+            'EFG,[(\'anything\')],local_roles: no dicts\r\n'
+            'FGH,[,local_roles: Error\r\n'
+            'GHI,"[{\'user\':\'bob\',\'local\':\'waeup.local.DepartmentManager\'}]",'
+            'local_roles: user_name or local_role missing\r\n'
+            )
+        # Bob got a local role in faculty ABC.
+        abc = self.app['faculties']['ABC']
+        role_map = IPrincipalRoleMap(abc)
+        local_role, user_name, setting = role_map.getPrincipalsAndRoles()[0]
+        self.assertEqual(user_name, 'bob')
+        self.assertEqual(local_role, 'waeup.local.DepartmentManager')
+        shutil.rmtree(os.path.dirname(fin_file))
 
 class TestDepartmentProcessors(unittest.TestCase):