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

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

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