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

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

Rename attribute core_or_elective to mandatory.

The following commands in debug mode will fix existing databases:

from zope.component import getUtility
from zope.catalog.interfaces import ICatalog
from zope.intid.interfaces import IIntIds
import transaction
sm = rootfutminna?.getSiteManager()
cat = sm.getUtility(ICatalog, name='certcourses_catalog')
cat.values()[0].documentCount()
results = cat.apply({'course_code':(None,None)})
uidutil = getUtility(IIntIds, context=cat)
for r in results:

... o = uidutil.getObject(r)
... o.mandatory = o.core_or_elective
...

cat = sm.getUtility(ICatalog, name='coursetickets_catalog')
cat.values()[0].documentCount()
results = cat.apply({'code':(None,None)})
uidutil = getUtility(IIntIds, context=cat)
for r in results:

... o = uidutil.getObject(r)
... o.mandatory = o.core_or_elective
...

transaction.commit()
  • Property svn:keywords set to Id
File size: 26.5 KB
Line 
1## $Id: test_batching.py 7665 2012-02-17 12:06:10Z 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.mandatory, 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        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.