Changeset 9209 for main/waeup.kofa/branches/uli-zc-async/src/waeup/kofa/university/tests

Timestamp:

20 Sep 2012, 08:49:37 (12 years ago)

Author:

uli

Message:

Merge changes from update branch (includes trunk changes until r9107).

Location:

main/waeup.kofa/branches/uli-zc-async

Files:

• . (modified) (1 prop)
• src/waeup/kofa/university/tests/sample_certificate_data.csv (copied) (copied from main/waeup.kofa/branches/uli-async-update/src/waeup/kofa/university/tests/sample_certificate_data.csv)
• src/waeup/kofa/university/tests/sample_course_data.csv (copied) (copied from main/waeup.kofa/branches/uli-async-update/src/waeup/kofa/university/tests/sample_course_data.csv)
• src/waeup/kofa/university/tests/sample_department_data.csv (copied) (copied from main/waeup.kofa/branches/uli-async-update/src/waeup/kofa/university/tests/sample_department_data.csv)
• src/waeup/kofa/university/tests/sample_faculty_data.csv (copied) (copied from main/waeup.kofa/branches/uli-async-update/src/waeup/kofa/university/tests/sample_faculty_data.csv)
• src/waeup/kofa/university/tests/sample_faculty_data_update.csv (copied) (copied from main/waeup.kofa/branches/uli-async-update/src/waeup/kofa/university/tests/sample_faculty_data_update.csv)
• src/waeup/kofa/university/tests/test_batching.py (modified) (11 diffs)
• src/waeup/kofa/university/tests/test_export.py (modified) (24 diffs)
• src/waeup/kofa/university/tests/test_university.py (modified) (1 diff)

main/waeup.kofa/branches/uli-zc-async/src/waeup/kofa/university/tests/test_batching.py

@@ -19 +19 @@
 # Tests for university related batching
 import unittest
-
+import tempfile
+import shutil
+import os
+from zope.component.hooks import setSite, clearSite
+from zope.component import createObject
+from zope.securitypolicy.interfaces import (
+    IPrincipalRoleMap, IPrincipalRoleManager)
+from zope.testbrowser.testing import Browser
 from zope.interface.verify import verifyClass, verifyObject
 from waeup.kofa.interfaces import IBatchProcessor
 from waeup.kofa.testing import FunctionalTestCase, FunctionalLayer
