source: main/waeup.kofa/trunk/src/waeup/kofa/university/tests/test_batching.py @ 13066

Last change on this file since 13066 was 12868, checked in by Henrik Bettermann, 10 years ago

More docs.
  • Property svn:keywords set to Id
File size: 31.6 KB
Line 
1## $Id: test_batching.py 12868 2015-04-21 20:56:58Z henrik $
2##
3## Copyright (C) 2011 Uli Fouquet & Henrik Bettermann
4## This program is free software; you can redistribute it and/or modify
5## it under the terms of the GNU General Public License as published by
6## the Free Software Foundation; either version 2 of the License, or
7## (at your option) any later version.
8##
9## This program is distributed in the hope that it will be useful,
10## but WITHOUT ANY WARRANTY; without even the implied warranty of
11## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
12## GNU General Public License for more details.
13##
14## You should have received a copy of the GNU General Public License
15## along with this program; if not, write to the Free Software
16## Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17##
18
19# Tests for university related batching
20import unittest
21import tempfile
22import shutil
23import os
24from zope.component.hooks import setSite, clearSite
25from zope.component import createObject
26from zope.securitypolicy.interfaces import (
27    IPrincipalRoleMap, IPrincipalRoleManager)
28from zope.testbrowser.testing import Browser
29from zope.interface.verify import verifyClass, verifyObject
30from waeup.kofa.interfaces import IBatchProcessor
31from waeup.kofa.testing import FunctionalTestCase, FunctionalLayer
32from waeup.kofa.app import University
33from waeup.kofa.university.batching import (
34    FacultyProcessor, DepartmentProcessor, CourseProcessor,
35    CertificateProcessor, CertificateCourseProcessor)
36from waeup.kofa.university.certificate import Certificate, CertificateCourse
37from waeup.kofa.university.course import Course
38from waeup.kofa.university import Faculty, Department
39from waeup.kofa.university.batching import FacultyProcessor
40
41FACULTY_SAMPLE_DATA = open(
42    os.path.join(os.path.dirname(__file__), 'sample_faculty_data.csv'),
43    'rb').read()
44
45FACULTY_HEADER_FIELDS = FACULTY_SAMPLE_DATA.split(
46    '\n')[0].split(',')
47
48FACULTY_SAMPLE_DATA_UPDATE = open(
49    os.path.join(os.path.dirname(__file__), 'sample_faculty_data_update.csv'),
50    'rb').read()
51
52FACULTY_HEADER_FIELDS_UPDATE = FACULTY_SAMPLE_DATA_UPDATE.split(
53    '\n')[0].split(',')
54
55DEPARTMENT_SAMPLE_DATA = open(
56    os.path.join(os.path.dirname(__file__), 'sample_department_data.csv'),
57    'rb').read()
58
59DEPARTMENT_HEADER_FIELDS = DEPARTMENT_SAMPLE_DATA.split(
60    '\n')[0].split(',')
61
62CERTIFICATE_SAMPLE_DATA = open(
63    os.path.join(os.path.dirname(__file__), 'sample_certificate_data.csv'),
64    'rb').read()
65
66CERTIFICATE_HEADER_FIELDS = CERTIFICATE_SAMPLE_DATA.split(
67    '\n')[0].split(',')
68
69COURSE_SAMPLE_DATA = open(
70    os.path.join(os.path.dirname(__file__), 'sample_course_data.csv'),
71    'rb').read()
72
73COURSE_HEADER_FIELDS = COURSE_SAMPLE_DATA.split(
74    '\n')[0].split(',')
75
76COURSE_SAMPLE_DATA_UPDATE = open(
77    os.path.join(os.path.dirname(__file__), 'sample_course_data_update.csv'),
78    'rb').read()
79
80COURSE_HEADER_FIELDS_UPDATE = COURSE_SAMPLE_DATA_UPDATE.split(
81    '\n')[0].split(',')
82
83
84CERTIFICATECOURSE_SAMPLE_DATA = open(
85    os.path.join(os.path.dirname(__file__), 'sample_certificatecourse_data.csv'),
86    'rb').read()
87
88CERTIFICATECOURSE_HEADER_FIELDS = CERTIFICATECOURSE_SAMPLE_DATA.split(
89    '\n')[0].split(',')
90
91class UniversityProcessorSetup(FunctionalTestCase):
92
93    layer = FunctionalLayer
94
95    def setUp(self):
96        super(UniversityProcessorSetup, self).setUp()
97        self.dc_root = tempfile.mkdtemp()
98        self.workdir = tempfile.mkdtemp()
99        app = University()
100        app['datacenter'].setStoragePath(self.dc_root)
101        self.getRootFolder()['app'] = app
102        self.app = self.getRootFolder()['app']
103        setSite(app)
104
105        self.app['users'].addUser('bob', 'bobssecret')
106        self.app['users'].addUser('anne', 'annessecret')
107
108        # Populate university
109        self.certificate = createObject('waeup.Certificate')
110        self.certificate.code = 'CRT1'
111        self.app['faculties']['FAC1'] = Faculty(code='FAC1')
112        self.app['faculties']['FAC1']['DEP1'] = Department(code='DEP1')
113        self.app['faculties']['FAC1']['DEP1'].certificates.addCertificate(
114            self.certificate)
115        self.course = createObject('waeup.Course')
116        self.course.code = 'CRS1'
117        self.app['faculties']['FAC1']['DEP1'].courses.addCourse(
118            self.course)
119        #self.app['faculties']['fac1']['dep1'].certificates['CERT1'].addCertCourse(
120        #    self.course, level=100)
121       
122        self.logfile = os.path.join(
123            self.app['datacenter'].storage, 'logs', 'main.log')
124        return
125
126    def tearDown(self):
127        super(UniversityProcessorSetup, self).tearDown()
128        shutil.rmtree(self.workdir)
129        shutil.rmtree(self.dc_root)
130        clearSite()
131        return
132
133class TestFacultyProcessor(UniversityProcessorSetup):
134
135    def setUp(self):
136        super(TestFacultyProcessor, self).setUp()
137
138        self.browser = Browser()
139        self.browser.handleErrors = False
140        self.datacenter_path = 'http://localhost/app/datacenter'
141        self.app['datacenter'].setStoragePath(self.dc_root)
142
143        self.proc = FacultyProcessor()
144        self.site1 = dict(faculties=dict())
145        self.site2 = dict(faculties=dict(FAC='pseudo faculty'))
146        self.row = dict(code='FAC')
147
148        self.csv_file_faculty = os.path.join(self.workdir, 'sample_faculty_data.csv')
149        open(self.csv_file_faculty, 'wb').write(FACULTY_SAMPLE_DATA)
150        self.csv_file_faculty_update = os.path.join(self.workdir, 'sample_faculty_data_update.csv')
151        open(self.csv_file_faculty_update, 'wb').write(FACULTY_SAMPLE_DATA_UPDATE)
152        return
153
154    def test_ifaces(self):
155        # Make sure we fullfill all interface contracts
156        verifyClass(IBatchProcessor, FacultyProcessor)
157        verifyObject(IBatchProcessor, self.proc)
158        return
159
160    def test_get_entry(self):
161        # if a faculty exists already, we will get it
162        result1 = self.proc.getEntry(self.row, self.site1)
163        result2 = self.proc.getEntry(self.row, self.site2)
164        self.assertTrue(result1 is None)
165        self.assertEqual(result2, 'pseudo faculty')
166        return
167
168    def test_del_entry(self):
169        # make sure we can del entries.
170        self.proc.delEntry(self.row, self.site2)
171        self.assertTrue('FAC' not in self.site2.keys())
172        return
173
174    def test_checkConversion(self):
175        # Make sure we can check conversions.
176        errs, inv_errs, conv_dict = self.proc.checkConversion(
177            dict(faculty_code='ABC', local_roles='[]'))
178        self.assertEqual(len(errs),0)
179
180        errs, inv_errs, conv_dict = self.proc.checkConversion(
181            dict(faculty_code='ABC',
182            local_roles="['nonsense'"
183            ))
184        self.assertEqual(len(errs),1)
185        self.assertEqual(errs, [('local_roles', 'Error')])
186        errs, inv_errs, conv_dict = self.proc.checkConversion(
187            dict(faculty_code='ABC',
188            local_roles="('abc')"
189            ))
190        self.assertEqual(len(errs),1)
191        self.assertEqual(errs, [('local_roles', 'no list')])
192        errs, inv_errs, conv_dict = self.proc.checkConversion(
193            dict(faculty_code='ABC',
194            local_roles="[('ABC')]"
195            ))
196        self.assertEqual(len(errs),1)
197        self.assertEqual(errs, [('local_roles', 'no dicts')])
198        errs, inv_errs, conv_dict = self.proc.checkConversion(
199            dict(faculty_code='ABC',
200            local_roles="('abc')"
201            ))
202        self.assertEqual(len(errs),1)
203        self.assertEqual(errs, [('local_roles', 'no list')])
204        errs, inv_errs, conv_dict = self.proc.checkConversion(
205            dict(faculty_code='ABC', local_roles=
206            "[{'name':'bob','local_role':'waeup.local.DepartmentManager'},]"
207            ))
208        self.assertEqual(len(errs),1)
209        self.assertEqual(errs, [('local_roles', 'user_name or local_role missing')])
210        errs, inv_errs, conv_dict = self.proc.checkConversion(
211            dict(faculty_code='ABC', local_roles=
212            "[{'user_name':'bob','localrole':'waeup.local.DepartmentManager'},]"
213            ))
214        self.assertEqual(len(errs),1)
215        self.assertEqual(errs, [('local_roles', 'user_name or local_role missing')])
216        errs, inv_errs, conv_dict = self.proc.checkConversion(
217            dict(faculty_code='ABC', local_roles=
218            "[{'user_name':'bob','local_role':'waeup.local.Boss'},]"
219            ))
220        self.assertEqual(len(errs),1)
221        self.assertEqual(errs, [('local_roles', 'waeup.local.Boss not allowed')])
222        errs, inv_errs, conv_dict = self.proc.checkConversion(
223            dict(faculty_code='ABC', local_roles=
224            "[{'user_name':'john','local_role':'waeup.local.DepartmentManager'},]"
225            ))
226        self.assertEqual(len(errs),1)
227        self.assertEqual(errs, [('local_roles', 'john does not exist')])
228        errs, inv_errs, conv_dict = self.proc.checkConversion(
229            dict(faculty_code='ABC', local_roles=
230            "[{'user_name':'bob','local_role':'waeup.local.DepartmentManager'},]"
231            ))
232        self.assertEqual(len(errs),0)
233        return
234
235    def test_import(self):
236        num, num_warns, fin_file, fail_file = self.proc.doImport(
237            self.csv_file_faculty, FACULTY_HEADER_FIELDS)
238        content = open(fail_file).read()
239        self.assertEqual(num_warns,5)
240        self.assertEqual(
241            content,
242            'code,local_roles,--ERRORS--\r\n'
243            'CDE,"[{\'user_name\':\'alice\',\'local_role\':\'waeup.local.DepartmentManager\'}]",'
244            'local_roles: alice does not exist\r\n'
245            'DEF,"[{\'user_name\':\'bob\',\'local_role\':\'waeup.local.Boss\'}]",'
246            'local_roles: waeup.local.Boss not allowed\r\n'
247            'EFG,[(\'anything\')],local_roles: no dicts\r\n'
248            'FGH,[,local_roles: Error\r\n'
249            'GHI,"[{\'user\':\'bob\',\'local\':\'waeup.local.DepartmentManager\'}]",'
250            'local_roles: user_name or local_role missing\r\n'
251            )
252        # Bob got a local role in faculty ABC.
253        abc = self.app['faculties']['ABC']
254        role_map = IPrincipalRoleMap(abc)
255        local_role, user_name, setting = role_map.getPrincipalsAndRoles()[0]
256        self.assertEqual(user_name, 'bob')
257        self.assertEqual(local_role, 'waeup.local.DepartmentManager')
258        shutil.rmtree(os.path.dirname(fin_file))
259        return
260
261    def test_import_update(self):
262        self.app['faculties']['FAC2'] = Faculty(code='FAC2')
263        self.app['faculties']['FAC3'] = Faculty(code='FAC3')
264        self.app['faculties']['FAC4'] = Faculty(code='FAC4')
265
266        role_manager1 = IPrincipalRoleManager(self.app['faculties']['FAC1'])
267        role_manager1.assignRoleToPrincipal('alfonsrole', 'alfons')
268        role_map1 = IPrincipalRoleMap(self.app['faculties']['FAC1'])
269        self.assertEqual(len(role_map1.getPrincipalsAndRoles()), 1)
270
271        role_manager2 = IPrincipalRoleManager(self.app['faculties']['FAC2'])
272        role_manager2.assignRoleToPrincipal('alfonsrole', 'alfons')
273        role_map2 = IPrincipalRoleMap(self.app['faculties']['FAC2'])
274        self.assertEqual(len(role_map2.getPrincipalsAndRoles()), 1)
275
276        role_manager3 = IPrincipalRoleManager(self.app['faculties']['FAC3'])
277        role_manager3.assignRoleToPrincipal('alfonsrole', 'alfons')
278        role_map3 = IPrincipalRoleMap(self.app['faculties']['FAC3'])
279        self.assertEqual(len(role_map3.getPrincipalsAndRoles()), 1)
280
281        role_manager4 = IPrincipalRoleManager(self.app['faculties']['FAC4'])
282        role_manager4.assignRoleToPrincipal('alfonsrole', 'alfons')
283        role_map4 = IPrincipalRoleMap(self.app['faculties']['FAC4'])
284        self.assertEqual(len(role_map4.getPrincipalsAndRoles()), 1)
285
286        local_role, user_name, setting = role_map2.getPrincipalsAndRoles()[0]
287        self.assertEqual(user_name, 'alfons')
288        self.assertEqual(local_role, 'alfonsrole')
289
290        num, num_warns, fin_file, fail_file = self.proc.doImport(
291            self.csv_file_faculty_update, FACULTY_HEADER_FIELDS_UPDATE, 'update')
292        self.assertEqual(num_warns,0)
293        # Local roles have been removed in FAC1 due to deletion marker.
294        self.assertEqual(len(role_map1.getPrincipalsAndRoles()), 0)
295        # Old local roles have been removed and new roles have been added in FAC2.
296        self.assertEqual(len(role_map2.getPrincipalsAndRoles()), 1)
297        local_role, user_name, setting = role_map2.getPrincipalsAndRoles()[0]
298        self.assertEqual(user_name, 'bob')
299        self.assertEqual(local_role, 'waeup.local.DepartmentManager')
300        # Local roles are not touched in FAC3 due to ignore marker.
301        self.assertEqual(len(role_map3.getPrincipalsAndRoles()), 1)
302        local_role, user_name, setting = role_map3.getPrincipalsAndRoles()[0]
303        self.assertEqual(user_name, 'alfons')
304        self.assertEqual(local_role, 'alfonsrole')
305        # Local roles are not touched in FAC4 due to empty cell.
306        self.assertEqual(len(role_map4.getPrincipalsAndRoles()), 1)
307        local_role, user_name, setting = role_map4.getPrincipalsAndRoles()[0]
308        self.assertEqual(user_name, 'alfons')
309        self.assertEqual(local_role, 'alfonsrole')
310        shutil.rmtree(os.path.dirname(fin_file))
311        return
312       
313    def test_import_update_logging(self):
314        self.app['faculties']['FAC2'] = Faculty(code='FAC2')
315        self.app['faculties']['FAC3'] = Faculty(code='FAC3')
316        self.app['faculties']['FAC4'] = Faculty(code='FAC4')
317        num, num_warns, fin_file, fail_file = self.proc.doImport(
318            self.csv_file_faculty_update, FACULTY_HEADER_FIELDS_UPDATE, 'update')
319        self.assertEqual(num_warns,0)
320        logcontent = open(self.logfile).read()
321        # Logging message from updateEntry
322        self.assertTrue(
323            'INFO - system - Faculty Processor - sample_faculty_data_update - '
324            'FAC1 - updated: code=FAC1'
325            in logcontent)
326        self.assertTrue(
327            'INFO - system - Faculty Processor - sample_faculty_data_update - '
328            'FAC2 - updated: local_roles=bob|waeup.local.DepartmentManager, '
329            'code=FAC2'
330            in logcontent)
331        self.assertTrue(
332            'INFO - system - Faculty Processor - sample_faculty_data_update - '
333            'FAC3 - updated: code=FAC3'
334            in logcontent)
335        self.assertTrue(
336            'INFO - system - Faculty Processor - sample_faculty_data_update - '
337            'FAC4 - updated: code=FAC4'
338            in logcontent)
339        shutil.rmtree(os.path.dirname(fin_file))
340
341    def test_upload_import_reupload(self):
342        self.browser.addHeader('Authorization', 'Basic mgr:mgrpw')
343        self.browser.open(self.datacenter_path)
344        self.assertEqual(self.browser.headers['Status'], '200 Ok')
345        self.assertEqual(self.browser.url, self.datacenter_path)
346        self.browser.getLink("Upload data").click()
347        file = open(self.csv_file_faculty)
348        ctrl = self.browser.getControl(name='uploadfile:file')
349        file_ctrl = ctrl.mech_control
350        file_ctrl.add_file(file, filename='sample_faculty_data.csv')
351        self.browser.getControl('Upload').click()
352        self.browser.getLink('Process data').click()
353        self.browser.getControl(name="select").click()
354        importerselect = self.browser.getControl(name='importer')
355        importerselect.getControl('Faculty Processor').selected = True
356        modeselect = self.browser.getControl(name='mode')
357        modeselect.getControl(value='create').selected = True
358        self.browser.getControl('Proceed to step 3').click()
359        self.assertTrue('Header fields OK' in self.browser.contents)
360        self.browser.getControl('Perform import').click()
361        self.assertTrue('Successfully processed 1 rows' in self.browser.contents)
362        # We import the same file a second time.
363        self.browser.open(self.datacenter_path)
364        self.browser.getLink("Upload data").click()
365        file = open(self.csv_file_faculty)
366        ctrl = self.browser.getControl(name='uploadfile:file')
367        file_ctrl = ctrl.mech_control
368        file_ctrl.add_file(file, filename='sample_faculty_data.csv')
369        self.browser.getControl('Upload').click()
370        self.assertTrue(
371            'File with same name was uploaded earlier' in self.browser.contents)
372        return
373
374class TestDepartmentProcessor(UniversityProcessorSetup):
375
376    def setUp(self):
377        super(TestDepartmentProcessor, self).setUp()
378        self.proc = DepartmentProcessor()
379        self.site0 = dict()
380        self.site1 = dict(faculties=dict())
381        self.site2 = dict(faculties=dict(FAC=dict()))
382        self.site3 = dict(faculties=dict(FAC=dict(DPT='pseudo department')))
383        self.row = dict(code='DPT', faculty_code='FAC')
384
385        self.csv_file_department = os.path.join(self.workdir, 'sample_department_data.csv')
386        open(self.csv_file_department, 'wb').write(DEPARTMENT_SAMPLE_DATA)
387        return
388
389    def test_ifaces(self):
390        # Make sure we fullfill all interface contracts
391        verifyClass(IBatchProcessor, DepartmentProcessor)
392        verifyObject(IBatchProcessor, self.proc)
393        return
394
395    def test_parents_exist(self):
396        # make sure we lookup parents correctly.
397        result0 = self.proc.parentsExist(self.row, self.site0)
398        result1 = self.proc.parentsExist(self.row, self.site1)
399        result2 = self.proc.parentsExist(self.row, self.site2)
400        result3 = self.proc.parentsExist(self.row, self.site3)
401        self.assertTrue(result0 is False)
402        self.assertTrue(result1 is False)
403        self.assertTrue(result2 is True)
404        self.assertTrue(result3 is True)
405        return
406
407    def test_entry_exists(self):
408        # make sure we lookup entries correctly.
409        result0 = self.proc.entryExists(self.row, dict())
410        result1 = self.proc.entryExists(self.row, self.site1)
411        result2 = self.proc.entryExists(self.row, self.site2)
412        result3 = self.proc.entryExists(self.row, self.site3)
413        self.assertTrue(result0 is False)
414        self.assertTrue(result1 is False)
415        self.assertTrue(result2 is False)
416        self.assertTrue(result3 is True)
417        return
418
419    def test_get_entry(self):
420        # we can get a dept. if it exists
421        result1 = self.proc.getEntry(self.row, self.site2)
422        result2 = self.proc.getEntry(self.row, self.site3)
423        self.assertTrue(result1 is None)
424        self.assertEqual(result2, 'pseudo department')
425        return
426
427    def test_del_entry(self):
428        # we can delete departments
429        self.proc.delEntry(self.row, self.site3)
430        self.assertTrue('DPT' not in self.site3['faculties']['FAC'].keys())
431        return
432
433    def test_import(self):
434        num, num_warns, fin_file, fail_file = self.proc.doImport(
435            self.csv_file_department, DEPARTMENT_HEADER_FIELDS)
436        content = open(fail_file).read()
437        self.assertEqual(num_warns,6)
438        self.assertEqual(
439            content,
440            'faculty_code,code,local_roles,--ERRORS--\r\n'
441            'FAC1,DEP2,"[{\'user_name\':\'alice\',\'local_role\':\'waeup.local.DepartmentManager\'}]",'
442            'local_roles: alice does not exist\r\n'
443            'FAC1,DEP2,"[{\'user_name\':\'anne\',\'local_role\':\'waeup.local.Boss\'}]",'
444            'local_roles: waeup.local.Boss not allowed\r\n'
445            'FAC1,DEP2,[(\'anything\')],local_roles: no dicts\r\n'
446            'FAC1,DEP2,[,local_roles: Error\r\n'
447            'FAC1,DEP2,"[{\'user\':\'anne\',\'local\':\'waeup.local.DepartmentManager\'}]",'
448            'local_roles: user_name or local_role missing\r\n'
449            'FAC11,DEP2,"[{\'user_name\':\'anne\',\'local_role\':\'waeup.local.DepartmentManager\'}]",'
450            'Not all parents do exist yet.\r\n'
451            )
452        # Anne got a local role in department DEP2.
453        dep = self.app['faculties']['FAC1']['DEP2']
454        role_map = IPrincipalRoleMap(dep)
455        local_role, user_name, setting = role_map.getPrincipalsAndRoles()[0]
456        self.assertEqual(user_name, 'anne')
457        self.assertEqual(local_role, 'waeup.local.DepartmentManager')
458        shutil.rmtree(os.path.dirname(fin_file))
459        return
460
461class CourseProcessorTests(UniversityProcessorSetup):
462
463    def setUp(self):
464        super(CourseProcessorTests, self).setUp()
465        self.proc = CourseProcessor()
466        self.row1 = dict(department_code='DEP2', faculty_code='FAC1', code="CRS1")
467        self.row2 = dict(department_code='DEP1', faculty_code='FAC2', code="CRS1")
468        self.row3 = dict(department_code='DEP1', faculty_code='FAC1', code="CRS2")
469        self.row4 = dict(department_code='DEP1', faculty_code='FAC1', code="CRS1")
470
471        self.csv_file_course = os.path.join(
472            self.workdir, 'sample_course_data.csv')
473        open(self.csv_file_course, 'wb').write(COURSE_SAMPLE_DATA)
474        self.csv_file_course_update = os.path.join(
475            self.workdir, 'sample_course_data_update.csv')
476        open(self.csv_file_course_update, 'wb').write(COURSE_SAMPLE_DATA_UPDATE)
477        return
478
479    def test_ifaces(self):
480        # Make sure we fullfill all interface contracts
481        verifyClass(IBatchProcessor, CourseProcessor)
482        verifyObject(IBatchProcessor, self.proc)
483        return
484
485    def test_parents_exist(self):
486        # make sure we lookup parents correctly
487        result1 = self.proc.parentsExist(self.row1, self.app)
488        result2 = self.proc.parentsExist(self.row2, self.app)
489        result3 = self.proc.parentsExist(self.row3, self.app)
490        self.assertTrue(result1 is False)
491        self.assertTrue(result2 is False)
492        self.assertTrue(result3 is True)
493        return
494
495    def test_entry_exists(self):
496        # make sure we find an entry if it exists
497        result1 = self.proc.entryExists(self.row1, self.app)
498        result2 = self.proc.entryExists(self.row2, self.app)
499        result3 = self.proc.entryExists(self.row3, self.app)
500        result4 = self.proc.entryExists(self.row4, self.app)
501        self.assertTrue(result1 is False)
502        self.assertTrue(result2 is False)
503        self.assertTrue(result3 is False)
504        self.assertTrue(result4 is True)
505        return
506
507    def test_get_entry(self):
508        # make sure we can get an entry if it exists
509        result1 = self.proc.getEntry(self.row1, self.app)
510        result2 = self.proc.getEntry(self.row4, self.app)
511        self.assertTrue(result1 is None)
512        self.assertTrue(result2 is self.course)
513        return
514
515    def test_del_entry(self):
516        # make sure we can delete entries
517        self.assertTrue('CRS1' in self.app['faculties']['FAC1']['DEP1'].courses.keys())
518        self.proc.delEntry(self.row4, self.app)
519        self.assertTrue('CRS1' not in self.app['faculties']['FAC1']['DEP1'].courses.keys())
520        return
521
522    def test_import(self):
523        num, num_warns, fin_file, fail_file = self.proc.doImport(
524            self.csv_file_course, COURSE_HEADER_FIELDS)
525        content = open(fail_file).read()
526        self.assertEqual(num_warns,6)
527        self.assertEqual(
528            content,
529            'faculty_code,department_code,code,local_roles,--ERRORS--\r\n'
530            'FAC1,DEP1,CRS2,"[{\'user_name\':\'alice\',\'local_role\':\'waeup.local.Lecturer\'}]",'
531            'local_roles: alice does not exist\r\n'
532            'FAC1,DEP1,CRS2,"[{\'user_name\':\'anne\',\'local_role\':\'waeup.local.Boss\'}]",'
533            'local_roles: waeup.local.Boss not allowed\r\n'
534            'FAC1,DEP1,CRS2,[(\'anything\')],local_roles: no dicts\r\n'
535            'FAC1,DEP1,CRS2,[,local_roles: Error\r\n'
536            'FAC1,DEP1,CRS2,"[{\'user\':\'anne\',\'local\':\'waeup.local.Lecturer\'}]",'
537            'local_roles: user_name or local_role missing\r\n'
538            'FAC11,DEP2,CRS2,"[{\'user_name\':\'anne\',\'local_role\':\'waeup.local.Lecturer\'}]",'
539            'Not all parents do exist yet.\r\n'
540            )
541        # Anne got a local role in course CRS2.
542        dep = self.app['faculties']['FAC1']['DEP1'].courses['CRS2']
543        role_map = IPrincipalRoleMap(dep)
544        local_role, user_name, setting = role_map.getPrincipalsAndRoles()[0]
545        self.assertEqual(user_name, 'anne')
546        self.assertEqual(local_role, 'waeup.local.Lecturer')
547        shutil.rmtree(os.path.dirname(fin_file))
548        return
549
550    def test_import_update(self):
551        num, num_warns, fin_file, fail_file = self.proc.doImport(
552            self.csv_file_course, COURSE_HEADER_FIELDS)
553        shutil.rmtree(os.path.dirname(fin_file))
554        num, num_warns, fin_file, fail_file = self.proc.doImport(
555            self.csv_file_course_update, COURSE_HEADER_FIELDS_UPDATE, 'update')
556        self.assertEqual(num_warns,0)
557        self.assertEqual(
558            self.app['faculties']['FAC1']['DEP1'].courses['CRS2'].title,
559            'New Title')
560        shutil.rmtree(os.path.dirname(fin_file))
561        return
562
563
564class CertificateProcessorTests(UniversityProcessorSetup):
565
566    def setUp(self):
567        super(CertificateProcessorTests, self).setUp()
568        self.proc = CertificateProcessor()
569        self.row1 = dict(department_code='DEP2', faculty_code='FAC1', code="CRT1")
570        self.row2 = dict(department_code='DEP1', faculty_code='FAC2', code="CRT1")
571        self.row3 = dict(department_code='DEP1', faculty_code='FAC1', code="CRT2")
572        self.row4 = dict(department_code='DEP1', faculty_code='FAC1', code="CRT1")
573
574        self.csv_file_certificate = os.path.join(self.workdir, 'sample_certificate_data.csv')
575        open(self.csv_file_certificate, 'wb').write(CERTIFICATE_SAMPLE_DATA)
576        return
577
578    def test_ifaces(self):
579        # Make sure we fullfill all interface contracts
580        verifyClass(IBatchProcessor, CourseProcessor)
581        verifyObject(IBatchProcessor, self.proc)
582        return
583
584    def test_parents_exist(self):
585        # make sure we lookup parents correctly
586        result1 = self.proc.parentsExist(self.row1, self.app)
587        result2 = self.proc.parentsExist(self.row2, self.app)
588        result3 = self.proc.parentsExist(self.row3, self.app)
589        self.assertTrue(result1 is False)
590        self.assertTrue(result2 is False)
591        self.assertTrue(result3 is True)
592        return
593
594    def test_entry_exists(self):
595        # make sure we find an entry if it exists
596        result1 = self.proc.entryExists(self.row1, self.app)
597        result2 = self.proc.entryExists(self.row2, self.app)
598        result3 = self.proc.entryExists(self.row3, self.app)
599        result4 = self.proc.entryExists(self.row4, self.app)
600        self.assertTrue(result1 is False)
601        self.assertTrue(result2 is False)
602        self.assertTrue(result3 is False)
603        self.assertTrue(result4 is True)
604        return
605
606    def test_get_entry(self):
607        # make sure we can get an entry if it exists
608        result1 = self.proc.getEntry(self.row1, self.app)
609        result2 = self.proc.getEntry(self.row4, self.app)
610        self.assertTrue(result1 is None)
611        self.assertTrue(result2 is self.certificate)
612        return
613
614    def test_del_entry(self):
615        # make sure we can delete entries
616        self.assertTrue('CRT1' in self.app['faculties']['FAC1']['DEP1'].certificates.keys())
617        self.proc.delEntry(self.row4, self.app)
618        self.assertTrue('CRT1' not in self.app['faculties']['FAC1']['DEP1'].certificates.keys())
619        return
620
621    def test_import(self):
622        num, num_warns, fin_file, fail_file = self.proc.doImport(
623            self.csv_file_certificate, CERTIFICATE_HEADER_FIELDS)
624        content = open(fail_file).read()
625        self.assertEqual(num_warns,6)
626        self.assertEqual(
627            content,
628            'faculty_code,department_code,code,local_roles,--ERRORS--\r\n'
629            'FAC1,DEP1,CRT2,"[{\'user_name\':\'alice\',\'local_role\':\'waeup.local.CourseAdviser100\'}]",'
630            'local_roles: alice does not exist\r\n'
631            'FAC1,DEP1,CRT2,"[{\'user_name\':\'anne\',\'local_role\':\'waeup.local.Boss\'}]",'
632            'local_roles: waeup.local.Boss not allowed\r\n'
633            'FAC1,DEP1,CRT2,[(\'anything\')],local_roles: no dicts\r\n'
634            'FAC1,DEP1,CRT2,[,local_roles: Error\r\n'
635            'FAC1,DEP1,CRT2,"[{\'user\':\'anne\',\'local\':\'waeup.local.CourseAdviser100\'}]",'
636            'local_roles: user_name or local_role missing\r\n'
637            'FAC11,DEP2,CRT2,"[{\'user_name\':\'anne\',\'local_role\':\'waeup.local.CourseAdviser100\'}]",'
638            'Not all parents do exist yet.\r\n'
639            )
640        # Anne got a local role in certificate CRT2.
641        dep = self.app['faculties']['FAC1']['DEP1'].certificates['CRT2']
642        role_map = IPrincipalRoleMap(dep)
643        local_role, user_name, setting = role_map.getPrincipalsAndRoles()[0]
644        self.assertEqual(user_name, 'anne')
645        self.assertEqual(local_role, 'waeup.local.CourseAdviser100')
646        shutil.rmtree(os.path.dirname(fin_file))
647       
648        logcontent = open(self.logfile).read()
649        # Logging message from updateEntry
650        self.assertTrue(
651            'INFO - system - Certificate Processor - '
652            'sample_certificate_data - CRT2 - '
653            'updated: local_roles=anne|waeup.local.CourseAdviser100, code=CRT2'
654            in logcontent)       
655       
656        return
657
658
659class CertCourseProcessorTests(UniversityProcessorSetup):
660
661    layer = FunctionalLayer
662
663    def setUp(self):
664        super(CertCourseProcessorTests, self).setUp()
665        self.proc = CertificateCourseProcessor()
666        self.certificate.addCertCourse(self.course)
667        self.row1 = dict(
668            department_code='DEP1',
669            faculty_code='FAC1',
670            certificate_code='CRT1',
671            course=self.course, level='100')
672        self.row2 = dict(
673            department_code='DEP1',
674            faculty_code='FAC1',
675            certificate_code='CRT2',
676            course=self.course, level='100')
677        self.csv_file_certificatecourse = os.path.join(
678            self.workdir, 'sample_certificatecourse_data.csv')
679        open(self.csv_file_certificatecourse, 'wb').write(CERTIFICATECOURSE_SAMPLE_DATA)
680        return
681
682    def test_ifaces(self):
683        # Make sure we fullfill all interface contracts
684        verifyClass(IBatchProcessor, CertificateCourseProcessor)
685        verifyObject(IBatchProcessor, self.proc)
686        return
687
688    def test_parents_exist(self):
689        # make sure we can find all certificate parents
690        result1 = self.proc.parentsExist(self.row1, self.app)
691        self.assertTrue(result1)
692        result2 = self.proc.parentsExist(self.row2, self.app)
693        self.assertFalse(result2)
694        return
695
696    def test_entry_exists(self):
697        # make sure we find an entry if it exists
698        result1 = self.proc.entryExists(self.row1, self.app)
699        self.assertTrue(result1)
700        result2 = self.proc.entryExists(self.row2, self.app)
701        self.assertFalse(result2)
702        return
703
704    def test_get_entry(self):
705        # make sure we can get an entry if it exists
706        result1 = self.proc.getEntry(self.row1, self.app)
707        self.assertTrue(result1 is self.certificate['CRS1_100'])
708        result2 = self.proc.getEntry(self.row2, self.app)
709        self.assertFalse(result2 is self.certificate['CRS1_100'])
710        return
711
712    def test_del_entry(self):
713        # make sure we can delete entries
714        self.assertTrue('CRS1_100' in self.certificate.keys())
715        self.proc.delEntry(self.row1, self.app)
716        self.assertTrue('CRS1_100' not in self.certificate.keys())
717        return
718
719    def test_import(self):
720        num, num_warns, fin_file, fail_file = self.proc.doImport(
721            self.csv_file_certificatecourse, CERTIFICATECOURSE_HEADER_FIELDS)
722        content = open(fail_file).read()
723        self.assertEqual(num_warns,2)
724        self.assertEqual(
725            content,
726            'faculty_code,course,level,department_code,certificate_code,'
727            '--ERRORS--\r\nFAC1,CRS1,100,DEP1,CRT1,'
728            'This object already exists.\r\nFAC1,CRS1,100,DEP1,CRT2,'
729            'Not all parents do exist yet.\r\n'
730
731            )
732        logcontent = open(self.logfile).read()
733        # Logging message from updateEntry
734        self.assertTrue(
735            'INFO - system - CertificateCourse Processor - '
736            'sample_certificatecourse_data - CRT1 - updated: '
737            'course=CRS1, level=200\n'
738            in logcontent)
739
740        return
Note: See TracBrowser for help on using the repository browser.