source: main/waeup.sirp/trunk/src/waeup/sirp/students/tests/test_batching.py @ 7633

Last change on this file since 7633 was 7623, checked in by Henrik Bettermann, 13 years ago

Implement first version of payment importer.
  • Property svn:keywords set to Id
File size: 26.5 KB
Line 
1## $Id: test_batching.py 7623 2012-02-10 12:08:47Z 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"""Unit tests for students-related data importers.
19"""
20import os
21import shutil
22import tempfile
23import unittest
24import datetime
25from time import time
26from zope.component import createObject
27from zope.component.hooks import setSite, clearSite
28from zope.interface.verify import verifyClass, verifyObject
29
30from waeup.sirp.app import University
31from waeup.sirp.university.faculty import Faculty
32from waeup.sirp.university.department import Department
33from waeup.sirp.students.batching import (
34    StudentProcessor, StudentStudyCourseProcessor,
35    StudentStudyLevelProcessor, CourseTicketProcessor,
36    StudentOnlinePaymentProcessor)
37from waeup.sirp.students.student import Student
38from waeup.sirp.testing import FunctionalLayer, FunctionalTestCase
39from waeup.sirp.interfaces import IBatchProcessor
40
41STUDENT_SAMPLE_DATA = open(
42    os.path.join(os.path.dirname(__file__), 'sample_student_data.csv'),
43    'rb').read()
44
45STUDENT_HEADER_FIELDS = STUDENT_SAMPLE_DATA.split(
46    '\n')[0].split(',')
47
48STUDENT_SAMPLE_DATA_UPDATE = open(
49    os.path.join(os.path.dirname(__file__), 'sample_student_data_update.csv'),
50    'rb').read()
51
52STUDENT_HEADER_FIELDS_UPDATE = STUDENT_SAMPLE_DATA_UPDATE.split(
53    '\n')[0].split(',')
54
55STUDENT_SAMPLE_DATA_UPDATE2 = open(
56    os.path.join(os.path.dirname(__file__), 'sample_student_data_update2.csv'),
57    'rb').read()
58
59STUDENT_HEADER_FIELDS_UPDATE2 = STUDENT_SAMPLE_DATA_UPDATE2.split(
60    '\n')[0].split(',')
61
62STUDYCOURSE_SAMPLE_DATA = open(
63    os.path.join(os.path.dirname(__file__), 'sample_studycourse_data.csv'),
64    'rb').read()
65
66STUDYCOURSE_HEADER_FIELDS = STUDYCOURSE_SAMPLE_DATA.split(
67    '\n')[0].split(',')
68
69STUDENT_SAMPLE_DATA_MIGRATION = open(
70    os.path.join(os.path.dirname(__file__), 'sample_student_data_migration.csv'),
71    'rb').read()
72
73STUDENT_HEADER_FIELDS_MIGRATION = STUDENT_SAMPLE_DATA_MIGRATION.split(
74    '\n')[0].split(',')
75
76STUDYLEVEL_SAMPLE_DATA = open(
77    os.path.join(os.path.dirname(__file__), 'sample_studylevel_data.csv'),
78    'rb').read()
79
80STUDYLEVEL_HEADER_FIELDS = STUDYLEVEL_SAMPLE_DATA.split(
81    '\n')[0].split(',')
82
83COURSETICKET_SAMPLE_DATA = open(
84    os.path.join(os.path.dirname(__file__), 'sample_courseticket_data.csv'),
85    'rb').read()
86
87COURSETICKET_HEADER_FIELDS = COURSETICKET_SAMPLE_DATA.split(
88    '\n')[0].split(',')
89
90PAYMENT_SAMPLE_DATA = open(
91    os.path.join(os.path.dirname(__file__), 'sample_payment_data.csv'),
92    'rb').read()
93
94PAYMENT_HEADER_FIELDS = PAYMENT_SAMPLE_DATA.split(
95    '\n')[0].split(',')
96
97class StudentImporterTest(FunctionalTestCase):
98
99    layer = FunctionalLayer
100
101    def setUp(self):
102        super(StudentImporterTest, self).setUp()
103        # Setup a sample site for each test
104        app = University()
105        self.dc_root = tempfile.mkdtemp()
106        app['datacenter'].setStoragePath(self.dc_root)
107
108        # Prepopulate the ZODB...
109        self.getRootFolder()['app'] = app
110        # we add the site immediately after creation to the
111        # ZODB. Catalogs and other local utilities are not setup
112        # before that step.
113        self.app = self.getRootFolder()['app']
114        # Set site here. Some of the following setup code might need
115        # to access grok.getSite() and should get our new app then
116        setSite(app)
117
118        # Add student with subobjects
119        student = Student()
120        student.firstname = u'Anna'
121        student.lastname = u'Tester'
122        student.reg_number = u'123'
123        student.matric_number = u'234'
124        self.app['students'].addStudent(student)
125        self.student = self.app['students'][student.student_id]
126        self.importer = StudentProcessor()
127        self.workdir = tempfile.mkdtemp()
128        self.csv_file = os.path.join(self.workdir, 'sample_student_data.csv')
129        self.csv_file_update = os.path.join(
130            self.workdir, 'sample_student_data_update.csv')
131        self.csv_file_update2 = os.path.join(
132            self.workdir, 'sample_student_data_update2.csv')
133        self.csv_file_migration = os.path.join(
134            self.workdir, 'sample_student_data_migration.csv')
135        open(self.csv_file, 'wb').write(STUDENT_SAMPLE_DATA)
136        open(self.csv_file_update, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE)
137        open(self.csv_file_update2, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE2)
138        open(self.csv_file_migration, 'wb').write(STUDENT_SAMPLE_DATA_MIGRATION)
139
140    def tearDown(self):
141        super(StudentImporterTest, self).tearDown()
142        shutil.rmtree(self.workdir)
143        shutil.rmtree(self.dc_root)
144        clearSite()
145        return
146
147    def test_interface(self):
148        # Make sure we fulfill the interface contracts.
149        assert verifyObject(IBatchProcessor, self.importer) is True
150        assert verifyClass(
151            IBatchProcessor, StudentProcessor) is True
152
153    def test_parentsExist(self):
154        self.assertFalse(self.importer.parentsExist(None, dict()))
155        self.assertTrue(self.importer.parentsExist(None, self.app))
156
157    def test_entryExists(self):
158        assert self.importer.entryExists(
159            dict(student_id='ID_NONE'), self.app) is False
160        assert self.importer.entryExists(
161            dict(reg_number='123'), self.app) is True
162
163    def test_getParent(self):
164        parent = self.importer.getParent(None, self.app)
165        assert parent is self.app['students']
166
167    def test_getEntry(self):
168        assert self.importer.getEntry(
169            dict(student_id='ID_NONE'), self.app) is None
170        assert self.importer.getEntry(
171            dict(student_id=self.student.student_id), self.app) is self.student
172
173    def test_addEntry(self):
174        new_student = Student()
175        self.importer.addEntry(
176            new_student, dict(), self.app)
177        assert len(self.app['students'].keys()) == 2
178
179    def test_checkConversion(self):
180        errs, inv_errs, conv_dict = self.importer.checkConversion(
181            dict(reg_number='1', reg_state='admitted'))
182        self.assertEqual(len(errs),0)
183        errs, inv_errs, conv_dict = self.importer.checkConversion(
184            dict(reg_number='1', reg_state=''))
185        self.assertEqual(len(errs),1)
186        self.assertTrue(('reg_state', 'no value provided') in errs)
187        errs, inv_errs, conv_dict = self.importer.checkConversion(
188            dict(reg_number='1', reg_state='nonsense'))
189        self.assertEqual(len(errs),1)
190        self.assertTrue(('reg_state', 'not allowed') in errs)
191
192    def test_delEntry(self):
193        assert self.student.student_id in self.app['students'].keys()
194        self.importer.delEntry(
195            dict(reg_number=self.student.reg_number), self.app)
196        assert self.student.student_id not in self.app['students'].keys()
197
198    def test_import(self):
199        num, num_warns, fin_file, fail_file = self.importer.doImport(
200            self.csv_file, STUDENT_HEADER_FIELDS)
201        self.assertEqual(num_warns,0)
202        assert len(self.app['students'].keys()) == 4
203        shutil.rmtree(os.path.dirname(fin_file))
204
205    def test_import_update(self):
206        num, num_warns, fin_file, fail_file = self.importer.doImport(
207            self.csv_file, STUDENT_HEADER_FIELDS)
208        shutil.rmtree(os.path.dirname(fin_file))
209        num, num_warns, fin_file, fail_file = self.importer.doImport(
210            self.csv_file_update, STUDENT_HEADER_FIELDS_UPDATE, 'update')
211        self.assertEqual(num_warns,0)
212        shutil.rmtree(os.path.dirname(fin_file))
213
214    def test_import_update2(self):
215        num, num_warns, fin_file, fail_file = self.importer.doImport(
216            self.csv_file, STUDENT_HEADER_FIELDS)
217        shutil.rmtree(os.path.dirname(fin_file))
218        num, num_warns, fin_file, fail_file = self.importer.doImport(
219            self.csv_file_update2, STUDENT_HEADER_FIELDS_UPDATE2, 'update')
220        self.assertEqual(num_warns,0)
221        shutil.rmtree(os.path.dirname(fin_file))
222
223    def test_import_remove(self):
224        num, num_warns, fin_file, fail_file = self.importer.doImport(
225            self.csv_file, STUDENT_HEADER_FIELDS)
226        shutil.rmtree(os.path.dirname(fin_file))
227        num, num_warns, fin_file, fail_file = self.importer.doImport(
228            self.csv_file_update, STUDENT_HEADER_FIELDS_UPDATE, 'remove')
229        self.assertEqual(num_warns,0)
230        shutil.rmtree(os.path.dirname(fin_file))
231
232    def test_import_migration_data(self):
233        num, num_warns, fin_file, fail_file = self.importer.doImport(
234            self.csv_file_migration, STUDENT_HEADER_FIELDS_MIGRATION)
235        self.assertEqual(num_warns,2)
236        assert len(self.app['students'].keys()) == 4
237        self.assertTrue('A123456' in self.app['students'].keys())
238        self.assertEqual(self.app['students']['A123456'].state,'clearance started')
239        self.assertEqual(self.app['students']['A123456'].date_of_birth,
240            datetime.date(1990, 1, 2))
241        self.assertFalse(self.app['students']['A123456'].clearance_locked)
242        self.assertEqual(self.app['students']['B123456'].state,'cleared')
243        self.assertEqual(self.app['students']['B123456'].date_of_birth,
244            datetime.date(1990, 1, 3))
245        self.assertTrue(self.app['students']['B123456'].clearance_locked)
246        history = ' '.join(self.app['students']['A123456'].history.messages)
247        self.assertTrue(
248            "State 'clearance started' set by system" in history)
249        shutil.rmtree(os.path.dirname(fin_file))
250
251
252class StudentStudyCourseImporterTest(FunctionalTestCase):
253
254    layer = FunctionalLayer
255
256    def setUp(self):
257        super(StudentStudyCourseImporterTest, self).setUp()
258        self.dc_root = tempfile.mkdtemp()
259        self.workdir = tempfile.mkdtemp()
260        app = University()
261        app['datacenter'].setStoragePath(self.dc_root)
262        self.getRootFolder()['app'] = app
263        self.app = self.getRootFolder()['app']
264        setSite(app)
265
266        # Import students with subobjects
267        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
268        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
269        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
270            student_file, STUDENT_HEADER_FIELDS)
271        shutil.rmtree(os.path.dirname(fin_file))
272
273        # Populate university
274        self.certificate = createObject('waeup.Certificate')
275        self.certificate.code = 'CERT1'
276        self.certificate.application_category = 'basic'
277        self.certificate.start_level = 200
278        self.certificate.end_level = 500
279        self.app['faculties']['fac1'] = Faculty()
280        self.app['faculties']['fac1']['dep1'] = Department()
281        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
282            self.certificate)
283
284        self.importer = StudentStudyCourseProcessor()
285        self.csv_file = os.path.join(
286            self.workdir, 'sample_studycourse_data.csv')
287        open(self.csv_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
288        return
289
290    def tearDown(self):
291        super(StudentStudyCourseImporterTest, self).tearDown()
292        shutil.rmtree(self.workdir)
293        shutil.rmtree(self.dc_root)
294        clearSite()
295        return
296
297    def test_interface(self):
298        # Make sure we fulfill the interface contracts.
299        assert verifyObject(IBatchProcessor, self.importer) is True
300        assert verifyClass(
301            IBatchProcessor, StudentStudyCourseProcessor) is True
302
303    def test_entryExists(self):
304        assert self.importer.entryExists(
305            dict(reg_number='REG_NONE'), self.app) is False
306        assert self.importer.entryExists(
307            dict(reg_number='1'), self.app) is True
308
309    def test_getEntry(self):
310        student = self.importer.getEntry(
311            dict(reg_number='1'), self.app).__parent__
312        self.assertEqual(student.reg_number,'1')
313
314    def test_checkConversion(self):
315        errs, inv_errs, conv_dict = self.importer.checkConversion(
316            dict(reg_number='1', certificate='CERT1', current_level='200'))
317        self.assertEqual(len(errs),0)
318        errs, inv_errs, conv_dict = self.importer.checkConversion(
319            dict(reg_number='1', certificate='CERT999'))
320        self.assertEqual(len(errs),1)
321        self.assertTrue(('certificate', u'Invalid value') in errs)
322        errs, inv_errs, conv_dict = self.importer.checkConversion(
323            dict(reg_number='1', certificate='CERT1', current_level='100'))
324        self.assertEqual(len(errs),1)
325        self.assertTrue(('current_level','not in range') in errs)
326        # If we import only current_level, no conversion checking is done.
327        errs, inv_errs, conv_dict = self.importer.checkConversion(
328            dict(reg_number='1', current_level='100'))
329        self.assertEqual(len(errs),0)
330
331    def test_import(self):
332        num, num_warns, fin_file, fail_file = self.importer.doImport(
333            self.csv_file, STUDYCOURSE_HEADER_FIELDS,'update')
334        self.assertEqual(num_warns,1)
335        studycourse = self.importer.getEntry(dict(reg_number='1'), self.app)
336        self.assertEqual(studycourse.certificate.code, u'CERT1')
337        shutil.rmtree(os.path.dirname(fin_file))
338
339class StudentStudyLevelImporterTest(FunctionalTestCase):
340
341    layer = FunctionalLayer
342
343    def setUp(self):
344        super(StudentStudyLevelImporterTest, self).setUp()
345        self.dc_root = tempfile.mkdtemp()
346        self.workdir = tempfile.mkdtemp()
347        app = University()
348        app['datacenter'].setStoragePath(self.dc_root)
349        self.getRootFolder()['app'] = app
350        self.app = self.getRootFolder()['app']
351        setSite(app)
352
353        # Import students with subobjects
354        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
355        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
356        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
357            student_file, STUDENT_HEADER_FIELDS)
358        shutil.rmtree(os.path.dirname(fin_file))
359
360        # Populate university
361        self.certificate = createObject('waeup.Certificate')
362        self.certificate.code = 'CERT1'
363        self.certificate.application_category = 'basic'
364        self.certificate.start_level = 200
365        self.certificate.end_level = 500
366        self.app['faculties']['fac1'] = Faculty()
367        self.app['faculties']['fac1']['dep1'] = Department()
368        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
369            self.certificate)
370
371        # Update study courses
372        studycourse_file = os.path.join(
373            self.workdir, 'sample_studycourse_data.csv')
374        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
375        importer = StudentStudyCourseProcessor()
376        num, num_warns, fin_file, fail_file = importer.doImport(
377            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
378        shutil.rmtree(os.path.dirname(fin_file))
379
380        self.importer = StudentStudyLevelProcessor()
381        self.csv_file = os.path.join(
382            self.workdir, 'sample_studylevel_data.csv')
383        open(self.csv_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
384
385    def tearDown(self):
386        super(StudentStudyLevelImporterTest, self).tearDown()
387        shutil.rmtree(self.workdir)
388        shutil.rmtree(self.dc_root)
389        clearSite()
390        return
391
392    def test_interface(self):
393        # Make sure we fulfill the interface contracts.
394        assert verifyObject(IBatchProcessor, self.importer) is True
395        assert verifyClass(
396            IBatchProcessor, StudentStudyLevelProcessor) is True
397
398    def test_checkConversion(self):
399        errs, inv_errs, conv_dict = self.importer.checkConversion(
400            dict(reg_number='1', level='220'))
401        self.assertEqual(len(errs),0)
402        errs, inv_errs, conv_dict = self.importer.checkConversion(
403            dict(reg_number='1', level='900'))
404        self.assertEqual(len(errs),1)
405        self.assertTrue(('level','no valid integer') in errs)
406        errs, inv_errs, conv_dict = self.importer.checkConversion(
407            dict(reg_number='1', level='xyz'))
408        self.assertEqual(len(errs),1)
409        self.assertTrue(('level','no integer') in errs)
410
411    def test_import(self):
412        num, num_warns, fin_file, fail_file = self.importer.doImport(
413            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
414        self.assertEqual(num_warns,2)
415        assert self.importer.entryExists(
416            dict(reg_number='1', level='100'), self.app) is True
417        studylevel = self.importer.getEntry(
418            dict(reg_number='1', level='100'), self.app)
419        self.assertEqual(studylevel.__parent__.certificate.code, u'CERT1')
420        self.assertEqual(studylevel.level_session, 2008)
421        self.assertEqual(studylevel.level_verdict, 'A')
422        self.assertEqual(studylevel.level, 100)
423        shutil.rmtree(os.path.dirname(fin_file))
424       
425
426class CourseTicketImporterTest(FunctionalTestCase):
427
428    layer = FunctionalLayer
429
430    def setUp(self):
431        super(CourseTicketImporterTest, self).setUp()
432        self.dc_root = tempfile.mkdtemp()
433        self.workdir = tempfile.mkdtemp()
434        app = University()
435        app['datacenter'].setStoragePath(self.dc_root)
436        self.getRootFolder()['app'] = app
437        self.app = self.getRootFolder()['app']
438        setSite(app)
439
440        # Import students with subobjects
441        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
442        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
443        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
444            student_file, STUDENT_HEADER_FIELDS)
445        shutil.rmtree(os.path.dirname(fin_file))
446
447        # Populate university
448        self.certificate = createObject('waeup.Certificate')
449        self.certificate.code = 'CERT1'
450        self.certificate.application_category = 'basic'
451        self.certificate.start_level = 200
452        self.certificate.end_level = 500
453        self.app['faculties']['fac1'] = Faculty()
454        self.app['faculties']['fac1']['dep1'] = Department()
455        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
456            self.certificate)
457        self.course = createObject('waeup.Course')
458        self.course.code = 'COURSE1'
459        self.course.semester = 1
460        self.course.credits = 10
461        self.course.passmark = 40
462        self.app['faculties']['fac1']['dep1'].courses.addCourse(
463            self.course)
464        self.app['faculties']['fac1']['dep1'].certificates['CERT1'].addCourseRef(
465            self.course, level=100)
466
467        # Update study courses
468        studycourse_file = os.path.join(
469            self.workdir, 'sample_studycourse_data.csv')
470        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
471        importer = StudentStudyCourseProcessor()
472        num, num_warns, fin_file, fail_file = importer.doImport(
473            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
474        shutil.rmtree(os.path.dirname(fin_file))
475
476        # Import study levels
477        importer = StudentStudyLevelProcessor()
478        studylevel_file = os.path.join(
479            self.workdir, 'sample_studylevel_data.csv')
480        open(studylevel_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
481        num, num_warns, fin_file, fail_file = importer.doImport(
482            studylevel_file, STUDYLEVEL_HEADER_FIELDS,'create')
483        shutil.rmtree(os.path.dirname(fin_file))
484
485        self.importer = CourseTicketProcessor()
486        self.csv_file = os.path.join(
487            self.workdir, 'sample_courseticket_data.csv')
488        open(self.csv_file, 'wb').write(COURSETICKET_SAMPLE_DATA)
489
490    def tearDown(self):
491        super(CourseTicketImporterTest, self).tearDown()
492        shutil.rmtree(self.workdir)
493        shutil.rmtree(self.dc_root)
494        clearSite()
495        return
496
497    def test_interface(self):
498        # Make sure we fulfill the interface contracts.
499        assert verifyObject(IBatchProcessor, self.importer) is True
500        assert verifyClass(
501            IBatchProcessor, CourseTicketProcessor) is True
502
503    def test_checkConversion(self):
504        errs, inv_errs, conv_dict = self.importer.checkConversion(
505            dict(reg_number='1', code='COURSE1', level='220'))
506        self.assertEqual(len(errs),0)
507        errs, inv_errs, conv_dict = self.importer.checkConversion(
508            dict(reg_number='1', code='COURSE2', level='220'))
509        self.assertEqual(len(errs),1)
510        self.assertTrue(('code','non-existent') in errs)
511
512    def test_import(self):
513
514        num, num_warns, fin_file, fail_file = self.importer.doImport(
515            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
516
517        self.assertEqual(num_warns,2)
518        assert self.importer.entryExists(
519            dict(reg_number='1', level='100', code='COURSE1'), self.app) is True
520        courseticket = self.importer.getEntry(
521            dict(reg_number='1', level='100', code='COURSE1'), self.app)
522        self.assertEqual(courseticket.__parent__.__parent__.certificate.code, u'CERT1')
523        self.assertEqual(courseticket.score, 1)
524        self.assertEqual(courseticket.core_or_elective, True)
525        self.assertEqual(courseticket.fcode, 'NA')
526        self.assertEqual(courseticket.dcode, 'NA')
527        self.assertEqual(courseticket.code, 'COURSE1')
528        self.assertEqual(courseticket.title, 'Unnamed Course')
529        self.assertEqual(courseticket.credits, 10)
530        self.assertEqual(courseticket.passmark, 40)
531        self.assertEqual(courseticket.semester, 1)
532        #import pdb; pdb.set_trace()
533        shutil.rmtree(os.path.dirname(fin_file))
534
535class PaymentImporterTest(FunctionalTestCase):
536
537    layer = FunctionalLayer
538
539    def setUp(self):
540        super(PaymentImporterTest, self).setUp()
541        self.dc_root = tempfile.mkdtemp()
542        self.workdir = tempfile.mkdtemp()
543        app = University()
544        app['datacenter'].setStoragePath(self.dc_root)
545        self.getRootFolder()['app'] = app
546        self.app = self.getRootFolder()['app']
547        setSite(app)
548
549        # Add student with payment
550        student = Student()
551        student.firstname = u'Anna'
552        student.lastname = u'Tester'
553        student.reg_number = u'123'
554        student.matric_number = u'234'
555        self.app['students'].addStudent(student)
556        self.student = self.app['students'][student.student_id]
557        payment = createObject(u'waeup.StudentOnlinePayment')
558        payment.p_id = 'p123'
559        self.student['payments'][payment.p_id] = payment
560
561        # Import students with subobjects
562        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
563        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
564        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
565            student_file, STUDENT_HEADER_FIELDS)
566        shutil.rmtree(os.path.dirname(fin_file))
567
568        self.importer = StudentOnlinePaymentProcessor()
569        self.csv_file = os.path.join(
570            self.workdir, 'sample_payment_data.csv')
571        open(self.csv_file, 'wb').write(PAYMENT_SAMPLE_DATA)
572
573    def tearDown(self):
574        super(PaymentImporterTest, self).tearDown()
575        shutil.rmtree(self.workdir)
576        shutil.rmtree(self.dc_root)
577        clearSite()
578        return
579
580    def test_interface(self):
581        # Make sure we fulfill the interface contracts.
582        assert verifyObject(IBatchProcessor, self.importer) is True
583        assert verifyClass(
584            IBatchProcessor, StudentOnlinePaymentProcessor) is True
585
586    def test_getEntry(self):
587        assert self.importer.getEntry(
588            dict(student_id='ID_NONE', p_id='nonsense'), self.app) is None
589        assert self.importer.getEntry(
590            dict(student_id=self.student.student_id, p_id='p123'),
591            self.app) is self.student['payments']['p123']
592        assert self.importer.getEntry(
593            dict(student_id=self.student.student_id, p_id='XXXXXX123'),
594            self.app) is self.student['payments']['p123']
595
596    def test_addEntry(self):
597        self.assertEqual(len(self.student['payments'].keys()),1)
598        payment1 = createObject(u'waeup.StudentOnlinePayment')
599        payment1.p_id = 'p234'
600        self.importer.addEntry(
601            payment1, dict(student_id=self.student.student_id, p_id='p234'),
602            self.app)
603        self.assertEqual(len(self.student['payments'].keys()),2)
604        self.assertEqual(self.student['payments']['p234'].p_id, 'p234')
605        payment2 = createObject(u'waeup.StudentOnlinePayment')
606        payment1.p_id = 'nonsense'
607        # payment1.p_id will be replaced if p_id doesn't start with 'p'
608        self.importer.addEntry(
609            payment2, dict(student_id=self.student.student_id, p_id='XXXXXX456'),
610            self.app)
611        self.assertEqual(len(self.student['payments'].keys()),3)
612        self.assertEqual(self.student['payments']['p456'].p_id, 'p456')
613
614    def test_checkConversion(self):
615        errs, inv_errs, conv_dict = self.importer.checkConversion(
616            dict(reg_number='1', p_id='3816951266236341955'))
617        self.assertEqual(len(errs),0)
618        errs, inv_errs, conv_dict = self.importer.checkConversion(
619            dict(reg_number='1', p_id='p1266236341955'))
620        self.assertEqual(len(errs),0)
621        errs, inv_errs, conv_dict = self.importer.checkConversion(
622            dict(reg_number='1', p_id='nonsense'))
623        self.assertEqual(len(errs),1)
624        timestamp = "%d" % int(time()*1000)
625        p_id = "p%s" % timestamp
626        errs, inv_errs, conv_dict = self.importer.checkConversion(
627            dict(reg_number='1', p_id=p_id))
628        self.assertEqual(len(errs),0)
629
630    def test_import(self):
631        num, num_warns, fin_file, fail_file = self.importer.doImport(
632            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
633        self.assertEqual(num_warns,0)
634        payment = self.importer.getEntry(dict(reg_number='1',
635            p_id='p1290797973744'), self.app)
636        self.assertEqual(payment.p_id, 'p1290797973744')
637        cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S")
638        self.assertEqual(cdate, "2010-11-26 19:59:33")
639        shutil.rmtree(os.path.dirname(fin_file))
640
641def test_suite():
642    suite = unittest.TestSuite()
643    for testcase in [
644        StudentImporterTest,StudentStudyCourseImporterTest,
645        StudentStudyLevelImporterTest,CourseTicketImporterTest,
646        PaymentImporterTest,]:
647        suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
648                testcase
649                )
650        )
651    return suite
652
653
Note: See TracBrowser for help on using the repository browser.