+from waeup.kofa.app import University
 from waeup.kofa.university.batching import (
     FacultyProcessor, DepartmentProcessor, CourseProcessor,
@@ -28 +36 @@
 from waeup.kofa.university.certificate import Certificate, CertificateCourse
 from waeup.kofa.university.course import Course
-from waeup.kofa.university.department import Department
-
-
-class TestFacultyProcessor(unittest.TestCase):
+from waeup.kofa.university import Faculty, Department
+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(',')
+
+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(',')
+
+DEPARTMENT_SAMPLE_DATA = open(
+    os.path.join(os.path.dirname(__file__), 'sample_department_data.csv'),
+    'rb').read()
+
+DEPARTMENT_HEADER_FIELDS = DEPARTMENT_SAMPLE_DATA.split(
+    '\n')[0].split(',')
+
+CERTIFICATE_SAMPLE_DATA = open(
+    os.path.join(os.path.dirname(__file__), 'sample_certificate_data.csv'),
+    'rb').read()
+
+CERTIFICATE_HEADER_FIELDS = CERTIFICATE_SAMPLE_DATA.split(
+    '\n')[0].split(',')
+
+COURSE_SAMPLE_DATA = open(
+    os.path.join(os.path.dirname(__file__), 'sample_course_data.csv'),
+    'rb').read()
+
+COURSE_HEADER_FIELDS = COURSE_SAMPLE_DATA.split(
+    '\n')[0].split(',')
+
+class UniversityProcessorSetup(FunctionalTestCase):
+
+    layer = FunctionalLayer
 
     def setUp(self):
+        super(UniversityProcessorSetup, self).setUp()
+        self.dc_root = tempfile.mkdtemp()
+        self.workdir = tempfile.mkdtemp()
+        app = University()
+        self.getRootFolder()['app'] = app
+        self.app = self.getRootFolder()['app']
+        setSite(app)
+
+        self.app['users'].addUser('bob', 'bobssecret')
+        self.app['users'].addUser('anne', 'annessecret')
+
+        # Populate university
+        self.certificate = createObject('waeup.Certificate')
+        self.certificate.code = 'CRT1'
+        self.app['faculties']['FAC1'] = Faculty(code='FAC1')
+        self.app['faculties']['FAC1']['DEP1'] = Department(code='DEP1')
+        self.app['faculties']['FAC1']['DEP1'].certificates.addCertificate(
+            self.certificate)
+        self.course = createObject('waeup.Course')
+        self.course.code = 'CRS1'
+        self.app['faculties']['FAC1']['DEP1'].courses.addCourse(
+            self.course)
+        #self.app['faculties']['fac1']['dep1'].certificates['CERT1'].addCertCourse(
+        #    self.course, level=100)
+        
+        self.logfile = os.path.join(
+            self.app['datacenter'].storage, 'logs', 'main.log')
+        return
+        return
+
+    def tearDown(self):
+        super(UniversityProcessorSetup, self).tearDown()
+        shutil.rmtree(self.workdir)
+        shutil.rmtree(self.dc_root)
+        clearSite()
+        return
+
+class TestFacultyProcessor(UniversityProcessorSetup):
+
+    def setUp(self):
+        super(TestFacultyProcessor, self).setUp()
+
+        self.browser = Browser()
+        self.browser.handleErrors = False
+        self.datacenter_path = 'http://localhost/app/datacenter'
+        self.app['datacenter'].setStoragePath(self.dc_root)
+
         self.proc = FacultyProcessor()
         self.site1 = dict(faculties=dict())
         self.site2 = dict(faculties=dict(FAC='pseudo faculty'))
         self.row = dict(code='FAC')
+
+        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
 
@@ -60 +157 @@
         return
 
-class TestDepartmentProcessors(unittest.TestCase):
+    def test_checkConversion(self):
+        # Make sure we can check conversions.
+        errs, inv_errs, conv_dict = self.proc.checkConversion(
+            dict(faculty_code='ABC', local_roles='[]'))
+        self.assertEqual(len(errs),0)
+
+        errs, inv_errs, conv_dict = self.proc.checkConversion(
+            dict(faculty_code='ABC',
+            local_roles="['nonsense'"
+            ))
+        self.assertEqual(len(errs),1)
+        self.assertEqual(errs, [('local_roles', 'Error')])
+        errs, inv_errs, conv_dict = self.proc.checkConversion(
+            dict(faculty_code='ABC',
+            local_roles="('abc')"
+            ))
+        self.assertEqual(len(errs),1)
+        self.assertEqual(errs, [('local_roles', 'no list')])
+        errs, inv_errs, conv_dict = self.proc.checkConversion(
+            dict(faculty_code='ABC',
+            local_roles="[('ABC')]"
+            ))
+        self.assertEqual(len(errs),1)
+        self.assertEqual(errs, [('local_roles', 'no dicts')])
+        errs, inv_errs, conv_dict = self.proc.checkConversion(
+            dict(faculty_code='ABC',
+            local_roles="('abc')"
+            ))
+        self.assertEqual(len(errs),1)
+        self.assertEqual(errs, [('local_roles', 'no list')])
+        errs, inv_errs, conv_dict = self.proc.checkConversion(
+            dict(faculty_code='ABC', local_roles=
+            "[{'name':'bob','local_role':'waeup.local.DepartmentManager'},]"
+            ))
+        self.assertEqual(len(errs),1)
+        self.assertEqual(errs, [('local_roles', 'user_name or local_role missing')])
+        errs, inv_errs, conv_dict = self.proc.checkConversion(
+            dict(faculty_code='ABC', local_roles=
+            "[{'user_name':'bob','localrole':'waeup.local.DepartmentManager'},]"
+            ))
+        self.assertEqual(len(errs),1)
+        self.assertEqual(errs, [('local_roles', 'user_name or local_role missing')])
+        errs, inv_errs, conv_dict = self.proc.checkConversion(
+            dict(faculty_code='ABC', local_roles=
+            "[{'user_name':'bob','local_role':'waeup.local.Boss'},]"
+            ))
+        self.assertEqual(len(errs),1)
+        self.assertEqual(errs, [('local_roles', 'waeup.local.Boss not allowed')])
+        errs, inv_errs, conv_dict = self.proc.checkConversion(
+            dict(faculty_code='ABC', local_roles=
+            "[{'user_name':'john','local_role':'waeup.local.DepartmentManager'},]"
+            ))
+        self.assertEqual(len(errs),1)
+        self.assertEqual(errs, [('local_roles', 'john does not exist')])
+        errs, inv_errs, conv_dict = self.proc.checkConversion(
+            dict(faculty_code='ABC', local_roles=
+            "[{'user_name':'bob','local_role':'waeup.local.DepartmentManager'},]"
+            ))
+        self.assertEqual(len(errs),0)
+        return
+
+    def test_import(self):
+        num, num_warns, fin_file, fail_file = self.proc.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))
+        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
+        
+    def test_import_update_logging(self):
+        self.app['faculties']['FAC2'] = Faculty(code='FAC2')
+        self.app['faculties']['FAC3'] = Faculty(code='FAC3')
+        self.app['faculties']['FAC4'] = Faculty(code='FAC4')
+        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)
+        logcontent = open(self.logfile).read()
+        # Logging message from updateEntry
+        self.assertTrue(
+            'INFO - system - Department Processor - DEP2 - '
+            'Record updated: local_roles=anne|waeup.local.DepartmentManager, '
+            'code=DEP2'
+            in logcontent)
+
+    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_faculty)
+        ctrl = self.browser.getControl(name='uploadfile:file')
+        file_ctrl = ctrl.mech_control
+        file_ctrl.add_file(file, filename='sample_faculty_data.csv')
+        self.browser.getControl('Upload').click()
+        self.browser.getLink('Process data').click()
+        self.browser.getControl(name="select").click()
+        importerselect = self.browser.getControl(name='importer')
+        importerselect.getControl('Faculty 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 1 rows' in self.browser.contents)
+        # 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_faculty)
+        ctrl = self.browser.getControl(name='uploadfile:file')
+        file_ctrl = ctrl.mech_control
+        file_ctrl.add_file(file, filename='sample_faculty_data.csv')
+        self.browser.getControl('Upload').click()
+        self.assertTrue(
+            'File with same name was uploaded earlier' in self.browser.contents)
+        return
+
+class TestDepartmentProcessor(UniversityProcessorSetup):
 
     def setUp(self):
