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

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

Enable mixed imports of rows with or without student_id.
  • Property svn:keywords set to Id
File size: 26.5 KB
Line 
1## $Id: test_batching.py 7643 2012-02-14 09:32:28Z 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        self.assertEqual(self.app['students']['X666666'].reg_number,'1')
204        shutil.rmtree(os.path.dirname(fin_file))
205
206    def test_import_update(self):
207        num, num_warns, fin_file, fail_file = self.importer.doImport(
208            self.csv_file, STUDENT_HEADER_FIELDS)
209        shutil.rmtree(os.path.dirname(fin_file))
210        num, num_warns, fin_file, fail_file = self.importer.doImport(
211            self.csv_file_update, STUDENT_HEADER_FIELDS_UPDATE, 'update')
212        self.assertEqual(num_warns,0)
213        shutil.rmtree(os.path.dirname(fin_file))
214
215    def test_import_update2(self):
216        num, num_warns, fin_file, fail_file = self.importer.doImport(
217            self.csv_file, STUDENT_HEADER_FIELDS)
218        shutil.rmtree(os.path.dirname(fin_file))
219        num, num_warns, fin_file, fail_file = self.importer.doImport(
220            self.csv_file_update2, STUDENT_HEADER_FIELDS_UPDATE2, 'update')
221        self.assertEqual(num_warns,0)
222        shutil.rmtree(os.path.dirname(fin_file))
223
224    def test_import_remove(self):
225        num, num_warns, fin_file, fail_file = self.importer.doImport(
226            self.csv_file, STUDENT_HEADER_FIELDS)
227        shutil.rmtree(os.path.dirname(fin_file))
228        num, num_warns, fin_file, fail_file = self.importer.doImport(
229            self.csv_file_update, STUDENT_HEADER_FIELDS_UPDATE, 'remove')
230        self.assertEqual(num_warns,0)
231        shutil.rmtree(os.path.dirname(fin_file))
232
233    def test_import_migration_data(self):
234        num, num_warns, fin_file, fail_file = self.importer.doImport(
235            self.csv_file_migration, STUDENT_HEADER_FIELDS_MIGRATION)
236        self.assertEqual(num_warns,2)
237        assert len(self.app['students'].keys()) == 4
238        self.assertTrue('A123456' in self.app['students'].keys())
239        self.assertEqual(self.app['students']['A123456'].state,'clearance started')
240        self.assertEqual(self.app['students']['A123456'].date_of_birth,
241            datetime.date(1990, 1, 2))
242        self.assertFalse(self.app['students']['A123456'].clearance_locked)
243        self.assertEqual(self.app['students']['B123456'].state,'cleared')
244        self.assertEqual(self.app['students']['B123456'].date_of_birth,
245            datetime.date(1990, 1, 3))
246        self.assertTrue(self.app['students']['B123456'].clearance_locked)
247        history = ' '.join(self.app['students']['A123456'].history.messages)
248        self.assertTrue(
249            "State 'clearance started' set by system" in history)
250        shutil.rmtree(os.path.dirname(fin_file))
251
252
253class StudentStudyCourseImporterTest(FunctionalTestCase):
254
255    layer = FunctionalLayer
256
257    def setUp(self):
258        super(StudentStudyCourseImporterTest, self).setUp()
259        self.dc_root = tempfile.mkdtemp()
260        self.workdir = tempfile.mkdtemp()
261        app = University()
262        app['datacenter'].setStoragePath(self.dc_root)
263        self.getRootFolder()['app'] = app
264        self.app = self.getRootFolder()['app']
265        setSite(app)
266
267        # Import students with subobjects
268        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
269        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
270        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
271            student_file, STUDENT_HEADER_FIELDS)
272        shutil.rmtree(os.path.dirname(fin_file))
273
274        # Populate university
275        self.certificate = createObject('waeup.Certificate')
276        self.certificate.code = 'CERT1'
277        self.certificate.application_category = 'basic'
278        self.certificate.start_level = 200
279        self.certificate.end_level = 500
280        self.app['faculties']['fac1'] = Faculty()
281        self.app['faculties']['fac1']['dep1'] = Department()
282        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
283            self.certificate)
284
285        self.importer = StudentStudyCourseProcessor()
286        self.csv_file = os.path.join(
287            self.workdir, 'sample_studycourse_data.csv')
288        open(self.csv_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
289        return
290
291    def tearDown(self):
292        super(StudentStudyCourseImporterTest, self).tearDown()
293        shutil.rmtree(self.workdir)
294        shutil.rmtree(self.dc_root)
295        clearSite()
296        return
297
298    def test_interface(self):
299        # Make sure we fulfill the interface contracts.
300        assert verifyObject(IBatchProcessor, self.importer) is True
301        assert verifyClass(
302            IBatchProcessor, StudentStudyCourseProcessor) is True
303
304    def test_entryExists(self):
305        assert self.importer.entryExists(
306            dict(reg_number='REG_NONE'), self.app) is False
307        assert self.importer.entryExists(
308            dict(reg_number='1'), self.app) is True
309
310    def test_getEntry(self):
311        student = self.importer.getEntry(
312            dict(reg_number='1'), self.app).__parent__
313        self.assertEqual(student.reg_number,'1')
314
315    def test_checkConversion(self):
316        errs, inv_errs, conv_dict = self.importer.checkConversion(
317            dict(reg_number='1', certificate='CERT1', current_level='200'))
318        self.assertEqual(len(errs),0)
319        errs, inv_errs, conv_dict = self.importer.checkConversion(
320            dict(reg_number='1', certificate='CERT999'))
321        self.assertEqual(len(errs),1)
322        self.assertTrue(('certificate', u'Invalid value') in errs)
323        errs, inv_errs, conv_dict = self.importer.checkConversion(
324            dict(reg_number='1', certificate='CERT1', current_level='100'))
325        self.assertEqual(len(errs),1)
326        self.assertTrue(('current_level','not in range') in errs)
327        # If we import only current_level, no conversion checking is done.
328        errs, inv_errs, conv_dict = self.importer.checkConversion(
329            dict(reg_number='1', current_level='100'))
330        self.assertEqual(len(errs),0)
331
332    def test_import(self):
333        num, num_warns, fin_file, fail_file = self.importer.doImport(
334            self.csv_file, STUDYCOURSE_HEADER_FIELDS,'update')
335        self.assertEqual(num_warns,1)
336        studycourse = self.importer.getEntry(dict(reg_number='1'), self.app)
337        self.assertEqual(studycourse.certificate.code, u'CERT1')
338        shutil.rmtree(os.path.dirname(fin_file))
339
340class StudentStudyLevelImporterTest(FunctionalTestCase):
341
342    layer = FunctionalLayer
343
344    def setUp(self):
345        super(StudentStudyLevelImporterTest, self).setUp()
346        self.dc_root = tempfile.mkdtemp()
347        self.workdir = tempfile.mkdtemp()
348        app = University()
349        app['datacenter'].setStoragePath(self.dc_root)
350        self.getRootFolder()['app'] = app
351        self.app = self.getRootFolder()['app']
352        setSite(app)
353
354        # Import students with subobjects
355        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
356        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
357        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
358            student_file, STUDENT_HEADER_FIELDS)
359        shutil.rmtree(os.path.dirname(fin_file))
360
361        # Populate university
362        self.certificate = createObject('waeup.Certificate')
363        self.certificate.code = 'CERT1'
364        self.certificate.application_category = 'basic'
365        self.certificate.start_level = 200
366        self.certificate.end_level = 500
367        self.app['faculties']['fac1'] = Faculty()
368        self.app['faculties']['fac1']['dep1'] = Department()
369        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
370            self.certificate)
371
372        # Update study courses
373        studycourse_file = os.path.join(
374            self.workdir, 'sample_studycourse_data.csv')
375        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
376        importer = StudentStudyCourseProcessor()
377        num, num_warns, fin_file, fail_file = importer.doImport(
378            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
379        shutil.rmtree(os.path.dirname(fin_file))
380
381        self.importer = StudentStudyLevelProcessor()
382        self.csv_file = os.path.join(
383            self.workdir, 'sample_studylevel_data.csv')
384        open(self.csv_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
385
386    def tearDown(self):
387        super(StudentStudyLevelImporterTest, self).tearDown()
388        shutil.rmtree(self.workdir)
389        shutil.rmtree(self.dc_root)
390        clearSite()
391        return
392
393    def test_interface(self):
394        # Make sure we fulfill the interface contracts.
395        assert verifyObject(IBatchProcessor, self.importer) is True
396        assert verifyClass(
397            IBatchProcessor, StudentStudyLevelProcessor) is True
398
399    def test_checkConversion(self):
400        errs, inv_errs, conv_dict = self.importer.checkConversion(
401            dict(reg_number='1', level='220'))
402        self.assertEqual(len(errs),0)
403        errs, inv_errs, conv_dict = self.importer.checkConversion(
404            dict(reg_number='1', level='900'))
405        self.assertEqual(len(errs),1)
406        self.assertTrue(('level','no valid integer') in errs)
407        errs, inv_errs, conv_dict = self.importer.checkConversion(
408            dict(reg_number='1', level='xyz'))
409        self.assertEqual(len(errs),1)
410        self.assertTrue(('level','no integer') in errs)
411
412    def test_import(self):
413        num, num_warns, fin_file, fail_file = self.importer.doImport(
414            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
415        self.assertEqual(num_warns,2)
416        assert self.importer.entryExists(
417            dict(reg_number='1', level='100'), self.app) is True
418        studylevel = self.importer.getEntry(
419            dict(reg_number='1', level='100'), self.app)
420        self.assertEqual(studylevel.__parent__.certificate.code, u'CERT1')
421        self.assertEqual(studylevel.level_session, 2008)
422        self.assertEqual(studylevel.level_verdict, 'A')
423        self.assertEqual(studylevel.level, 100)
424        shutil.rmtree(os.path.dirname(fin_file))
425       
426
427class CourseTicketImporterTest(FunctionalTestCase):
428
429    layer = FunctionalLayer
430
431    def setUp(self):
432        super(CourseTicketImporterTest, self).setUp()
433        self.dc_root = tempfile.mkdtemp()
434        self.workdir = tempfile.mkdtemp()
435        app = University()
436        app['datacenter'].setStoragePath(self.dc_root)
437        self.getRootFolder()['app'] = app
438        self.app = self.getRootFolder()['app']
439        setSite(app)
440
441        # Import students with subobjects
442        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
443        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
444        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
445            student_file, STUDENT_HEADER_FIELDS)
446        shutil.rmtree(os.path.dirname(fin_file))
447
448        # Populate university
449        self.certificate = createObject('waeup.Certificate')
450        self.certificate.code = 'CERT1'
451        self.certificate.application_category = 'basic'
452        self.certificate.start_level = 200
453        self.certificate.end_level = 500
454        self.app['faculties']['fac1'] = Faculty()
455        self.app['faculties']['fac1']['dep1'] = Department()
456        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
457            self.certificate)
458        self.course = createObject('waeup.Course')
459        self.course.code = 'COURSE1'
460        self.course.semester = 1
461        self.course.credits = 10
462        self.course.passmark = 40
463        self.app['faculties']['fac1']['dep1'].courses.addCourse(
464            self.course)
465        self.app['faculties']['fac1']['dep1'].certificates['CERT1'].addCourseRef(
466            self.course, level=100)
467
468        # Update study courses
469        studycourse_file = os.path.join(
470            self.workdir, 'sample_studycourse_data.csv')
471        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
472        importer = StudentStudyCourseProcessor()
473        num, num_warns, fin_file, fail_file = importer.doImport(
474            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
475        shutil.rmtree(os.path.dirname(fin_file))
476
477        # Import study levels
478        importer = StudentStudyLevelProcessor()
479        studylevel_file = os.path.join(
480            self.workdir, 'sample_studylevel_data.csv')
481        open(studylevel_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
482        num, num_warns, fin_file, fail_file = importer.doImport(
483            studylevel_file, STUDYLEVEL_HEADER_FIELDS,'create')
484        shutil.rmtree(os.path.dirname(fin_file))
485
486        self.importer = CourseTicketProcessor()
487        self.csv_file = os.path.join(
488            self.workdir, 'sample_courseticket_data.csv')
489        open(self.csv_file, 'wb').write(COURSETICKET_SAMPLE_DATA)
490
491    def tearDown(self):
492        super(CourseTicketImporterTest, self).tearDown()
493        shutil.rmtree(self.workdir)
494        shutil.rmtree(self.dc_root)
495        clearSite()
496        return
497
498    def test_interface(self):
499        # Make sure we fulfill the interface contracts.
500        assert verifyObject(IBatchProcessor, self.importer) is True
501        assert verifyClass(
502            IBatchProcessor, CourseTicketProcessor) is True
503
504    def test_checkConversion(self):
505        errs, inv_errs, conv_dict = self.importer.checkConversion(
506            dict(reg_number='1', code='COURSE1', level='220'))
507        self.assertEqual(len(errs),0)
508        errs, inv_errs, conv_dict = self.importer.checkConversion(
509            dict(reg_number='1', code='COURSE2', level='220'))
510        self.assertEqual(len(errs),1)
511        self.assertTrue(('code','non-existent') in errs)
512
513    def test_import(self):
514
515        num, num_warns, fin_file, fail_file = self.importer.doImport(
516            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
517
518        self.assertEqual(num_warns,2)
519        assert self.importer.entryExists(
520            dict(reg_number='1', level='100', code='COURSE1'), self.app) is True
521        courseticket = self.importer.getEntry(
522            dict(reg_number='1', level='100', code='COURSE1'), self.app)
523        self.assertEqual(courseticket.__parent__.__parent__.certificate.code, u'CERT1')
524        self.assertEqual(courseticket.score, 1)
525        self.assertEqual(courseticket.core_or_elective, True)
526        self.assertEqual(courseticket.fcode, 'NA')
527        self.assertEqual(courseticket.dcode, 'NA')
528        self.assertEqual(courseticket.code, 'COURSE1')
529        self.assertEqual(courseticket.title, 'Unnamed Course')
530        self.assertEqual(courseticket.credits, 10)
531        self.assertEqual(courseticket.passmark, 40)
532        self.assertEqual(courseticket.semester, 1)
533        #import pdb; pdb.set_trace()
534        shutil.rmtree(os.path.dirname(fin_file))
535
536class PaymentImporterTest(FunctionalTestCase):
537
538    layer = FunctionalLayer
539
540    def setUp(self):
541        super(PaymentImporterTest, self).setUp()
542        self.dc_root = tempfile.mkdtemp()
543        self.workdir = tempfile.mkdtemp()
544        app = University()
545        app['datacenter'].setStoragePath(self.dc_root)
546        self.getRootFolder()['app'] = app
547        self.app = self.getRootFolder()['app']
548        setSite(app)
549
550        # Add student with payment
551        student = Student()
552        student.firstname = u'Anna'
553        student.lastname = u'Tester'
554        student.reg_number = u'123'
555        student.matric_number = u'234'
556        self.app['students'].addStudent(student)
557        self.student = self.app['students'][student.student_id]
558        payment = createObject(u'waeup.StudentOnlinePayment')
559        payment.p_id = 'p123'
560        self.student['payments'][payment.p_id] = payment
561
562        # Import students with subobjects
563        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
564        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
565        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
566            student_file, STUDENT_HEADER_FIELDS)
567        shutil.rmtree(os.path.dirname(fin_file))
568
569        self.importer = StudentOnlinePaymentProcessor()
570        self.csv_file = os.path.join(
571            self.workdir, 'sample_payment_data.csv')
572        open(self.csv_file, 'wb').write(PAYMENT_SAMPLE_DATA)
573
574    def tearDown(self):
575        super(PaymentImporterTest, self).tearDown()
576        shutil.rmtree(self.workdir)
577        shutil.rmtree(self.dc_root)
578        clearSite()
579        return
580
581    def test_interface(self):
582        # Make sure we fulfill the interface contracts.
583        assert verifyObject(IBatchProcessor, self.importer) is True
584        assert verifyClass(
585            IBatchProcessor, StudentOnlinePaymentProcessor) is True
586
587    def test_getEntry(self):
588        assert self.importer.getEntry(
589            dict(student_id='ID_NONE', p_id='nonsense'), self.app) is None
590        assert self.importer.getEntry(
591            dict(student_id=self.student.student_id, p_id='p123'),
592            self.app) is self.student['payments']['p123']
593        assert self.importer.getEntry(
594            dict(student_id=self.student.student_id, p_id='XXXXXX123'),
595            self.app) is self.student['payments']['p123']
596
597    def test_addEntry(self):
598        self.assertEqual(len(self.student['payments'].keys()),1)
599        payment1 = createObject(u'waeup.StudentOnlinePayment')
600        payment1.p_id = 'p234'
601        self.importer.addEntry(
602            payment1, dict(student_id=self.student.student_id, p_id='p234'),
603            self.app)
604        self.assertEqual(len(self.student['payments'].keys()),2)
605        self.assertEqual(self.student['payments']['p234'].p_id, 'p234')
606        payment2 = createObject(u'waeup.StudentOnlinePayment')
607        payment1.p_id = 'nonsense'
608        # payment1.p_id will be replaced if p_id doesn't start with 'p'
609        self.importer.addEntry(
610            payment2, dict(student_id=self.student.student_id, p_id='XXXXXX456'),
611            self.app)
612        self.assertEqual(len(self.student['payments'].keys()),3)
613        self.assertEqual(self.student['payments']['p456'].p_id, 'p456')
614
615    def test_checkConversion(self):
616        errs, inv_errs, conv_dict = self.importer.checkConversion(
617            dict(reg_number='1', p_id='3816951266236341955'))
618        self.assertEqual(len(errs),0)
619        errs, inv_errs, conv_dict = self.importer.checkConversion(
620            dict(reg_number='1', p_id='p1266236341955'))
621        self.assertEqual(len(errs),0)
622        errs, inv_errs, conv_dict = self.importer.checkConversion(
623            dict(reg_number='1', p_id='nonsense'))
624        self.assertEqual(len(errs),1)
625        timestamp = "%d" % int(time()*1000)
626        p_id = "p%s" % timestamp
627        errs, inv_errs, conv_dict = self.importer.checkConversion(
628            dict(reg_number='1', p_id=p_id))
629        self.assertEqual(len(errs),0)
630
631    def test_import(self):
632        num, num_warns, fin_file, fail_file = self.importer.doImport(
633            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
634        self.assertEqual(num_warns,0)
635        payment = self.importer.getEntry(dict(reg_number='1',
636            p_id='p1290797973744'), self.app)
637        self.assertEqual(payment.p_id, 'p1290797973744')
638        cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S")
639        self.assertEqual(cdate, "2010-11-26 19:59:33")
640        shutil.rmtree(os.path.dirname(fin_file))
641
642def test_suite():
643    suite = unittest.TestSuite()
644    for testcase in [
645        StudentImporterTest,StudentStudyCourseImporterTest,
646        StudentStudyLevelImporterTest,CourseTicketImporterTest,
647        PaymentImporterTest,]:
648        suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
649                testcase
650                )
651        )
652    return suite
653
654
Note: See TracBrowser for help on using the repository browser.