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

Last change on this file since 13418 was 13394, checked in by Henrik Bettermann, 9 years ago

Implement portal maintenance mode.
  • Property svn:keywords set to Id
File size: 32.2 KB
Line 
1## $Id: test_batching.py 13394 2015-11-06 05:43:37Z 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.assertTrue('Portal must be in maintenance mode for data import.'
354            in self.browser.contents)
355        self.browser.getLink("Switch").click()
356        self.assertTrue('Maintenance mode enabled' in self.browser.contents)
357        # Maintenance mode is now set.
358        self.assertEqual(
359            self.app['configuration'].maintmode_enabled_by, 'zope.mgr')
360        self.browser.getLink('Process data').click()
361        self.browser.getControl(name="select").click()
362        importerselect = self.browser.getControl(name='importer')
363        importerselect.getControl('Faculty Processor').selected = True
364        modeselect = self.browser.getControl(name='mode')
365        modeselect.getControl(value='create').selected = True
366        self.browser.getControl('Proceed to step 3').click()
367        self.assertTrue('Header fields OK' in self.browser.contents)
368        self.browser.getControl('Perform import').click()
369        self.assertTrue('Successfully processed 1 rows' in self.browser.contents)
370        self.assertTrue('Maintenance mode disabled' in self.browser.contents)
371        # Maintenance mode is really disabled.
372        self.assertEqual(
373            self.app['configuration'].maintmode_enabled_by, None)
374        # We import the same file a second time.
375        self.browser.open(self.datacenter_path)
376        self.browser.getLink("Upload data").click()
377        file = open(self.csv_file_faculty)
378        ctrl = self.browser.getControl(name='uploadfile:file')
379        file_ctrl = ctrl.mech_control
380        file_ctrl.add_file(file, filename='sample_faculty_data.csv')
381        self.browser.getControl('Upload').click()
382        self.assertTrue(
383            'File with same name was uploaded earlier' in self.browser.contents)
384        return
385
386class TestDepartmentProcessor(UniversityProcessorSetup):
387
388    def setUp(self):
389        super(TestDepartmentProcessor, self).setUp()
390        self.proc = DepartmentProcessor()
391        self.site0 = dict()
392        self.site1 = dict(faculties=dict())
393        self.site2 = dict(faculties=dict(FAC=dict()))
394        self.site3 = dict(faculties=dict(FAC=dict(DPT='pseudo department')))
395        self.row = dict(code='DPT', faculty_code='FAC')
396
397        self.csv_file_department = os.path.join(self.workdir, 'sample_department_data.csv')
398        open(self.csv_file_department, 'wb').write(DEPARTMENT_SAMPLE_DATA)
399        return
400
401    def test_ifaces(self):
402        # Make sure we fullfill all interface contracts
403        verifyClass(IBatchProcessor, DepartmentProcessor)
404        verifyObject(IBatchProcessor, self.proc)
405        return
406
407    def test_parents_exist(self):
408        # make sure we lookup parents correctly.
409        result0 = self.proc.parentsExist(self.row, self.site0)
410        result1 = self.proc.parentsExist(self.row, self.site1)
411        result2 = self.proc.parentsExist(self.row, self.site2)
412        result3 = self.proc.parentsExist(self.row, self.site3)
413        self.assertTrue(result0 is False)
414        self.assertTrue(result1 is False)
415        self.assertTrue(result2 is True)
416        self.assertTrue(result3 is True)
417        return
418
419    def test_entry_exists(self):
420        # make sure we lookup entries correctly.
421        result0 = self.proc.entryExists(self.row, dict())
422        result1 = self.proc.entryExists(self.row, self.site1)
423        result2 = self.proc.entryExists(self.row, self.site2)
424        result3 = self.proc.entryExists(self.row, self.site3)
425        self.assertTrue(result0 is False)
426        self.assertTrue(result1 is False)
427        self.assertTrue(result2 is False)
428        self.assertTrue(result3 is True)
429        return
430
431    def test_get_entry(self):
432        # we can get a dept. if it exists
433        result1 = self.proc.getEntry(self.row, self.site2)
434        result2 = self.proc.getEntry(self.row, self.site3)
435        self.assertTrue(result1 is None)
436        self.assertEqual(result2, 'pseudo department')
437        return
438
439    def test_del_entry(self):
440        # we can delete departments
441        self.proc.delEntry(self.row, self.site3)
442        self.assertTrue('DPT' not in self.site3['faculties']['FAC'].keys())
443        return
444
445    def test_import(self):
446        num, num_warns, fin_file, fail_file = self.proc.doImport(
447            self.csv_file_department, DEPARTMENT_HEADER_FIELDS)
448        content = open(fail_file).read()
449        self.assertEqual(num_warns,6)
450        self.assertEqual(
451            content,
452            'faculty_code,code,local_roles,--ERRORS--\r\n'
453            'FAC1,DEP2,"[{\'user_name\':\'alice\',\'local_role\':\'waeup.local.DepartmentManager\'}]",'
454            'local_roles: alice does not exist\r\n'
455            'FAC1,DEP2,"[{\'user_name\':\'anne\',\'local_role\':\'waeup.local.Boss\'}]",'
456            'local_roles: waeup.local.Boss not allowed\r\n'
457            'FAC1,DEP2,[(\'anything\')],local_roles: no dicts\r\n'
458            'FAC1,DEP2,[,local_roles: Error\r\n'
459            'FAC1,DEP2,"[{\'user\':\'anne\',\'local\':\'waeup.local.DepartmentManager\'}]",'
460            'local_roles: user_name or local_role missing\r\n'
461            'FAC11,DEP2,"[{\'user_name\':\'anne\',\'local_role\':\'waeup.local.DepartmentManager\'}]",'
462            'Not all parents do exist yet.\r\n'
463            )
464        # Anne got a local role in department DEP2.
465        dep = self.app['faculties']['FAC1']['DEP2']
466        role_map = IPrincipalRoleMap(dep)
467        local_role, user_name, setting = role_map.getPrincipalsAndRoles()[0]
468        self.assertEqual(user_name, 'anne')
469        self.assertEqual(local_role, 'waeup.local.DepartmentManager')
470        shutil.rmtree(os.path.dirname(fin_file))
471        return
472
473class CourseProcessorTests(UniversityProcessorSetup):
474
475    def setUp(self):
476        super(CourseProcessorTests, self).setUp()
477        self.proc = CourseProcessor()
478        self.row1 = dict(department_code='DEP2', faculty_code='FAC1', code="CRS1")
479        self.row2 = dict(department_code='DEP1', faculty_code='FAC2', code="CRS1")
480        self.row3 = dict(department_code='DEP1', faculty_code='FAC1', code="CRS2")
481        self.row4 = dict(department_code='DEP1', faculty_code='FAC1', code="CRS1")
482
483        self.csv_file_course = os.path.join(
484            self.workdir, 'sample_course_data.csv')
485        open(self.csv_file_course, 'wb').write(COURSE_SAMPLE_DATA)
486        self.csv_file_course_update = os.path.join(
487            self.workdir, 'sample_course_data_update.csv')
488        open(self.csv_file_course_update, 'wb').write(COURSE_SAMPLE_DATA_UPDATE)
489        return
490
491    def test_ifaces(self):
492        # Make sure we fullfill all interface contracts
493        verifyClass(IBatchProcessor, CourseProcessor)
494        verifyObject(IBatchProcessor, self.proc)
495        return
496
497    def test_parents_exist(self):
498        # make sure we lookup parents correctly
499        result1 = self.proc.parentsExist(self.row1, self.app)
500        result2 = self.proc.parentsExist(self.row2, self.app)
501        result3 = self.proc.parentsExist(self.row3, self.app)
502        self.assertTrue(result1 is False)
503        self.assertTrue(result2 is False)
504        self.assertTrue(result3 is True)
505        return
506
507    def test_entry_exists(self):
508        # make sure we find an entry if it exists
509        result1 = self.proc.entryExists(self.row1, self.app)
510        result2 = self.proc.entryExists(self.row2, self.app)
511        result3 = self.proc.entryExists(self.row3, self.app)
512        result4 = self.proc.entryExists(self.row4, self.app)
513        self.assertTrue(result1 is False)
514        self.assertTrue(result2 is False)
515        self.assertTrue(result3 is False)
516        self.assertTrue(result4 is True)
517        return
518
519    def test_get_entry(self):
520        # make sure we can get an entry if it exists
521        result1 = self.proc.getEntry(self.row1, self.app)
522        result2 = self.proc.getEntry(self.row4, self.app)
523        self.assertTrue(result1 is None)
524        self.assertTrue(result2 is self.course)
525        return
526
527    def test_del_entry(self):
528        # make sure we can delete entries
529        self.assertTrue('CRS1' in self.app['faculties']['FAC1']['DEP1'].courses.keys())
530        self.proc.delEntry(self.row4, self.app)
531        self.assertTrue('CRS1' not in self.app['faculties']['FAC1']['DEP1'].courses.keys())
532        return
533
534    def test_import(self):
535        num, num_warns, fin_file, fail_file = self.proc.doImport(
536            self.csv_file_course, COURSE_HEADER_FIELDS)
537        content = open(fail_file).read()
538        self.assertEqual(num_warns,6)
539        self.assertEqual(
540            content,
541            'faculty_code,department_code,code,local_roles,--ERRORS--\r\n'
542            'FAC1,DEP1,CRS2,"[{\'user_name\':\'alice\',\'local_role\':\'waeup.local.Lecturer\'}]",'
543            'local_roles: alice does not exist\r\n'
544            'FAC1,DEP1,CRS2,"[{\'user_name\':\'anne\',\'local_role\':\'waeup.local.Boss\'}]",'
545            'local_roles: waeup.local.Boss not allowed\r\n'
546            'FAC1,DEP1,CRS2,[(\'anything\')],local_roles: no dicts\r\n'
547            'FAC1,DEP1,CRS2,[,local_roles: Error\r\n'
548            'FAC1,DEP1,CRS2,"[{\'user\':\'anne\',\'local\':\'waeup.local.Lecturer\'}]",'
549            'local_roles: user_name or local_role missing\r\n'
550            'FAC11,DEP2,CRS2,"[{\'user_name\':\'anne\',\'local_role\':\'waeup.local.Lecturer\'}]",'
551            'Not all parents do exist yet.\r\n'
552            )
553        # Anne got a local role in course CRS2.
554        dep = self.app['faculties']['FAC1']['DEP1'].courses['CRS2']
555        role_map = IPrincipalRoleMap(dep)
556        local_role, user_name, setting = role_map.getPrincipalsAndRoles()[0]
557        self.assertEqual(user_name, 'anne')
558        self.assertEqual(local_role, 'waeup.local.Lecturer')
559        shutil.rmtree(os.path.dirname(fin_file))
560        return
561
562    def test_import_update(self):
563        num, num_warns, fin_file, fail_file = self.proc.doImport(
564            self.csv_file_course, COURSE_HEADER_FIELDS)
565        shutil.rmtree(os.path.dirname(fin_file))
566        num, num_warns, fin_file, fail_file = self.proc.doImport(
567            self.csv_file_course_update, COURSE_HEADER_FIELDS_UPDATE, 'update')
568        self.assertEqual(num_warns,0)
569        self.assertEqual(
570            self.app['faculties']['FAC1']['DEP1'].courses['CRS2'].title,
571            'New Title')
572        shutil.rmtree(os.path.dirname(fin_file))
573        return
574
575
576class CertificateProcessorTests(UniversityProcessorSetup):
577
578    def setUp(self):
579        super(CertificateProcessorTests, self).setUp()
580        self.proc = CertificateProcessor()
581        self.row1 = dict(department_code='DEP2', faculty_code='FAC1', code="CRT1")
582        self.row2 = dict(department_code='DEP1', faculty_code='FAC2', code="CRT1")
583        self.row3 = dict(department_code='DEP1', faculty_code='FAC1', code="CRT2")
584        self.row4 = dict(department_code='DEP1', faculty_code='FAC1', code="CRT1")
585
586        self.csv_file_certificate = os.path.join(self.workdir, 'sample_certificate_data.csv')
587        open(self.csv_file_certificate, 'wb').write(CERTIFICATE_SAMPLE_DATA)
588        return
589
590    def test_ifaces(self):
591        # Make sure we fullfill all interface contracts
592        verifyClass(IBatchProcessor, CourseProcessor)
593        verifyObject(IBatchProcessor, self.proc)
594        return
595
596    def test_parents_exist(self):
597        # make sure we lookup parents correctly
598        result1 = self.proc.parentsExist(self.row1, self.app)
599        result2 = self.proc.parentsExist(self.row2, self.app)
600        result3 = self.proc.parentsExist(self.row3, self.app)
601        self.assertTrue(result1 is False)
602        self.assertTrue(result2 is False)
603        self.assertTrue(result3 is True)
604        return
605
606    def test_entry_exists(self):
607        # make sure we find an entry if it exists
608        result1 = self.proc.entryExists(self.row1, self.app)
609        result2 = self.proc.entryExists(self.row2, self.app)
610        result3 = self.proc.entryExists(self.row3, self.app)
611        result4 = self.proc.entryExists(self.row4, self.app)
612        self.assertTrue(result1 is False)
613        self.assertTrue(result2 is False)
614        self.assertTrue(result3 is False)
615        self.assertTrue(result4 is True)
616        return
617
618    def test_get_entry(self):
619        # make sure we can get an entry if it exists
620        result1 = self.proc.getEntry(self.row1, self.app)
621        result2 = self.proc.getEntry(self.row4, self.app)
622        self.assertTrue(result1 is None)
623        self.assertTrue(result2 is self.certificate)
624        return
625
626    def test_del_entry(self):
627        # make sure we can delete entries
628        self.assertTrue('CRT1' in self.app['faculties']['FAC1']['DEP1'].certificates.keys())
629        self.proc.delEntry(self.row4, self.app)
630        self.assertTrue('CRT1' not in self.app['faculties']['FAC1']['DEP1'].certificates.keys())
631        return
632
633    def test_import(self):
634        num, num_warns, fin_file, fail_file = self.proc.doImport(
635            self.csv_file_certificate, CERTIFICATE_HEADER_FIELDS)
636        content = open(fail_file).read()
637        self.assertEqual(num_warns,6)
638        self.assertEqual(
639            content,
640            'faculty_code,department_code,code,local_roles,--ERRORS--\r\n'
641            'FAC1,DEP1,CRT2,"[{\'user_name\':\'alice\',\'local_role\':\'waeup.local.CourseAdviser100\'}]",'
642            'local_roles: alice does not exist\r\n'
643            'FAC1,DEP1,CRT2,"[{\'user_name\':\'anne\',\'local_role\':\'waeup.local.Boss\'}]",'
644            'local_roles: waeup.local.Boss not allowed\r\n'
645            'FAC1,DEP1,CRT2,[(\'anything\')],local_roles: no dicts\r\n'
646            'FAC1,DEP1,CRT2,[,local_roles: Error\r\n'
647            'FAC1,DEP1,CRT2,"[{\'user\':\'anne\',\'local\':\'waeup.local.CourseAdviser100\'}]",'
648            'local_roles: user_name or local_role missing\r\n'
649            'FAC11,DEP2,CRT2,"[{\'user_name\':\'anne\',\'local_role\':\'waeup.local.CourseAdviser100\'}]",'
650            'Not all parents do exist yet.\r\n'
651            )
652        # Anne got a local role in certificate CRT2.
653        dep = self.app['faculties']['FAC1']['DEP1'].certificates['CRT2']
654        role_map = IPrincipalRoleMap(dep)
655        local_role, user_name, setting = role_map.getPrincipalsAndRoles()[0]
656        self.assertEqual(user_name, 'anne')
657        self.assertEqual(local_role, 'waeup.local.CourseAdviser100')
658        shutil.rmtree(os.path.dirname(fin_file))
659       
660        logcontent = open(self.logfile).read()
661        # Logging message from updateEntry
662        self.assertTrue(
663            'INFO - system - Certificate Processor - '
664            'sample_certificate_data - CRT2 - '
665            'updated: local_roles=anne|waeup.local.CourseAdviser100, code=CRT2'
666            in logcontent)       
667       
668        return
669
670
671class CertCourseProcessorTests(UniversityProcessorSetup):
672
673    layer = FunctionalLayer
674
675    def setUp(self):
676        super(CertCourseProcessorTests, self).setUp()
677        self.proc = CertificateCourseProcessor()
678        self.certificate.addCertCourse(self.course)
679        self.row1 = dict(
680            department_code='DEP1',
681            faculty_code='FAC1',
682            certificate_code='CRT1',
683            course=self.course, level='100')
684        self.row2 = dict(
685            department_code='DEP1',
686            faculty_code='FAC1',
687            certificate_code='CRT2',
688            course=self.course, level='100')
689        self.csv_file_certificatecourse = os.path.join(
690            self.workdir, 'sample_certificatecourse_data.csv')
691        open(self.csv_file_certificatecourse, 'wb').write(CERTIFICATECOURSE_SAMPLE_DATA)
692        return
693
694    def test_ifaces(self):
695        # Make sure we fullfill all interface contracts
696        verifyClass(IBatchProcessor, CertificateCourseProcessor)
697        verifyObject(IBatchProcessor, self.proc)
698        return
699
700    def test_parents_exist(self):
701        # make sure we can find all certificate parents
702        result1 = self.proc.parentsExist(self.row1, self.app)
703        self.assertTrue(result1)
704        result2 = self.proc.parentsExist(self.row2, self.app)
705        self.assertFalse(result2)
706        return
707
708    def test_entry_exists(self):
709        # make sure we find an entry if it exists
710        result1 = self.proc.entryExists(self.row1, self.app)
711        self.assertTrue(result1)
712        result2 = self.proc.entryExists(self.row2, self.app)
713        self.assertFalse(result2)
714        return
715
716    def test_get_entry(self):
717        # make sure we can get an entry if it exists
718        result1 = self.proc.getEntry(self.row1, self.app)
719        self.assertTrue(result1 is self.certificate['CRS1_100'])
720        result2 = self.proc.getEntry(self.row2, self.app)
721        self.assertFalse(result2 is self.certificate['CRS1_100'])
722        return
723
724    def test_del_entry(self):
725        # make sure we can delete entries
726        self.assertTrue('CRS1_100' in self.certificate.keys())
727        self.proc.delEntry(self.row1, self.app)
728        self.assertTrue('CRS1_100' not in self.certificate.keys())
729        return
730
731    def test_import(self):
732        num, num_warns, fin_file, fail_file = self.proc.doImport(
733            self.csv_file_certificatecourse, CERTIFICATECOURSE_HEADER_FIELDS)
734        content = open(fail_file).read()
735        self.assertEqual(num_warns,2)
736        self.assertEqual(
737            content,
738            'faculty_code,course,level,department_code,certificate_code,'
739            '--ERRORS--\r\nFAC1,CRS1,100,DEP1,CRT1,'
740            'This object already exists.\r\nFAC1,CRS1,100,DEP1,CRT2,'
741            'Not all parents do exist yet.\r\n'
742
743            )
744        logcontent = open(self.logfile).read()
745        # Logging message from updateEntry
746        self.assertTrue(
747            'INFO - system - CertificateCourse Processor - '
748            'sample_certificatecourse_data - CRT1 - updated: '
749            'course=CRS1, level=200\n'
750            in logcontent)
751
752        return
Note: See TracBrowser for help on using the repository browser.