+        super(TestDepartmentProcessor, self).setUp()
         self.proc = DepartmentProcessor()
         self.site0 = dict()
@@ -69 +354 @@
         self.site3 = dict(faculties=dict(FAC=dict(DPT='pseudo department')))
         self.row = dict(code='DPT', faculty_code='FAC')
+
+        self.csv_file_department = os.path.join(self.workdir, 'sample_department_data.csv')
+        open(self.csv_file_department, 'wb').write(DEPARTMENT_SAMPLE_DATA)
         return
 
@@ -115 +403 @@
         return
 
-class CourseProcessorTests(FunctionalTestCase):
-
-    layer = FunctionalLayer
+    def test_import(self):
+        num, num_warns, fin_file, fail_file = self.proc.doImport(
+            self.csv_file_department, DEPARTMENT_HEADER_FIELDS)
+        content = open(fail_file).read()
+        self.assertEqual(num_warns,6)
+        self.assertEqual(
+            content,
+            'faculty_code,code,local_roles,--ERRORS--\r\n'
+            'FAC1,DEP2,"[{\'user_name\':\'alice\',\'local_role\':\'waeup.local.DepartmentManager\'}]",'
+            'local_roles: alice does not exist\r\n'
+            'FAC1,DEP2,"[{\'user_name\':\'anne\',\'local_role\':\'waeup.local.Boss\'}]",'
+            'local_roles: waeup.local.Boss not allowed\r\n'
+            'FAC1,DEP2,[(\'anything\')],local_roles: no dicts\r\n'
+            'FAC1,DEP2,[,local_roles: Error\r\n'
+            'FAC1,DEP2,"[{\'user\':\'anne\',\'local\':\'waeup.local.DepartmentManager\'}]",'
+            'local_roles: user_name or local_role missing\r\n'
+            'FAC11,DEP2,"[{\'user_name\':\'anne\',\'local_role\':\'waeup.local.DepartmentManager\'}]",'
+            'Not all parents do exist yet. Skipping\r\n'
+            )
+        # Anne got a local role in department DEP2.
+        dep = self.app['faculties']['FAC1']['DEP2']
+        role_map = IPrincipalRoleMap(dep)
+        local_role, user_name, setting = role_map.getPrincipalsAndRoles()[0]
+        self.assertEqual(user_name, 'anne')
+        self.assertEqual(local_role, 'waeup.local.DepartmentManager')
+        shutil.rmtree(os.path.dirname(fin_file))
+        return
+
+class CourseProcessorTests(UniversityProcessorSetup):
 
     def setUp(self):
         super(CourseProcessorTests, self).setUp()
         self.proc = CourseProcessor()
-        self.site0 = dict()
-        self.site1 = dict(faculties=dict())
-        self.site2 = dict(faculties=dict(FAC=dict()))
-        self.department1 = Department(code='DPT')
-        self.department2 = Department(code='DPT')
-        self.course = Course(code='CRS')
-        self.department2.courses['CRS'] = self.course
-        self.site3 = dict(faculties=dict(FAC=dict(DPT=self.department1)))
-        self.site4 = dict(faculties=dict(FAC=dict(DPT=self.department2)))
-        self.row = dict(department_code='DPT', faculty_code='FAC', code="CRS")
+        self.row1 = dict(department_code='DEP2', faculty_code='FAC1', code="CRS1")
+        self.row2 = dict(department_code='DEP1', faculty_code='FAC2', code="CRS1")
+        self.row3 = dict(department_code='DEP1', faculty_code='FAC1', code="CRS2")
+        self.row4 = dict(department_code='DEP1', faculty_code='FAC1', code="CRS1")
+
+        self.csv_file_course = os.path.join(self.workdir, 'sample_course_data.csv')
+        open(self.csv_file_course, 'wb').write(COURSE_SAMPLE_DATA)
         return
 
@@ -142 +453 @@
     def test_parents_exist(self):
         # make sure we lookup parents correctly
