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

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

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