-        result0 = self.proc.parentsExist(self.row, self.site0)
-        result1 = self.proc.parentsExist(self.row, self.site1)
-        result2 = self.proc.parentsExist(self.row, self.site2)
-        result3 = self.proc.parentsExist(self.row, self.site3)
-        result4 = self.proc.parentsExist(self.row, self.site4)
-        self.assertTrue(result0 is False)
+        result1 = self.proc.parentsExist(self.row1, self.app)
+        result2 = self.proc.parentsExist(self.row2, self.app)
+        result3 = self.proc.parentsExist(self.row3, self.app)
         self.assertTrue(result1 is False)
         self.assertTrue(result2 is False)
         self.assertTrue(result3 is True)
-        self.assertTrue(result4 is True)
         return
 
     def test_entry_exists(self):
         # make sure we find an entry if it exists
-        result0 = self.proc.entryExists(self.row, self.site0)
-        result1 = self.proc.entryExists(self.row, self.site1)
-        result2 = self.proc.entryExists(self.row, self.site2)
-        result3 = self.proc.entryExists(self.row, self.site3)
-        result4 = self.proc.entryExists(self.row, self.site4)
-        self.assertTrue(result0 is False)
+        result1 = self.proc.entryExists(self.row1, self.app)
+        result2 = self.proc.entryExists(self.row2, self.app)
+        result3 = self.proc.entryExists(self.row3, self.app)
+        result4 = self.proc.entryExists(self.row4, self.app)
         self.assertTrue(result1 is False)
         self.assertTrue(result2 is False)
@@ -170 +475 @@
     def test_get_entry(self):
         # make sure we can get an entry if it exists
-        result1 = self.proc.getEntry(self.row, self.site3)
-        result2 = self.proc.getEntry(self.row, self.site4)
+        result1 = self.proc.getEntry(self.row1, self.app)
+        result2 = self.proc.getEntry(self.row4, self.app)
         self.assertTrue(result1 is None)
         self.assertTrue(result2 is self.course)
@@ -178 +483 @@
     def test_del_entry(self):
         # make sure we can delete entries
-        self.assertTrue('CRS' in self.department2.courses.keys())
-        self.proc.delEntry(self.row, self.site4)
-        self.assertTrue('CRS' not in self.department2.courses.keys())
-        return
-
-class CertificateProcessorTests(FunctionalTestCase):
-
-    layer = FunctionalLayer
+        self.assertTrue('CRS1' in self.app['faculties']['FAC1']['DEP1'].courses.keys())
+        self.proc.delEntry(self.row4, self.app)
+        self.assertTrue('CRS1' not in self.app['faculties']['FAC1']['DEP1'].courses.keys())
+        return
+
+    def test_import(self):
+        num, num_warns, fin_file, fail_file = self.proc.doImport(
+            self.csv_file_course, COURSE_HEADER_FIELDS)
+        content = open(fail_file).read()
+        self.assertEqual(num_warns,6)
+        self.assertEqual(
+            content,
+            'faculty_code,department_code,code,local_roles,--ERRORS--\r\n'
+            'FAC1,DEP1,CRS2,"[{\'user_name\':\'alice\',\'local_role\':\'waeup.local.Lecturer\'}]",'
+            'local_roles: alice does not exist\r\n'
+            'FAC1,DEP1,CRS2,"[{\'user_name\':\'anne\',\'local_role\':\'waeup.local.Boss\'}]",'
+            'local_roles: waeup.local.Boss not allowed\r\n'
+            'FAC1,DEP1,CRS2,[(\'anything\')],local_roles: no dicts\r\n'
+            'FAC1,DEP1,CRS2,[,local_roles: Error\r\n'
+            'FAC1,DEP1,CRS2,"[{\'user\':\'anne\',\'local\':\'waeup.local.Lecturer\'}]",'
+            'local_roles: user_name or local_role missing\r\n'
+            'FAC11,DEP2,CRS2,"[{\'user_name\':\'anne\',\'local_role\':\'waeup.local.Lecturer\'}]",'
+            'Not all parents do exist yet. Skipping\r\n'
+            )
+        # Anne got a local role in course CRS2.
+        dep = self.app['faculties']['FAC1']['DEP1'].courses['CRS2']
+        role_map = IPrincipalRoleMap(dep)
+        local_role, user_name, setting = role_map.getPrincipalsAndRoles()[0]
+        self.assertEqual(user_name, 'anne')
+        self.assertEqual(local_role, 'waeup.local.Lecturer')
+        shutil.rmtree(os.path.dirname(fin_file))
+        return
+
+
+class CertificateProcessorTests(UniversityProcessorSetup):
 
     def setUp(self):
         super(CertificateProcessorTests, self).setUp()
         self.proc = CertificateProcessor()
-        self.site0 = dict()
-        self.site1 = dict(faculties=dict())
-        self.site2 = dict(faculties=dict(FAC=dict()))
-        self.department1 = Department(code='DPT')
-        self.department2 = Department(code='DPT')
-        self.certificate = Certificate(code='CRT')
-        self.department2.certificates['CRT'] = self.certificate
-        self.site3 = dict(faculties=dict(FAC=dict(DPT=self.department1)))
-        self.site4 = dict(faculties=dict(FAC=dict(DPT=self.department2)))
-        self.row = dict(department_code='DPT', faculty_code='FAC', code="CRT")
+        self.row1 = dict(department_code='DEP2', faculty_code='FAC1', code="CRT1")
+        self.row2 = dict(department_code='DEP1', faculty_code='FAC2', code="CRT1")
+        self.row3 = dict(department_code='DEP1', faculty_code='FAC1', code="CRT2")
+        self.row4 = dict(department_code='DEP1', faculty_code='FAC1', code="CRT1")
+
+        self.csv_file_certificate = os.path.join(self.workdir, 'sample_certificate_data.csv')
+        open(self.csv_file_certificate, 'wb').write(CERTIFICATE_SAMPLE_DATA)
         return
 
     def test_ifaces(self):
         # Make sure we fullfill all interface contracts
-        verifyClass(IBatchProcessor, CertificateProcessor)
+        verifyClass(IBatchProcessor, CourseProcessor)
         verifyObject(IBatchProcessor, self.proc)
         return
 
     def test_parents_exist(self):
-        # make sure we can find all certificate parents
-        result0 = self.proc.parentsExist(self.row, self.site0)
-        result1 = self.proc.parentsExist(self.row, self.site1)
-        result2 = self.proc.parentsExist(self.row, self.site2)
-        result3 = self.proc.parentsExist(self.row, self.site3)
-        result4 = self.proc.parentsExist(self.row, self.site4)
-        self.assertTrue(result0 is False)
+        # make sure we lookup parents correctly
+        result1 = self.proc.parentsExist(self.row1, self.app)
+        result2 = self.proc.parentsExist(self.row2, self.app)
+        result3 = self.proc.parentsExist(self.row3, self.app)
         self.assertTrue(result1 is False)
         self.assertTrue(result2 is False)
         self.assertTrue(result3 is True)
-        self.assertTrue(result4 is True)
         return
 
     def test_entry_exists(self):
         # make sure we find an entry if it exists
-        result0 = self.proc.entryExists(self.row, self.site0)
-        result1 = self.proc.entryExists(self.row, self.site1)
-        result2 = self.proc.entryExists(self.row, self.site2)
-        result3 = self.proc.entryExists(self.row, self.site3)
-        result4 = self.proc.entryExists(self.row, self.site4)
-        self.assertTrue(result0 is False)
+        result1 = self.proc.entryExists(self.row1, self.app)
+        result2 = self.proc.entryExists(self.row2, self.app)
+        result3 = self.proc.entryExists(self.row3, self.app)
+        result4 = self.proc.entryExists(self.row4, self.app)
         self.assertTrue(result1 is False)
         self.assertTrue(result2 is False)
@@ -238 +561 @@
     def test_get_entry(self):
         # make sure we can get an entry if it exists
-        result1 = self.proc.getEntry(self.row, self.site3)
-        result2 = self.proc.getEntry(self.row, self.site4)
+        result1 = self.proc.getEntry(self.row1, self.app)
+        result2 = self.proc.getEntry(self.row4, self.app)
         self.assertTrue(result1 is None)
         self.assertTrue(result2 is self.certificate)
@@ -246 +569 @@
     def test_del_entry(self):
         # make sure we can delete entries
-        self.assertTrue('CRT' in self.department2.certificates.keys())
-        self.proc.delEntry(self.row, self.site4)
-        self.assertTrue('CRT' not in self.department2.certificates.keys())
-        return
+        self.assertTrue('CRT1' in self.app['faculties']['FAC1']['DEP1'].certificates.keys())
+        self.proc.delEntry(self.row4, self.app)
+        self.assertTrue('CRT1' not in self.app['faculties']['FAC1']['DEP1'].certificates.keys())
+        return
+
+    def test_import(self):
+        num, num_warns, fin_file, fail_file = self.proc.doImport(
+            self.csv_file_certificate, CERTIFICATE_HEADER_FIELDS)
+        content = open(fail_file).read()
+        self.assertEqual(num_warns,6)
+        self.assertEqual(
+            content,
+            'faculty_code,department_code,code,local_roles,--ERRORS--\r\n'
+            'FAC1,DEP1,CRT2,"[{\'user_name\':\'alice\',\'local_role\':\'waeup.local.CourseAdviser100\'}]",'
+            'local_roles: alice does not exist\r\n'
+            'FAC1,DEP1,CRT2,"[{\'user_name\':\'anne\',\'local_role\':\'waeup.local.Boss\'}]",'
+            'local_roles: waeup.local.Boss not allowed\r\n'
+            'FAC1,DEP1,CRT2,[(\'anything\')],local_roles: no dicts\r\n'
+            'FAC1,DEP1,CRT2,[,local_roles: Error\r\n'
+            'FAC1,DEP1,CRT2,"[{\'user\':\'anne\',\'local\':\'waeup.local.CourseAdviser100\'}]",'
+            'local_roles: user_name or local_role missing\r\n'
+            'FAC11,DEP2,CRT2,"[{\'user_name\':\'anne\',\'local_role\':\'waeup.local.CourseAdviser100\'}]",'
+            'Not all parents do exist yet. Skipping\r\n'
+            )
+        # Anne got a local role in certificate CRT2.
+        dep = self.app['faculties']['FAC1']['DEP1'].certificates['CRT2']
+        role_map = IPrincipalRoleMap(dep)
+        local_role, user_name, setting = role_map.getPrincipalsAndRoles()[0]
+        self.assertEqual(user_name, 'anne')
+        self.assertEqual(local_role, 'waeup.local.CourseAdviser100')
+        shutil.rmtree(os.path.dirname(fin_file))
+        
+        logcontent = open(self.logfile).read()
+        # Logging message from updateEntry
+        self.assertTrue(
+            'INFO - system - Certificate Processor - CRT2 - Record updated: '
+            'local_roles=anne|waeup.local.CourseAdviser100, code=CRT2'
+            in logcontent)        
+        
+        return
+
 
 class CertCourseProcessorTests(FunctionalTestCase):
@@ -266 +626 @@
         self.department2.certificates['CRT'] = self.certificate
         self.course = Course(code='CRS')
-        self.department2.certificates['CRT'].addCourseRef(self.course)
+        self.department2.certificates['CRT'].addCertCourse(self.course)
         self.cert_course = self.certificate['CRS_100']
         self.department2.courses['CRS'] = self.course

main/waeup.kofa/branches/uli-zc-async/src/waeup/kofa/university/tests/test_export.py

@@ -22 +22 @@
 from zope.component import queryUtility
 from zope.interface.verify import verifyObject, verifyClass
+from zope.securitypolicy.interfaces import IPrincipalRoleManager
 from waeup.kofa.interfaces import ICSVExporter
-from waeup.kofa.testing import KofaUnitTestLayer
+from waeup.kofa.testing import KofaUnitTestLayer, FunctionalLayer
 from waeup.kofa.university import (
     FacultiesContainer, Faculty, Department, Course, Certificate,
@@ -34 +35 @@
 class FacultyExporterTest(unittest.TestCase):
 
-    layer = KofaUnitTestLayer
+    layer = FunctionalLayer
 
     def setUp(self):
@@ -66 +67 @@
         self.assertEqual(
             result,
-            'code,title,title_prefix\r\n'
-            'F1,Faculty of Cheese,faculty\r\n'
+            'code,title,title_prefix,users_with_local_roles\r\n'
+            'F1,Faculty of Cheese,faculty,[]\r\n'
             )
         return
@@ -78 +79 @@
         self.assertEqual(
             result,
-            'code,title,title_prefix\r\n'
-            'F1,Faculty of Cheese,faculty\r\n'
+            'code,title,title_prefix,users_with_local_roles\r\n'
+            'F1,Faculty of Cheese,faculty,[]\r\n'
             )
         return
@@ -96 +97 @@
         self.assertEqual(
             result,
-            'code,title,title_prefix\r\n'
-            'F1,Faculty of Cheese,faculty\r\n'
-            'F2,Centre of Onion,centre\r\n'
+            'code,title,title_prefix,users_with_local_roles\r\n'
+            'F1,Faculty of Cheese,faculty,[]\r\n'
+            'F2,Centre of Onion,centre,[]\r\n'
             )
         return
@@ -114 +115 @@
         self.assertEqual(
             result,
-            'code,title,title_prefix\r\n'
-            'F1,Faculty of Cheese,faculty\r\n'
-            'F2,Centre of Onion,centre\r\n'
+            'code,title,title_prefix,users_with_local_roles\r\n'
+            'F1,Faculty of Cheese,faculty,[]\r\n'
+            'F2,Centre of Onion,centre,[]\r\n'
             )
         return
@@ -123 +124 @@
     # Tests for DepartmentExporter
 
-    layer = KofaUnitTestLayer
+    layer = FunctionalLayer
 
     def setUp(self):
@@ -141 +142 @@
         self.fac1.addDepartment(self.dept2)
         self.fac2.addDepartment(self.dept3)
+        role_manager = IPrincipalRoleManager(self.dept1)
+        role_manager.assignRoleToPrincipal(u'bobsrole', u'bob')
         return
 
@@ -167 +170 @@
         self.assertEqual(
             result,
-            'code,faculty_code,title,title_prefix\r\n'
-            'D1,F1,Department of Cheddar,department\r\n'
+            'code,faculty_code,title,title_prefix,users_with_local_roles\r\n'
+            'D1,F1,Department of Cheddar,department,'
+            '"[{\'user_name\': u\'bob\', \'local_role\': u\'bobsrole\'}]"\r\n'
             )
         return
@@ -178 +182 @@
         self.assertEqual(
             result,
-            'code,faculty_code,title,title_prefix\r\n'
-            'D1,F1,Department of Cheddar,department\r\n'
-            'D2,F1,Institue of Gouda,institute\r\n'
+            'code,faculty_code,title,title_prefix,users_with_local_roles\r\n'
+            'D1,F1,Department of Cheddar,department,'
+            '"[{\'user_name\': u\'bob\', \'local_role\': u\'bobsrole\'}]"\r\n'
+            'D2,F1,Institue of Gouda,institute,[]\r\n'
             )
         return
@@ -191 +196 @@
         self.assertEqual(
             result,
-            'code,faculty_code,title,title_prefix\r\n'
-            'D1,F1,Department of Cheddar,department\r\n'
-            'D2,F1,Institue of Gouda,institute\r\n'
-            'D3,F2,Department of Rings,department\r\n'
+            'code,faculty_code,title,title_prefix,users_with_local_roles\r\n'
+            'D1,F1,Department of Cheddar,department,'
+            '"[{\'user_name\': u\'bob\', \'local_role\': u\'bobsrole\'}]"\r\n'
+            'D2,F1,Institue of Gouda,institute,[]\r\n'
+            'D3,F2,Department of Rings,department,[]\r\n'
             )
         return
@@ -204 +210 @@
         self.assertEqual(
             result,
-            'code,faculty_code,title,title_prefix\r\n'
-            'D1,F1,Department of Cheddar,department\r\n'
-            'D2,F1,Institue of Gouda,institute\r\n'
-            'D3,F2,Department of Rings,department\r\n'
+            'code,faculty_code,title,title_prefix,users_with_local_roles\r\n'
+            'D1,F1,Department of Cheddar,department,'
+            '"[{\'user_name\': u\'bob\', \'local_role\': u\'bobsrole\'}]"\r\n'
+            'D2,F1,Institue of Gouda,institute,[]\r\n'
+            'D3,F2,Department of Rings,department,[]\r\n'
             )
         return
@@ -214 +221 @@
     # Tests for CourseExporter
 
-    layer = KofaUnitTestLayer
+    layer = FunctionalLayer
 
     def setUp(self):
@@ -260 +267 @@
         self.assertEqual(
             result,
-            'code,faculty_code,department_code,title,credits,passmark,semester\r\n'
-            'C1,F1,D1,Cheese Basics,0,40,1\r\n'
+            'code,faculty_code,department_code,title,credits,'
+            'passmark,semester,users_with_local_roles\r\n'
+            'C1,F1,D1,Cheese Basics,0,40,1,[]\r\n'
             )
         return
@@ -271 +279 @@
         self.assertEqual(
             result,
-            'code,faculty_code,department_code,title,credits,passmark,semester\r\n'
-            'C1,F1,D1,Cheese Basics,0,40,1\r\n'
-            'C2,F1,D1,Advanced Cheese Making,0,40,1\r\n'
+            'code,faculty_code,department_code,title,credits,passmark,'
+            'semester,users_with_local_roles\r\n'
+            'C1,F1,D1,Cheese Basics,0,40,1,[]\r\n'
+            'C2,F1,D1,Advanced Cheese Making,0,40,1,[]\r\n'
             )
         return
@@ -284 +293 @@
         self.assertEqual(
             result,
-            'code,faculty_code,department_code,title,credits,passmark,semester\r\n'
-            'C1,F1,D1,Cheese Basics,0,40,1\r\n'
-            'C2,F1,D1,Advanced Cheese Making,0,40,1\r\n'
-            'C3,F1,D2,Selling Cheese,0,40,1\r\n'
+            'code,faculty_code,department_code,title,credits,passmark,'
+            'semester,users_with_local_roles\r\n'
+            'C1,F1,D1,Cheese Basics,0,40,1,[]\r\n'
+            'C2,F1,D1,Advanced Cheese Making,0,40,1,[]\r\n'
+            'C3,F1,D2,Selling Cheese,0,40,1,[]\r\n'
             )
         return
@@ -297 +307 @@
         self.assertEqual(
             result,
-            'code,faculty_code,department_code,title,credits,passmark,semester\r\n'
-            'C1,F1,D1,Cheese Basics,0,40,1\r\n'
-            'C2,F1,D1,Advanced Cheese Making,0,40,1\r\n'
-            'C3,F1,D2,Selling Cheese,0,40,1\r\n'
+            'code,faculty_code,department_code,title,credits,passmark,'
+            'semester,users_with_local_roles\r\n'
+            'C1,F1,D1,Cheese Basics,0,40,1,[]\r\n'
+            'C2,F1,D1,Advanced Cheese Making,0,40,1,[]\r\n'
+            'C3,F1,D2,Selling Cheese,0,40,1,[]\r\n'
             )
         return
@@ -307 +318 @@
     # Tests for CertificateExporter
 
-    layer = KofaUnitTestLayer
+    layer = FunctionalLayer
 
     def setUp(self):
@@ -339 +350 @@
         self.dept1.certificates.addCertificate(self.cert2)
         self.dept2.certificates.addCertificate(self.cert3)
+        role_manager = IPrincipalRoleManager(self.cert1)
+        role_manager.assignRoleToPrincipal(u'bobsrole', u'bob')
         return
 
@@ -365 +378 @@
         self.assertEqual(
             result,
-            'code,faculty_code,department_code,title,study_mode,start_level,end_level,application_category,school_fee_1,school_fee_2\r\n'
-            'CERT1,F1,D1,Master of Cheese,ct_ft,100,300,basic,,\r\n'
+            'code,faculty_code,department_code,title,study_mode,start_level,'
+            'end_level,application_category,school_fee_1,'
+            'school_fee_2,school_fee_3,school_fee_4,users_with_local_roles\r\n'
+            'CERT1,F1,D1,Master of Cheese,ct_ft,100,300,basic,,,,,'
+            '"[{\'user_name\': u\'bob\', \'local_role\': u\'bobsrole\'}]"\r\n'
             )
         return
@@ -376 +392 @@
         self.assertEqual(
             result,
-            'code,faculty_code,department_code,title,study_mode,start_level,end_level,application_category,school_fee_1,school_fee_2\r\n'
-            'CERT1,F1,D1,Master of Cheese,ct_ft,100,300,basic,,\r\n'
-            'CERT2,F1,D1,Master of Cheddar,ct_ft,400,700,cest,,\r\n'
+            'code,faculty_code,department_code,title,study_mode,start_level,'
+            'end_level,application_category,school_fee_1,'
+            'school_fee_2,school_fee_3,school_fee_4,users_with_local_roles\r\n'
+            'CERT1,F1,D1,Master of Cheese,ct_ft,100,300,basic,,,,,'
+            '"[{\'user_name\': u\'bob\', \'local_role\': u\'bobsrole\'}]"\r\n'
+            'CERT2,F1,D1,Master of Cheddar,ct_ft,400,700,cest,,,,,[]\r\n'
             )
         return
@@ -389 +408 @@
         self.assertEqual(
             result,
-            'code,faculty_code,department_code,title,study_mode,start_level,end_level,application_category,school_fee_1,school_fee_2\r\n'
-            'CERT1,F1,D1,Master of Cheese,ct_ft,100,300,basic,,\r\n'
-            'CERT2,F1,D1,Master of Cheddar,ct_ft,400,700,cest,,\r\n'
-            'CERT3,F1,D2,Cert. of Rubbish,dp_pt,100,200,no,,\r\n'
+            'code,faculty_code,department_code,title,study_mode,start_level,'
+            'end_level,application_category,'
+            'school_fee_1,school_fee_2,school_fee_3,school_fee_4,'
+            'users_with_local_roles\r\n'
+            'CERT1,F1,D1,Master of Cheese,ct_ft,100,300,basic,,,,,'
+            '"[{\'user_name\': u\'bob\', \'local_role\': u\'bobsrole\'}]"\r\n'
+            'CERT2,F1,D1,Master of Cheddar,ct_ft,400,700,cest,,,,,[]\r\n'
+            'CERT3,F1,D2,Cert. of Rubbish,dp_pt,100,200,no,,,,,[]\r\n'
             )
         return
@@ -402 +425 @@
         self.assertEqual(
             result,
-            'code,faculty_code,department_code,title,study_mode,start_level,end_level,application_category,school_fee_1,school_fee_2\r\n'
-            'CERT1,F1,D1,Master of Cheese,ct_ft,100,300,basic,,\r\n'
-            'CERT2,F1,D1,Master of Cheddar,ct_ft,400,700,cest,,\r\n'
-            'CERT3,F1,D2,Cert. of Rubbish,dp_pt,100,200,no,,\r\n'
+            'code,faculty_code,department_code,title,study_mode,start_level,'
+            'end_level,application_category,'
+            'school_fee_1,school_fee_2,school_fee_3,school_fee_4,'
+            'users_with_local_roles\r\n'
+            'CERT1,F1,D1,Master of Cheese,ct_ft,100,300,basic,,,,,'
+            '"[{\'user_name\': u\'bob\', \'local_role\': u\'bobsrole\'}]"\r\n'
+            'CERT2,F1,D1,Master of Cheddar,ct_ft,400,700,cest,,,,,[]\r\n'
+            'CERT3,F1,D2,Cert. of Rubbish,dp_pt,100,200,no,,,,,[]\r\n'
             )
         return
@@ -444 +471 @@
         self.dept1.certificates.addCertificate(self.cert2)
         self.dept2.certificates.addCertificate(self.cert3)
-        self.cert1.addCourseRef(self.course1, 100, True)
-        self.cert1.addCourseRef(self.course2, 400, False)
-        self.cert3.addCourseRef(self.course3, 100, False)
+        self.cert1.addCertCourse(self.course1, 100, True)
+        self.cert1.addCertCourse(self.course2, 400, False)
+        self.cert3.addCertCourse(self.course3, 100, False)
         self.certcourse1 = self.cert1['C1_100']
         self.certcourse2 = self.cert1['C2_400']

main/waeup.kofa/branches/uli-zc-async/src/waeup/kofa/university/tests/test_university.py

@@ -33 +33 @@
 def test_suite():
     # collect doctests for university subpackage
-    suite = get_doctest_suite(['university/courserefs.txt',])
+    suite = get_doctest_suite(['university/certcourses.txt',])
     # add local unittests (actually only one)
     for testcase in [UniversitySubpackageTests,]: