1 | # -*- coding: utf-8 -*- |
---|
2 | ## $Id: test_batching.py 10842 2013-12-11 13:21:37Z henrik $ |
---|
3 | ## |
---|
4 | ## Copyright (C) 2011 Uli Fouquet & Henrik Bettermann |
---|
5 | ## This program is free software; you can redistribute it and/or modify |
---|
6 | ## it under the terms of the GNU General Public License as published by |
---|
7 | ## the Free Software Foundation; either version 2 of the License, or |
---|
8 | ## (at your option) any later version. |
---|
9 | ## |
---|
10 | ## This program is distributed in the hope that it will be useful, |
---|
11 | ## but WITHOUT ANY WARRANTY; without even the implied warranty of |
---|
12 | ## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
---|
13 | ## GNU General Public License for more details. |
---|
14 | ## |
---|
15 | ## You should have received a copy of the GNU General Public License |
---|
16 | ## along with this program; if not, write to the Free Software |
---|
17 | ## Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA |
---|
18 | ## |
---|
19 | """Unit tests for students-related data processors. |
---|
20 | """ |
---|
21 | import os |
---|
22 | import shutil |
---|
23 | import tempfile |
---|
24 | import unittest |
---|
25 | import datetime |
---|
26 | import grok |
---|
27 | from time import time |
---|
28 | from zope.event import notify |
---|
29 | from zope.component import createObject, queryUtility |
---|
30 | from zope.component.hooks import setSite, clearSite |
---|
31 | from zope.catalog.interfaces import ICatalog |
---|
32 | from zope.interface.verify import verifyClass, verifyObject |
---|
33 | from hurry.workflow.interfaces import IWorkflowState |
---|
34 | |
---|
35 | from waeup.kofa.app import University |
---|
36 | from waeup.kofa.interfaces import IBatchProcessor, FatalCSVError, IUserAccount |
---|
37 | from waeup.kofa.students.batching import ( |
---|
38 | StudentProcessor, StudentStudyCourseProcessor, |
---|
39 | StudentStudyLevelProcessor, CourseTicketProcessor, |
---|
40 | StudentOnlinePaymentProcessor, StudentVerdictProcessor) |
---|
41 | from waeup.kofa.students.payments import StudentOnlinePayment |
---|
42 | from waeup.kofa.students.student import Student |
---|
43 | from waeup.kofa.students.studylevel import StudentStudyLevel, CourseTicket |
---|
44 | from waeup.kofa.students.accommodation import BedTicket |
---|
45 | from waeup.kofa.testing import FunctionalLayer, FunctionalTestCase |
---|
46 | from waeup.kofa.university.faculty import Faculty |
---|
47 | from waeup.kofa.university.department import Department |
---|
48 | from waeup.kofa.hostels.hostel import Hostel, Bed, NOT_OCCUPIED |
---|
49 | |
---|
50 | |
---|
51 | STUDENT_SAMPLE_DATA = open( |
---|
52 | os.path.join(os.path.dirname(__file__), 'sample_student_data.csv'), |
---|
53 | 'rb').read() |
---|
54 | |
---|
55 | STUDENT_HEADER_FIELDS = STUDENT_SAMPLE_DATA.split( |
---|
56 | '\n')[0].split(',') |
---|
57 | |
---|
58 | STUDENT_SAMPLE_DATA_UPDATE = open( |
---|
59 | os.path.join(os.path.dirname(__file__), 'sample_student_data_update.csv'), |
---|
60 | 'rb').read() |
---|
61 | |
---|
62 | STUDENT_HEADER_FIELDS_UPDATE = STUDENT_SAMPLE_DATA_UPDATE.split( |
---|
63 | '\n')[0].split(',') |
---|
64 | |
---|
65 | STUDENT_SAMPLE_DATA_UPDATE2 = open( |
---|
66 | os.path.join(os.path.dirname(__file__), 'sample_student_data_update2.csv'), |
---|
67 | 'rb').read() |
---|
68 | |
---|
69 | STUDENT_HEADER_FIELDS_UPDATE2 = STUDENT_SAMPLE_DATA_UPDATE2.split( |
---|
70 | '\n')[0].split(',') |
---|
71 | |
---|
72 | STUDENT_SAMPLE_DATA_UPDATE3 = open( |
---|
73 | os.path.join(os.path.dirname(__file__), 'sample_student_data_update3.csv'), |
---|
74 | 'rb').read() |
---|
75 | |
---|
76 | STUDENT_HEADER_FIELDS_UPDATE3 = STUDENT_SAMPLE_DATA_UPDATE3.split( |
---|
77 | '\n')[0].split(',') |
---|
78 | |
---|
79 | STUDENT_SAMPLE_DATA_UPDATE4 = open( |
---|
80 | os.path.join(os.path.dirname(__file__), 'sample_student_data_update4.csv'), |
---|
81 | 'rb').read() |
---|
82 | |
---|
83 | STUDENT_HEADER_FIELDS_UPDATE4 = STUDENT_SAMPLE_DATA_UPDATE4.split( |
---|
84 | '\n')[0].split(',') |
---|
85 | |
---|
86 | STUDYCOURSE_SAMPLE_DATA = open( |
---|
87 | os.path.join(os.path.dirname(__file__), 'sample_studycourse_data.csv'), |
---|
88 | 'rb').read() |
---|
89 | |
---|
90 | STUDYCOURSE_HEADER_FIELDS = STUDYCOURSE_SAMPLE_DATA.split( |
---|
91 | '\n')[0].split(',') |
---|
92 | |
---|
93 | TRANSFER_SAMPLE_DATA = open( |
---|
94 | os.path.join(os.path.dirname(__file__), 'sample_transfer_data.csv'), |
---|
95 | 'rb').read() |
---|
96 | |
---|
97 | TRANSFER_HEADER_FIELDS = TRANSFER_SAMPLE_DATA.split( |
---|
98 | '\n')[0].split(',') |
---|
99 | |
---|
100 | VERDICT_SAMPLE_DATA = open( |
---|
101 | os.path.join(os.path.dirname(__file__), 'sample_verdict_data.csv'), |
---|
102 | 'rb').read() |
---|
103 | |
---|
104 | VERDICT_HEADER_FIELDS = VERDICT_SAMPLE_DATA.split( |
---|
105 | '\n')[0].split(',') |
---|
106 | |
---|
107 | STUDENT_SAMPLE_DATA_MIGRATION = open( |
---|
108 | os.path.join(os.path.dirname(__file__), |
---|
109 | 'sample_student_data_migration.csv'), |
---|
110 | 'rb').read() |
---|
111 | |
---|
112 | STUDENT_HEADER_FIELDS_MIGRATION = STUDENT_SAMPLE_DATA_MIGRATION.split( |
---|
113 | '\n')[0].split(',') |
---|
114 | |
---|
115 | STUDENT_SAMPLE_DATA_DUPLICATES = open( |
---|
116 | os.path.join(os.path.dirname(__file__), |
---|
117 | 'sample_student_data_duplicates.csv'), |
---|
118 | 'rb').read() |
---|
119 | |
---|
120 | STUDENT_HEADER_FIELDS_DUPLICATES = STUDENT_SAMPLE_DATA_DUPLICATES.split( |
---|
121 | '\n')[0].split(',') |
---|
122 | |
---|
123 | STUDENT_SAMPLE_DATA_EXTASCII = open( |
---|
124 | os.path.join(os.path.dirname(__file__), |
---|
125 | 'sample_student_data_extascii.csv'), |
---|
126 | 'rb').read() |
---|
127 | |
---|
128 | STUDENT_HEADER_FIELDS_EXTASCII = STUDENT_SAMPLE_DATA_EXTASCII.split( |
---|
129 | '\n')[0].split(',') |
---|
130 | |
---|
131 | STUDYLEVEL_SAMPLE_DATA = open( |
---|
132 | os.path.join(os.path.dirname(__file__), 'sample_studylevel_data.csv'), |
---|
133 | 'rb').read() |
---|
134 | |
---|
135 | STUDYLEVEL_HEADER_FIELDS = STUDYLEVEL_SAMPLE_DATA.split( |
---|
136 | '\n')[0].split(',') |
---|
137 | |
---|
138 | COURSETICKET_SAMPLE_DATA = open( |
---|
139 | os.path.join(os.path.dirname(__file__), 'sample_courseticket_data.csv'), |
---|
140 | 'rb').read() |
---|
141 | |
---|
142 | COURSETICKET_HEADER_FIELDS = COURSETICKET_SAMPLE_DATA.split( |
---|
143 | '\n')[0].split(',') |
---|
144 | |
---|
145 | PAYMENT_SAMPLE_DATA = open( |
---|
146 | os.path.join(os.path.dirname(__file__), 'sample_payment_data.csv'), |
---|
147 | 'rb').read() |
---|
148 | |
---|
149 | PAYMENT_HEADER_FIELDS = PAYMENT_SAMPLE_DATA.split( |
---|
150 | '\n')[0].split(',') |
---|
151 | |
---|
152 | PAYMENT_CREATE_SAMPLE_DATA = open( |
---|
153 | os.path.join(os.path.dirname(__file__), 'sample_create_payment_data.csv'), |
---|
154 | 'rb').read() |
---|
155 | |
---|
156 | PAYMENT_CREATE_HEADER_FIELDS = PAYMENT_CREATE_SAMPLE_DATA.split( |
---|
157 | '\n')[0].split(',') |
---|
158 | |
---|
159 | class StudentImportExportSetup(FunctionalTestCase): |
---|
160 | |
---|
161 | layer = FunctionalLayer |
---|
162 | |
---|
163 | def setUp(self): |
---|
164 | super(StudentImportExportSetup, self).setUp() |
---|
165 | self.dc_root = tempfile.mkdtemp() |
---|
166 | self.workdir = tempfile.mkdtemp() |
---|
167 | app = University() |
---|
168 | app['datacenter'].setStoragePath(self.dc_root) |
---|
169 | self.getRootFolder()['app'] = app |
---|
170 | self.app = self.getRootFolder()['app'] |
---|
171 | setSite(app) |
---|
172 | |
---|
173 | # Populate university |
---|
174 | self.certificate = createObject('waeup.Certificate') |
---|
175 | self.certificate.code = 'CERT1' |
---|
176 | self.certificate.application_category = 'basic' |
---|
177 | self.certificate.start_level = 200 |
---|
178 | self.certificate.end_level = 500 |
---|
179 | self.certificate.study_mode = u'ug_ft' |
---|
180 | self.app['faculties']['fac1'] = Faculty() |
---|
181 | self.app['faculties']['fac1']['dep1'] = Department() |
---|
182 | self.app['faculties']['fac1']['dep1'].certificates.addCertificate( |
---|
183 | self.certificate) |
---|
184 | |
---|
185 | # Create a hostel with two beds |
---|
186 | hostel = Hostel() |
---|
187 | hostel.hostel_id = u'hall-1' |
---|
188 | hostel.hostel_name = u'Hall 1' |
---|
189 | self.app['hostels'].addHostel(hostel) |
---|
190 | bed = Bed() |
---|
191 | bed.bed_id = u'hall-1_A_101_A' |
---|
192 | bed.bed_number = 1 |
---|
193 | bed.owner = NOT_OCCUPIED |
---|
194 | bed.bed_type = u'regular_male_fr' |
---|
195 | self.app['hostels'][hostel.hostel_id].addBed(bed) |
---|
196 | bed = Bed() |
---|
197 | bed.bed_id = u'hall-1_A_101_B' |
---|
198 | bed.bed_number = 2 |
---|
199 | bed.owner = NOT_OCCUPIED |
---|
200 | bed.bed_type = u'regular_female_fr' |
---|
201 | self.app['hostels'][hostel.hostel_id].addBed(bed) |
---|
202 | |
---|
203 | self.logfile = os.path.join( |
---|
204 | self.app['datacenter'].storage, 'logs', 'students.log') |
---|
205 | return |
---|
206 | |
---|
207 | def tearDown(self): |
---|
208 | super(StudentImportExportSetup, self).tearDown() |
---|
209 | shutil.rmtree(self.workdir) |
---|
210 | shutil.rmtree(self.dc_root) |
---|
211 | clearSite() |
---|
212 | return |
---|
213 | |
---|
214 | def setup_for_export(self): |
---|
215 | student = Student() |
---|
216 | student.student_id = u'A111111' |
---|
217 | self.app['students'][student.student_id] = self.student = student |
---|
218 | self.outfile = os.path.join(self.workdir, 'myoutput.csv') |
---|
219 | return |
---|
220 | |
---|
221 | def setup_student(self, student): |
---|
222 | # set predictable values for `student` |
---|
223 | student.matric_number = u'234' |
---|
224 | student.adm_code = u'my adm code' |
---|
225 | student.clearance_locked = False |
---|
226 | student.clr_code = u'my clr code' |
---|
227 | student.perm_address = u'Studentroad 21\nLagos 123456\n' |
---|
228 | student.reg_number = u'123' |
---|
229 | student.firstname = u'Anna' |
---|
230 | student.lastname = u'Tester' |
---|
231 | student.middlename = u'M.' |
---|
232 | student.date_of_birth = datetime.date(1981, 2, 4) |
---|
233 | student.sex = 'f' |
---|
234 | student.email = 'anna@sample.com' |
---|
235 | student.phone = u'+234-123-12345' |
---|
236 | student.notice = u'Some notice\nin lines.' |
---|
237 | student.nationality = u'NG' |
---|
238 | |
---|
239 | student['studycourse'].certificate = self.certificate |
---|
240 | student['studycourse'].entry_mode = 'ug_ft' |
---|
241 | student['studycourse'].entry_session = 2010 |
---|
242 | student['studycourse'].current_session = 2012 |
---|
243 | student['studycourse'].current_level = int(self.certificate.start_level) |
---|
244 | |
---|
245 | study_level = StudentStudyLevel() |
---|
246 | study_level.level_session = 2012 |
---|
247 | study_level.level_verdict = "A" |
---|
248 | study_level.level = 100 |
---|
249 | student['studycourse'].addStudentStudyLevel( |
---|
250 | self.certificate, study_level) |
---|
251 | |
---|
252 | ticket = CourseTicket() |
---|
253 | ticket.automatic = True |
---|
254 | ticket.carry_over = True |
---|
255 | ticket.code = u'CRS1' |
---|
256 | ticket.title = u'Course 1' |
---|
257 | ticket.fcode = u'FAC1' |
---|
258 | ticket.dcode = u'DEP1' |
---|
259 | ticket.credits = 100 |
---|
260 | ticket.passmark = 100 |
---|
261 | ticket.semester = 2 |
---|
262 | study_level[ticket.code] = ticket |
---|
263 | |
---|
264 | bedticket = BedTicket() |
---|
265 | bedticket.booking_session = 2004 |
---|
266 | bedticket.bed_type = u'any bed type' |
---|
267 | bedticket.bed = self.app['hostels']['hall-1']['hall-1_A_101_A'] |
---|
268 | student['accommodation'].addBedTicket(bedticket) |
---|
269 | |
---|
270 | self.add_payment(student) |
---|
271 | return student |
---|
272 | |
---|
273 | def add_payment(self, student): |
---|
274 | # get a payment with all fields set |
---|
275 | payment = StudentOnlinePayment() |
---|
276 | payment.creation_date = datetime.datetime(2012, 4, 1, 13, 12, 1) |
---|
277 | payment.p_id = 'my-id' |
---|
278 | payment.p_category = u'schoolfee' |
---|
279 | payment.p_state = 'paid' |
---|
280 | payment.ac = u'666' |
---|
281 | payment.p_item = u'p-item' |
---|
282 | payment.p_level = 100 |
---|
283 | payment.p_session = 2012 |
---|
284 | payment.payment_date = datetime.datetime(2012, 4, 1, 14, 12, 1) |
---|
285 | payment.amount_auth = 12.12 |
---|
286 | payment.r_amount_approved = 12.12 |
---|
287 | payment.r_code = u'r-code' |
---|
288 | # XXX: there is no addPayment method to give predictable names |
---|
289 | student['payments']['my-payment'] = payment |
---|
290 | return payment |
---|
291 | |
---|
292 | |
---|
293 | class StudentProcessorTest(StudentImportExportSetup): |
---|
294 | |
---|
295 | layer = FunctionalLayer |
---|
296 | |
---|
297 | def setUp(self): |
---|
298 | super(StudentProcessorTest, self).setUp() |
---|
299 | |
---|
300 | # Add student with subobjects |
---|
301 | student = Student() |
---|
302 | self.app['students'].addStudent(student) |
---|
303 | student = self.setup_student(student) |
---|
304 | notify(grok.ObjectModifiedEvent(student)) |
---|
305 | self.student = self.app['students'][student.student_id] |
---|
306 | |
---|
307 | self.processor = StudentProcessor() |
---|
308 | self.csv_file = os.path.join(self.workdir, 'sample_student_data.csv') |
---|
309 | self.csv_file_update = os.path.join( |
---|
310 | self.workdir, 'sample_student_data_update.csv') |
---|
311 | self.csv_file_update2 = os.path.join( |
---|
312 | self.workdir, 'sample_student_data_update2.csv') |
---|
313 | self.csv_file_update3 = os.path.join( |
---|
314 | self.workdir, 'sample_student_data_update3.csv') |
---|
315 | self.csv_file_update4 = os.path.join( |
---|
316 | self.workdir, 'sample_student_data_update4.csv') |
---|
317 | self.csv_file_migration = os.path.join( |
---|
318 | self.workdir, 'sample_student_data_migration.csv') |
---|
319 | self.csv_file_duplicates = os.path.join( |
---|
320 | self.workdir, 'sample_student_data_duplicates.csv') |
---|
321 | self.csv_file_extascii = os.path.join( |
---|
322 | self.workdir, 'sample_student_data_extascii.csv') |
---|
323 | open(self.csv_file, 'wb').write(STUDENT_SAMPLE_DATA) |
---|
324 | open(self.csv_file_update, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE) |
---|
325 | open(self.csv_file_update2, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE2) |
---|
326 | open(self.csv_file_update3, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE3) |
---|
327 | open(self.csv_file_update4, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE4) |
---|
328 | open(self.csv_file_migration, 'wb').write(STUDENT_SAMPLE_DATA_MIGRATION) |
---|
329 | open(self.csv_file_duplicates, 'wb').write(STUDENT_SAMPLE_DATA_DUPLICATES) |
---|
330 | open(self.csv_file_extascii, 'wb').write(STUDENT_SAMPLE_DATA_EXTASCII) |
---|
331 | |
---|
332 | def test_interface(self): |
---|
333 | # Make sure we fulfill the interface contracts. |
---|
334 | assert verifyObject(IBatchProcessor, self.processor) is True |
---|
335 | assert verifyClass( |
---|
336 | IBatchProcessor, StudentProcessor) is True |
---|
337 | |
---|
338 | def test_parentsExist(self): |
---|
339 | self.assertFalse(self.processor.parentsExist(None, dict())) |
---|
340 | self.assertTrue(self.processor.parentsExist(None, self.app)) |
---|
341 | |
---|
342 | def test_entryExists(self): |
---|
343 | assert self.processor.entryExists( |
---|
344 | dict(student_id='ID_NONE'), self.app) is False |
---|
345 | assert self.processor.entryExists( |
---|
346 | dict(reg_number='123'), self.app) is True |
---|
347 | |
---|
348 | def test_getParent(self): |
---|
349 | parent = self.processor.getParent(None, self.app) |
---|
350 | assert parent is self.app['students'] |
---|
351 | |
---|
352 | def test_getEntry(self): |
---|
353 | assert self.processor.getEntry( |
---|
354 | dict(student_id='ID_NONE'), self.app) is None |
---|
355 | assert self.processor.getEntry( |
---|
356 | dict(student_id=self.student.student_id), self.app) is self.student |
---|
357 | |
---|
358 | def test_addEntry(self): |
---|
359 | new_student = Student() |
---|
360 | self.processor.addEntry( |
---|
361 | new_student, dict(), self.app) |
---|
362 | assert len(self.app['students'].keys()) == 2 |
---|
363 | |
---|
364 | def test_checkConversion(self): |
---|
365 | # Make sure we can check conversions and that the stud_id |
---|
366 | # counter is not raised during such checks. |
---|
367 | initial_stud_id = self.app['students']._curr_stud_id |
---|
368 | errs, inv_errs, conv_dict = self.processor.checkConversion( |
---|
369 | dict(reg_number='1', state='admitted')) |
---|
370 | self.assertEqual(len(errs),0) |
---|
371 | # Empty state is allowed |
---|
372 | errs, inv_errs, conv_dict = self.processor.checkConversion( |
---|
373 | dict(reg_number='1', state='')) |
---|
374 | self.assertEqual(len(errs),0) |
---|
375 | #self.assertTrue(('state', 'no value provided') in errs) |
---|
376 | errs, inv_errs, conv_dict = self.processor.checkConversion( |
---|
377 | dict(reg_number='1', state='nonsense')) |
---|
378 | self.assertEqual(len(errs),1) |
---|
379 | self.assertTrue(('state', 'not allowed') in errs) |
---|
380 | new_stud_id = self.app['students']._curr_stud_id |
---|
381 | self.assertEqual(initial_stud_id, new_stud_id) |
---|
382 | return |
---|
383 | |
---|
384 | def test_checkUpdateRequirements(self): |
---|
385 | # Make sure that pg students can't be updated with wrong transition. |
---|
386 | err = self.processor.checkUpdateRequirements(self.student, |
---|
387 | dict(reg_number='1', state='returning'), self.app) |
---|
388 | self.assertTrue(err is None) |
---|
389 | self.certificate.study_mode = 'pg_ft' |
---|
390 | err = self.processor.checkUpdateRequirements(self.student, |
---|
391 | dict(reg_number='1', state='returning'), self.app) |
---|
392 | self.assertEqual(err, 'State not allowed (pg student).') |
---|
393 | IWorkflowState(self.student).setState('school fee paid') |
---|
394 | err = self.processor.checkUpdateRequirements(self.student, |
---|
395 | dict(reg_number='1', transition='reset6'), self.app) |
---|
396 | self.assertEqual(err, 'Transition not allowed (pg student).') |
---|
397 | err = self.processor.checkUpdateRequirements(self.student, |
---|
398 | dict(reg_number='1', transition='register_courses'), self.app) |
---|
399 | self.assertEqual(err, 'Transition not allowed (pg student).') |
---|
400 | |
---|
401 | |
---|
402 | def test_delEntry(self): |
---|
403 | assert self.student.student_id in self.app['students'].keys() |
---|
404 | self.processor.delEntry( |
---|
405 | dict(reg_number=self.student.reg_number), self.app) |
---|
406 | assert self.student.student_id not in self.app['students'].keys() |
---|
407 | |
---|
408 | def test_import(self): |
---|
409 | self.assertEqual(self.app['students']._curr_stud_id, 1000001) |
---|
410 | num, num_warns, fin_file, fail_file = self.processor.doImport( |
---|
411 | self.csv_file, STUDENT_HEADER_FIELDS) |
---|
412 | self.assertEqual(num_warns,0) |
---|
413 | self.assertEqual(len(self.app['students']), 10) |
---|
414 | self.assertEqual(self.app['students']['X666666'].reg_number,'1') |
---|
415 | self.assertEqual( |
---|
416 | self.app['students']['X666666'].state, 'courses validated') |
---|
417 | # Two new student_ids have been created. |
---|
418 | self.assertEqual(self.app['students']._curr_stud_id, 1000003) |
---|
419 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
420 | |
---|
421 | def test_import_extascii(self): |
---|
422 | self.assertEqual(self.app['students']._curr_stud_id, 1000001) |
---|
423 | num, num_warns, fin_file, fail_file = self.processor.doImport( |
---|
424 | self.csv_file_extascii, STUDENT_HEADER_FIELDS_EXTASCII) |
---|
425 | self.assertEqual(num_warns,0) |
---|
426 | self.assertEqual(len(self.app['students']), 3) |
---|
427 | self.assertEqual(self.app['students']['X111111'].reg_number,'1') |
---|
428 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
429 | |
---|
430 | def test_import_update(self): |
---|
431 | num, num_warns, fin_file, fail_file = self.processor.doImport( |
---|
432 | self.csv_file, STUDENT_HEADER_FIELDS) |
---|
433 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
434 | num, num_warns, fin_file, fail_file = self.processor.doImport( |
---|
435 | self.csv_file_update, STUDENT_HEADER_FIELDS_UPDATE, 'update') |
---|
436 | self.assertEqual(num_warns,0) |
---|
437 | # state has changed |
---|
438 | self.assertEqual(self.app['students']['X666666'].state,'admitted') |
---|
439 | # state has not changed |
---|
440 | self.assertEqual(self.app['students']['Y777777'].state, |
---|
441 | 'courses validated') |
---|
442 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
443 | |
---|
444 | def test_import_update2(self): |
---|
445 | num, num_warns, fin_file, fail_file = self.processor.doImport( |
---|
446 | self.csv_file, STUDENT_HEADER_FIELDS) |
---|
447 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
448 | num, num_warns, fin_file, fail_file = self.processor.doImport( |
---|
449 | self.csv_file_update2, STUDENT_HEADER_FIELDS_UPDATE2, 'update') |
---|
450 | self.assertEqual(num_warns,0) |
---|
451 | # The phone import value of Pieri was None. |
---|
452 | # Confirm that phone has not been cleared. |
---|
453 | container = self.app['students'] |
---|
454 | for key in container.keys(): |
---|
455 | if container[key].firstname == 'Aaren': |
---|
456 | aaren = container[key] |
---|
457 | break |
---|
458 | self.assertEqual(aaren.phone, '--1234') |
---|
459 | # The phone import value of Claus was a deletion marker. |
---|
460 | # Confirm that phone has been cleared. |
---|
461 | for key in container.keys(): |
---|
462 | if container[key].firstname == 'Claus': |
---|
463 | claus = container[key] |
---|
464 | break |
---|
465 | assert claus.phone is None |
---|
466 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
467 | |
---|
468 | def test_import_update3(self): |
---|
469 | num, num_warns, fin_file, fail_file = self.processor.doImport( |
---|
470 | self.csv_file, STUDENT_HEADER_FIELDS) |
---|
471 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
472 | num, num_warns, fin_file, fail_file = self.processor.doImport( |
---|
473 | self.csv_file_update3, STUDENT_HEADER_FIELDS_UPDATE3, 'update') |
---|
474 | content = open(fail_file).read() |
---|
475 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
476 | self.assertEqual( |
---|
477 | content, |
---|
478 | 'reg_number,student_id,transition,--ERRORS--\r\n' |
---|
479 | '<IGNORE>,X666666,request_clearance,Transition not allowed.\r\n' |
---|
480 | ) |
---|
481 | self.assertEqual(num_warns,1) |
---|
482 | self.assertEqual(self.app['students']['Y777777'].state,'returning') |
---|
483 | |
---|
484 | def test_import_update4(self): |
---|
485 | num, num_warns, fin_file, fail_file = self.processor.doImport( |
---|
486 | self.csv_file, STUDENT_HEADER_FIELDS) |
---|
487 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
488 | self.assertRaises( |
---|
489 | FatalCSVError, self.processor.doImport, self.csv_file_update4, |
---|
490 | STUDENT_HEADER_FIELDS_UPDATE4, 'update') |
---|
491 | |
---|
492 | def test_import_remove(self): |
---|
493 | num, num_warns, fin_file, fail_file = self.processor.doImport( |
---|
494 | self.csv_file, STUDENT_HEADER_FIELDS) |
---|
495 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
496 | num, num_warns, fin_file, fail_file = self.processor.doImport( |
---|
497 | self.csv_file_update, STUDENT_HEADER_FIELDS_UPDATE, 'remove') |
---|
498 | self.assertEqual(num_warns,0) |
---|
499 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
500 | |
---|
501 | def test_import_migration_data(self): |
---|
502 | num, num_warns, fin_file, fail_file = self.processor.doImport( |
---|
503 | self.csv_file_migration, STUDENT_HEADER_FIELDS_MIGRATION) |
---|
504 | content = open(fail_file).read() |
---|
505 | self.assertEqual(num_warns,2) |
---|
506 | assert len(self.app['students'].keys()) == 5 |
---|
507 | self.assertEqual( |
---|
508 | content, |
---|
509 | 'reg_number,firstname,student_id,sex,email,phone,state,date_of_birth,lastname,password,matric_number,--ERRORS--\r\n' |
---|
510 | '4,John,D123456,m,aa@aa.ng,1234,nonsense,1990-01-05,Wolter,mypw1,100003,state: not allowed\r\n' |
---|
511 | '5,John,E123456,x,aa@aa.ng,1234,,1990-01-06,Kennedy,,100004,sex: Invalid value\r\n' |
---|
512 | ) |
---|
513 | students = self.app['students'] |
---|
514 | self.assertTrue('A123456' in students.keys()) |
---|
515 | self.assertEqual(students['A123456'].state, 'clearance started') |
---|
516 | self.assertEqual(students['A123456'].date_of_birth, |
---|
517 | datetime.date(1990, 1, 2)) |
---|
518 | self.assertFalse(students['A123456'].clearance_locked) |
---|
519 | self.assertEqual(students['B123456'].state, 'cleared') |
---|
520 | self.assertEqual(students['B123456'].date_of_birth, |
---|
521 | datetime.date(1990, 1, 3)) |
---|
522 | self.assertTrue(students['B123456'].clearance_locked) |
---|
523 | history = ' '.join(students['A123456'].history.messages) |
---|
524 | self.assertTrue( |
---|
525 | "State 'clearance started' set by system" in history) |
---|
526 | # state was empty and student is thus in state created |
---|
527 | self.assertEqual(students['F123456'].state,'created') |
---|
528 | # passwords were set correctly |
---|
529 | self.assertEqual( |
---|
530 | IUserAccount(students['A123456']).checkPassword('mypw1'), True) |
---|
531 | self.assertEqual( |
---|
532 | IUserAccount(students['C123456']).checkPassword('mypw1'), True) |
---|
533 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
534 | |
---|
535 | def test_import_duplicate_data(self): |
---|
536 | num, num_warns, fin_file, fail_file = self.processor.doImport( |
---|
537 | self.csv_file_duplicates, STUDENT_HEADER_FIELDS_DUPLICATES) |
---|
538 | content = open(fail_file).read() |
---|
539 | self.assertEqual(num_warns,4) |
---|
540 | self.assertEqual( |
---|
541 | content, |
---|
542 | 'reg_number,firstname,student_id,sex,email,phone,state,date_of_birth,lastname,password,matric_number,--ERRORS--\r\n' |
---|
543 | '1,Aaren,B123456,m,aa@aa.ng,1234,cleared,1990-01-03,Finau,mypw1,100001,reg_number: reg_number\r\n' |
---|
544 | '2,Aaren,C123456,m,aa@aa.ng,1234,admitted,1990-01-04,Berson,mypw1,100000,matric_number: matric_number\r\n' |
---|
545 | '1,Frank,F123456,m,aa@aa.ng,1234,,1990-01-06,Meyer,,100000,reg_number: reg_number; matric_number: matric_number\r\n' |
---|
546 | '3,Uli,A123456,m,aa@aa.ng,1234,,1990-01-07,Schulz,,100002,This object already exists. Skipping.\r\n' |
---|
547 | ) |
---|
548 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
549 | |
---|
550 | class StudentStudyCourseProcessorTest(StudentImportExportSetup): |
---|
551 | |
---|
552 | def setUp(self): |
---|
553 | super(StudentStudyCourseProcessorTest, self).setUp() |
---|
554 | |
---|
555 | # Add student with subobjects |
---|
556 | student = Student() |
---|
557 | self.app['students'].addStudent(student) |
---|
558 | student = self.setup_student(student) |
---|
559 | notify(grok.ObjectModifiedEvent(student)) |
---|
560 | self.student = self.app['students'][student.student_id] |
---|
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.processor = StudentStudyCourseProcessor() |
---|
570 | self.csv_file = os.path.join( |
---|
571 | self.workdir, 'sample_studycourse_data.csv') |
---|
572 | open(self.csv_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA) |
---|
573 | self.csv_file_transfer = os.path.join( |
---|
574 | self.workdir, 'sample_transfer_data.csv') |
---|
575 | open(self.csv_file_transfer, 'wb').write(TRANSFER_SAMPLE_DATA) |
---|
576 | return |
---|
577 | |
---|
578 | def test_interface(self): |
---|
579 | # Make sure we fulfill the interface contracts. |
---|
580 | assert verifyObject(IBatchProcessor, self.processor) is True |
---|
581 | assert verifyClass( |
---|
582 | IBatchProcessor, StudentStudyCourseProcessor) is True |
---|
583 | |
---|
584 | def test_entryExists(self): |
---|
585 | assert self.processor.entryExists( |
---|
586 | dict(reg_number='REG_NONE'), self.app) is False |
---|
587 | assert self.processor.entryExists( |
---|
588 | dict(reg_number='1'), self.app) is True |
---|
589 | |
---|
590 | def test_getEntry(self): |
---|
591 | student = self.processor.getEntry( |
---|
592 | dict(reg_number='1'), self.app).__parent__ |
---|
593 | self.assertEqual(student.reg_number,'1') |
---|
594 | |
---|
595 | def test_checkConversion(self): |
---|
596 | errs, inv_errs, conv_dict = self.processor.checkConversion( |
---|
597 | dict(reg_number='1', certificate='CERT1', current_level='200')) |
---|
598 | self.assertEqual(len(errs),0) |
---|
599 | errs, inv_errs, conv_dict = self.processor.checkConversion( |
---|
600 | dict(reg_number='1', certificate='CERT999')) |
---|
601 | self.assertEqual(len(errs),1) |
---|
602 | self.assertTrue(('certificate', u'Invalid value') in errs) |
---|
603 | errs, inv_errs, conv_dict = self.processor.checkConversion( |
---|
604 | dict(reg_number='1', certificate='CERT1', current_level='100')) |
---|
605 | self.assertEqual(len(errs),1) |
---|
606 | self.assertTrue(('current_level','not in range') in errs) |
---|
607 | # If we import only current_level, no conversion checking is done. |
---|
608 | errs, inv_errs, conv_dict = self.processor.checkConversion( |
---|
609 | dict(reg_number='1', current_level='100')) |
---|
610 | self.assertEqual(len(errs),0) |
---|
611 | |
---|
612 | def test_checkUpdateRequirements(self): |
---|
613 | # Current level must be in range of certificate. |
---|
614 | # Since row has passed the converter, current_level is an integer. |
---|
615 | err = self.processor.checkUpdateRequirements( |
---|
616 | self.student['studycourse'], |
---|
617 | dict(reg_number='1', current_level=100), self.app) |
---|
618 | self.assertEqual(err, 'current_level not in range.') |
---|
619 | err = self.processor.checkUpdateRequirements( |
---|
620 | self.student['studycourse'], |
---|
621 | dict(reg_number='1', current_level=200), self.app) |
---|
622 | self.assertTrue(err is None) |
---|
623 | # We can update pg students. |
---|
624 | self.student['studycourse'].certificate.start_level=999 |
---|
625 | self.student['studycourse'].certificate.end_level=999 |
---|
626 | err = self.processor.checkUpdateRequirements( |
---|
627 | self.student['studycourse'], |
---|
628 | dict(reg_number='1', current_level=999), self.app) |
---|
629 | self.assertTrue(err is None) |
---|
630 | # Make sure that pg students can't be updated with wrong transition. |
---|
631 | IWorkflowState(self.student).setState('returning') |
---|
632 | err = self.processor.checkUpdateRequirements( |
---|
633 | self.student['studycourse'], |
---|
634 | dict(reg_number='1', current_level=999), self.app) |
---|
635 | self.assertEqual(err, 'Not a pg student.') |
---|
636 | # If certificate is not given in row (and has thus |
---|
637 | # successfully passed checkConversion) the certificate |
---|
638 | # attribute must be set. |
---|
639 | self.student['studycourse'].certificate = None |
---|
640 | err = self.processor.checkUpdateRequirements( |
---|
641 | self.student['studycourse'], |
---|
642 | dict(reg_number='1', current_level=100), self.app) |
---|
643 | self.assertEqual(err, 'No certificate to check level.') |
---|
644 | # When transferring students the method also checks |
---|
645 | # if the former studycourse is complete. |
---|
646 | err = self.processor.checkUpdateRequirements( |
---|
647 | self.student['studycourse'], |
---|
648 | dict(reg_number='1', certificate='CERT1', current_level=200, |
---|
649 | entry_mode='transfer'), self.app) |
---|
650 | self.assertEqual(err, 'Former study course record incomplete.') |
---|
651 | self.student['studycourse'].certificate = self.certificate |
---|
652 | self.student['studycourse'].entry_session = 2005 |
---|
653 | # The method doesn't care if current_level |
---|
654 | # is not in range of CERT1. This is done by checkConversion |
---|
655 | # if certificate is in row. |
---|
656 | err = self.processor.checkUpdateRequirements( |
---|
657 | self.student['studycourse'], |
---|
658 | dict(reg_number='1', certificate='CERT1', current_level=200, |
---|
659 | entry_mode='transfer'), self.app) |
---|
660 | self.assertTrue(err is None) |
---|
661 | |
---|
662 | def test_import(self): |
---|
663 | num, num_warns, fin_file, fail_file = self.processor.doImport( |
---|
664 | self.csv_file, STUDYCOURSE_HEADER_FIELDS,'update') |
---|
665 | self.assertEqual(num_warns,1) |
---|
666 | content = open(fail_file).read() |
---|
667 | self.assertTrue('current_level: not in range' in content) |
---|
668 | studycourse = self.processor.getEntry(dict(reg_number='1'), self.app) |
---|
669 | self.assertEqual(studycourse.certificate.code, u'CERT1') |
---|
670 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
671 | |
---|
672 | def test_import_transfer(self): |
---|
673 | self.certificate2 = createObject('waeup.Certificate') |
---|
674 | self.certificate2.code = 'CERT2' |
---|
675 | self.certificate2.application_category = 'basic' |
---|
676 | self.certificate2.start_level = 200 |
---|
677 | self.certificate2.end_level = 500 |
---|
678 | self.certificate2.study_mode = u'ug_pt' |
---|
679 | self.app['faculties']['fac1']['dep1'].certificates.addCertificate( |
---|
680 | self.certificate2) |
---|
681 | num, num_warns, fin_file, fail_file = self.processor.doImport( |
---|
682 | self.csv_file_transfer, TRANSFER_HEADER_FIELDS,'update') |
---|
683 | self.assertEqual(num_warns,0) |
---|
684 | self.assertEqual(self.student['studycourse'].certificate.code, 'CERT2') |
---|
685 | self.assertEqual(self.student['studycourse_1'].certificate.code, 'CERT1') |
---|
686 | self.assertEqual(self.student['studycourse'].entry_mode, 'transfer') |
---|
687 | self.assertEqual(self.student['studycourse_1'].entry_mode, 'ug_ft') |
---|
688 | self.assertEqual(self.student.current_mode, 'ug_pt') |
---|
689 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
690 | # Transer has bee logged. |
---|
691 | logcontent = open(self.logfile).read() |
---|
692 | self.assertTrue( |
---|
693 | 'INFO - system - K1000000 - transferred from CERT1 to CERT2\n' |
---|
694 | in logcontent) |
---|
695 | self.assertTrue( |
---|
696 | 'INFO - system - ' |
---|
697 | 'StudentStudyCourse Processor (update only) - ' |
---|
698 | 'sample_transfer_data - K1000000 - updated: entry_mode=transfer, ' |
---|
699 | 'certificate=CERT2, current_session=2009, current_level=300' |
---|
700 | in logcontent) |
---|
701 | # A history message has been added. |
---|
702 | history = ' '.join(self.student.history.messages) |
---|
703 | self.assertTrue( |
---|
704 | "Transferred from CERT1 to CERT2 by system" in history) |
---|
705 | # The catalog has been updated |
---|
706 | cat = queryUtility(ICatalog, name='students_catalog') |
---|
707 | results = list( |
---|
708 | cat.searchResults( |
---|
709 | certcode=('CERT2', 'CERT2'))) |
---|
710 | self.assertTrue(results[0] is self.student) |
---|
711 | results = list( |
---|
712 | cat.searchResults( |
---|
713 | current_session=(2009, 2009))) |
---|
714 | self.assertTrue(results[0] is self.student) |
---|
715 | results = list( |
---|
716 | cat.searchResults( |
---|
717 | certcode=('CERT1', 'CERT1'))) |
---|
718 | self.assertEqual(len(results), 0) |
---|
719 | |
---|
720 | class StudentStudyLevelProcessorTest(StudentImportExportSetup): |
---|
721 | |
---|
722 | def setUp(self): |
---|
723 | super(StudentStudyLevelProcessorTest, self).setUp() |
---|
724 | |
---|
725 | # Import students with subobjects |
---|
726 | student_file = os.path.join(self.workdir, 'sample_student_data.csv') |
---|
727 | open(student_file, 'wb').write(STUDENT_SAMPLE_DATA) |
---|
728 | num, num_warns, fin_file, fail_file = StudentProcessor().doImport( |
---|
729 | student_file, STUDENT_HEADER_FIELDS) |
---|
730 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
731 | |
---|
732 | # Update study courses |
---|
733 | studycourse_file = os.path.join( |
---|
734 | self.workdir, 'sample_studycourse_data.csv') |
---|
735 | open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA) |
---|
736 | processor = StudentStudyCourseProcessor() |
---|
737 | num, num_warns, fin_file, fail_file = processor.doImport( |
---|
738 | studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update') |
---|
739 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
740 | |
---|
741 | self.processor = StudentStudyLevelProcessor() |
---|
742 | self.csv_file = os.path.join( |
---|
743 | self.workdir, 'sample_studylevel_data.csv') |
---|
744 | open(self.csv_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA) |
---|
745 | |
---|
746 | def test_interface(self): |
---|
747 | # Make sure we fulfill the interface contracts. |
---|
748 | assert verifyObject(IBatchProcessor, self.processor) is True |
---|
749 | assert verifyClass( |
---|
750 | IBatchProcessor, StudentStudyLevelProcessor) is True |
---|
751 | |
---|
752 | def test_checkConversion(self): |
---|
753 | errs, inv_errs, conv_dict = self.processor.checkConversion( |
---|
754 | dict(reg_number='1', level='220')) |
---|
755 | self.assertEqual(len(errs),0) |
---|
756 | errs, inv_errs, conv_dict = self.processor.checkConversion( |
---|
757 | dict(reg_number='1', level='999')) |
---|
758 | self.assertEqual(len(errs),0) |
---|
759 | errs, inv_errs, conv_dict = self.processor.checkConversion( |
---|
760 | dict(reg_number='1', level='1000')) |
---|
761 | self.assertEqual(len(errs),1) |
---|
762 | self.assertTrue(('level','no valid integer') in errs) |
---|
763 | errs, inv_errs, conv_dict = self.processor.checkConversion( |
---|
764 | dict(reg_number='1', level='xyz')) |
---|
765 | self.assertEqual(len(errs),1) |
---|
766 | self.assertTrue(('level','no integer') in errs) |
---|
767 | |
---|
768 | def test_import(self): |
---|
769 | num, num_warns, fin_file, fail_file = self.processor.doImport( |
---|
770 | self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create') |
---|
771 | self.assertEqual(num_warns,2) |
---|
772 | assert self.processor.entryExists( |
---|
773 | dict(reg_number='1', level='100'), self.app) is True |
---|
774 | studylevel = self.processor.getEntry( |
---|
775 | dict(reg_number='1', level='100'), self.app) |
---|
776 | self.assertEqual(studylevel.__parent__.certificate.code, u'CERT1') |
---|
777 | self.assertEqual(studylevel.level_session, 2008) |
---|
778 | self.assertEqual(studylevel.level_verdict, None) |
---|
779 | self.assertEqual(studylevel.level, 100) |
---|
780 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
781 | |
---|
782 | logcontent = open(self.logfile).read() |
---|
783 | # Logging message from updateEntry, |
---|
784 | self.assertTrue( |
---|
785 | 'INFO - system - StudentStudyLevel Processor - ' |
---|
786 | 'sample_studylevel_data - K1000000 - updated: ' |
---|
787 | 'level=100, level_verdict=C, level_session=2009' |
---|
788 | in logcontent) |
---|
789 | |
---|
790 | def test_import_update(self): |
---|
791 | # We perform the same import twice, |
---|
792 | # the second time in update mode. The number |
---|
793 | # of warnings must be the same. |
---|
794 | num, num_warns, fin_file, fail_file = self.processor.doImport( |
---|
795 | self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create') |
---|
796 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
797 | num, num_warns, fin_file, fail_file = self.processor.doImport( |
---|
798 | self.csv_file, STUDYLEVEL_HEADER_FIELDS,'update') |
---|
799 | self.assertEqual(num_warns,2) |
---|
800 | studylevel = self.processor.getEntry( |
---|
801 | dict(reg_number='1', level='100'), self.app) |
---|
802 | self.assertEqual(studylevel.level, 100) |
---|
803 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
804 | |
---|
805 | def test_import_remove(self): |
---|
806 | # We perform the same import twice, |
---|
807 | # the second time in remove mode. The number |
---|
808 | # of warnings must be the same. |
---|
809 | num, num_warns, fin_file, fail_file = self.processor.doImport( |
---|
810 | self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create') |
---|
811 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
812 | num, num_warns, fin_file, fail_file = self.processor.doImport( |
---|
813 | self.csv_file, STUDYLEVEL_HEADER_FIELDS,'remove') |
---|
814 | assert self.processor.entryExists( |
---|
815 | dict(reg_number='1', level='100'), self.app) is False |
---|
816 | self.assertEqual(num_warns,2) |
---|
817 | |
---|
818 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
819 | |
---|
820 | class CourseTicketProcessorTest(StudentImportExportSetup): |
---|
821 | |
---|
822 | def setUp(self): |
---|
823 | super(CourseTicketProcessorTest, self).setUp() |
---|
824 | |
---|
825 | # Import students with subobjects |
---|
826 | student_file = os.path.join(self.workdir, 'sample_student_data.csv') |
---|
827 | open(student_file, 'wb').write(STUDENT_SAMPLE_DATA) |
---|
828 | num, num_warns, fin_file, fail_file = StudentProcessor().doImport( |
---|
829 | student_file, STUDENT_HEADER_FIELDS) |
---|
830 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
831 | |
---|
832 | # Add course and certificate course |
---|
833 | self.course = createObject('waeup.Course') |
---|
834 | self.course.code = 'COURSE1' |
---|
835 | self.course.semester = 1 |
---|
836 | self.course.credits = 10 |
---|
837 | self.course.passmark = 40 |
---|
838 | self.app['faculties']['fac1']['dep1'].courses.addCourse( |
---|
839 | self.course) |
---|
840 | self.app['faculties']['fac1']['dep1'].certificates[ |
---|
841 | 'CERT1'].addCertCourse( |
---|
842 | self.course, level=100) |
---|
843 | |
---|
844 | # Update study courses |
---|
845 | studycourse_file = os.path.join( |
---|
846 | self.workdir, 'sample_studycourse_data.csv') |
---|
847 | open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA) |
---|
848 | processor = StudentStudyCourseProcessor() |
---|
849 | num, num_warns, fin_file, fail_file = processor.doImport( |
---|
850 | studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update') |
---|
851 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
852 | |
---|
853 | # Import study levels |
---|
854 | processor = StudentStudyLevelProcessor() |
---|
855 | studylevel_file = os.path.join( |
---|
856 | self.workdir, 'sample_studylevel_data.csv') |
---|
857 | open(studylevel_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA) |
---|
858 | num, num_warns, fin_file, fail_file = processor.doImport( |
---|
859 | studylevel_file, STUDYLEVEL_HEADER_FIELDS,'create') |
---|
860 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
861 | |
---|
862 | self.processor = CourseTicketProcessor() |
---|
863 | self.csv_file = os.path.join( |
---|
864 | self.workdir, 'sample_courseticket_data.csv') |
---|
865 | open(self.csv_file, 'wb').write(COURSETICKET_SAMPLE_DATA) |
---|
866 | |
---|
867 | def test_interface(self): |
---|
868 | # Make sure we fulfill the interface contracts. |
---|
869 | assert verifyObject(IBatchProcessor, self.processor) is True |
---|
870 | assert verifyClass( |
---|
871 | IBatchProcessor, CourseTicketProcessor) is True |
---|
872 | |
---|
873 | def test_checkConversion(self): |
---|
874 | errs, inv_errs, conv_dict = self.processor.checkConversion( |
---|
875 | dict(reg_number='1', code='COURSE1', level='220')) |
---|
876 | self.assertEqual(len(errs),0) |
---|
877 | errs, inv_errs, conv_dict = self.processor.checkConversion( |
---|
878 | dict(reg_number='1', code='COURSE2', level='220')) |
---|
879 | self.assertEqual(len(errs),1) |
---|
880 | self.assertTrue(('code','non-existent') in errs) |
---|
881 | |
---|
882 | def test_import(self): |
---|
883 | num, num_warns, fin_file, fail_file = self.processor.doImport( |
---|
884 | self.csv_file, COURSETICKET_HEADER_FIELDS,'create') |
---|
885 | fail_file = open(fail_file).read() |
---|
886 | self.assertEqual(num_warns,5) |
---|
887 | self.assertEqual(fail_file, |
---|
888 | 'reg_number,code,mandatory,level,level_session,score,matric_number,--ERRORS--\r\n' |
---|
889 | '1,COURSE1,,nonsense,,5,,Not all parents do exist yet. Skipping\r\n' |
---|
890 | '1,NONSENSE,,100,,5,,code: non-existent\r\n' |
---|
891 | '1,COURSE1,,200,2004,6,,level_session: does not match 2008\r\n' |
---|
892 | '1,COURSE1,,300,2008,6,,level: does not exist\r\n' |
---|
893 | '1,COURSE1,,300,2008X,6,,level_session: Invalid value\r\n') |
---|
894 | assert self.processor.entryExists( |
---|
895 | dict(reg_number='1', level='100', code='COURSE1'), |
---|
896 | self.app) is True |
---|
897 | courseticket = self.processor.getEntry( |
---|
898 | dict(reg_number='1', level='100', code='COURSE1'), self.app) |
---|
899 | self.assertEqual(courseticket.__parent__.__parent__.certificate.code, |
---|
900 | u'CERT1') |
---|
901 | self.assertEqual(courseticket.score, 1) |
---|
902 | self.assertEqual(courseticket.mandatory, True) |
---|
903 | self.assertEqual(courseticket.fcode, 'NA') |
---|
904 | self.assertEqual(courseticket.dcode, 'NA') |
---|
905 | self.assertEqual(courseticket.code, 'COURSE1') |
---|
906 | self.assertEqual(courseticket.title, 'Unnamed Course') |
---|
907 | self.assertEqual(courseticket.credits, 10) |
---|
908 | self.assertEqual(courseticket.passmark, 40) |
---|
909 | self.assertEqual(courseticket.semester, 1) |
---|
910 | self.assertEqual(courseticket.level, 100) |
---|
911 | self.assertEqual(courseticket.level_session, 2008) |
---|
912 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
913 | logcontent = open(self.logfile).read() |
---|
914 | # Logging message from updateEntry, |
---|
915 | self.assertTrue( |
---|
916 | 'INFO - system - CourseTicket Processor - ' |
---|
917 | 'sample_courseticket_data - K1000000 - 100 - ' |
---|
918 | 'updated: code=COURSE1, ' |
---|
919 | 'mandatory=False, score=3' |
---|
920 | in logcontent) |
---|
921 | |
---|
922 | # The catalog has been updated |
---|
923 | cat = queryUtility(ICatalog, name='coursetickets_catalog') |
---|
924 | results = list( |
---|
925 | cat.searchResults( |
---|
926 | level=(100, 100))) |
---|
927 | self.assertEqual(len(results),3) |
---|
928 | |
---|
929 | def test_import_update(self): |
---|
930 | # We perform the same import twice, |
---|
931 | # the second time in update mode. The number |
---|
932 | # of warnings must be the same. |
---|
933 | num, num_warns, fin_file, fail_file = self.processor.doImport( |
---|
934 | self.csv_file, COURSETICKET_HEADER_FIELDS,'create') |
---|
935 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
936 | num, num_warns, fin_file, fail_file = self.processor.doImport( |
---|
937 | self.csv_file, COURSETICKET_HEADER_FIELDS,'update') |
---|
938 | fail_file = open(fail_file).read() |
---|
939 | self.assertEqual(num_warns,5) |
---|
940 | self.assertEqual(fail_file, |
---|
941 | 'reg_number,code,mandatory,level,level_session,score,matric_number,--ERRORS--\r\n' |
---|
942 | '1,COURSE1,<IGNORE>,nonsense,<IGNORE>,5,<IGNORE>,Cannot update: no such entry\r\n' |
---|
943 | '1,NONSENSE,<IGNORE>,100,<IGNORE>,5,<IGNORE>,code: non-existent\r\n' |
---|
944 | '1,COURSE1,<IGNORE>,200,2004,6,<IGNORE>,level_session: does not match 2008\r\n' |
---|
945 | '1,COURSE1,<IGNORE>,300,2008,6,<IGNORE>,level: does not exist\r\n' |
---|
946 | '1,COURSE1,<IGNORE>,300,2008X,6,<IGNORE>,level_session: Invalid value\r\n') |
---|
947 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
948 | |
---|
949 | def test_import_remove(self): |
---|
950 | # We perform the same import twice, |
---|
951 | # the second time in remove mode. The number |
---|
952 | # of warnings must be the same. |
---|
953 | num, num_warns, fin_file, fail_file = self.processor.doImport( |
---|
954 | self.csv_file, COURSETICKET_HEADER_FIELDS,'create') |
---|
955 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
956 | assert self.processor.entryExists( |
---|
957 | dict(reg_number='1', level='100', code='COURSE1'), self.app) is True |
---|
958 | num, num_warns, fin_file, fail_file = self.processor.doImport( |
---|
959 | self.csv_file, COURSETICKET_HEADER_FIELDS,'remove') |
---|
960 | self.assertEqual(num_warns,5) |
---|
961 | assert self.processor.entryExists( |
---|
962 | dict(reg_number='1', level='100', code='COURSE1'), self.app) is False |
---|
963 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
964 | logcontent = open(self.logfile).read() |
---|
965 | self.assertTrue( |
---|
966 | 'INFO - system - K1000000 - Course ticket in 100 removed: COURSE1' |
---|
967 | in logcontent) |
---|
968 | |
---|
969 | class PaymentProcessorTest(StudentImportExportSetup): |
---|
970 | |
---|
971 | def setUp(self): |
---|
972 | super(PaymentProcessorTest, self).setUp() |
---|
973 | |
---|
974 | # Add student with payment |
---|
975 | student = Student() |
---|
976 | student.firstname = u'Anna' |
---|
977 | student.lastname = u'Tester' |
---|
978 | student.reg_number = u'123' |
---|
979 | student.matric_number = u'234' |
---|
980 | self.app['students'].addStudent(student) |
---|
981 | self.student = self.app['students'][student.student_id] |
---|
982 | payment = createObject(u'waeup.StudentOnlinePayment') |
---|
983 | payment.p_id = 'p120' |
---|
984 | self.student['payments'][payment.p_id] = payment |
---|
985 | |
---|
986 | # Import students with subobjects |
---|
987 | student_file = os.path.join(self.workdir, 'sample_student_data.csv') |
---|
988 | open(student_file, 'wb').write(STUDENT_SAMPLE_DATA) |
---|
989 | num, num_warns, fin_file, fail_file = StudentProcessor().doImport( |
---|
990 | student_file, STUDENT_HEADER_FIELDS) |
---|
991 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
992 | |
---|
993 | self.processor = StudentOnlinePaymentProcessor() |
---|
994 | self.csv_file = os.path.join( |
---|
995 | self.workdir, 'sample_payment_data.csv') |
---|
996 | open(self.csv_file, 'wb').write(PAYMENT_SAMPLE_DATA) |
---|
997 | self.csv_file2 = os.path.join( |
---|
998 | self.workdir, 'sample_create_payment_data.csv') |
---|
999 | open(self.csv_file2, 'wb').write(PAYMENT_CREATE_SAMPLE_DATA) |
---|
1000 | |
---|
1001 | def test_interface(self): |
---|
1002 | # Make sure we fulfill the interface contracts. |
---|
1003 | assert verifyObject(IBatchProcessor, self.processor) is True |
---|
1004 | assert verifyClass( |
---|
1005 | IBatchProcessor, StudentOnlinePaymentProcessor) is True |
---|
1006 | |
---|
1007 | def test_getEntry(self): |
---|
1008 | assert self.processor.getEntry( |
---|
1009 | dict(student_id='ID_NONE', p_id='nonsense'), self.app) is None |
---|
1010 | assert self.processor.getEntry( |
---|
1011 | dict(student_id=self.student.student_id, p_id='p120'), |
---|
1012 | self.app) is self.student['payments']['p120'] |
---|
1013 | assert self.processor.getEntry( |
---|
1014 | dict(student_id=self.student.student_id, p_id='XXXXXX112'), |
---|
1015 | self.app) is self.student['payments']['p120'] |
---|
1016 | |
---|
1017 | def test_delEntry(self): |
---|
1018 | assert self.processor.getEntry( |
---|
1019 | dict(student_id=self.student.student_id, p_id='p120'), |
---|
1020 | self.app) is self.student['payments']['p120'] |
---|
1021 | self.assertEqual(len(self.student['payments'].keys()),1) |
---|
1022 | self.processor.delEntry( |
---|
1023 | dict(student_id=self.student.student_id, p_id='p120'), |
---|
1024 | self.app) |
---|
1025 | assert self.processor.getEntry( |
---|
1026 | dict(student_id=self.student.student_id, p_id='p120'), |
---|
1027 | self.app) is None |
---|
1028 | self.assertEqual(len(self.student['payments'].keys()),0) |
---|
1029 | |
---|
1030 | def test_addEntry(self): |
---|
1031 | self.assertEqual(len(self.student['payments'].keys()),1) |
---|
1032 | payment1 = createObject(u'waeup.StudentOnlinePayment') |
---|
1033 | payment1.p_id = 'p234' |
---|
1034 | self.processor.addEntry( |
---|
1035 | payment1, dict(student_id=self.student.student_id, p_id='p234'), |
---|
1036 | self.app) |
---|
1037 | self.assertEqual(len(self.student['payments'].keys()),2) |
---|
1038 | self.assertEqual(self.student['payments']['p234'].p_id, 'p234') |
---|
1039 | payment2 = createObject(u'waeup.StudentOnlinePayment') |
---|
1040 | payment1.p_id = 'nonsense' |
---|
1041 | # payment1.p_id will be replaced if p_id doesn't start with 'p' |
---|
1042 | # and is not an old PIN |
---|
1043 | self.processor.addEntry( |
---|
1044 | payment2, dict(student_id=self.student.student_id, p_id='XXXXXX456'), |
---|
1045 | self.app) |
---|
1046 | self.assertEqual(len(self.student['payments'].keys()),3) |
---|
1047 | self.assertEqual(self.student['payments']['p560'].p_id, 'p560') |
---|
1048 | |
---|
1049 | def test_checkConversion(self): |
---|
1050 | errs, inv_errs, conv_dict = self.processor.checkConversion( |
---|
1051 | dict(p_id='3816951266236341955')) |
---|
1052 | self.assertEqual(len(errs),0) |
---|
1053 | errs, inv_errs, conv_dict = self.processor.checkConversion( |
---|
1054 | dict(p_id='p1266236341955')) |
---|
1055 | self.assertEqual(len(errs),0) |
---|
1056 | errs, inv_errs, conv_dict = self.processor.checkConversion( |
---|
1057 | dict(p_id='ABC-11-1234567890')) |
---|
1058 | self.assertEqual(len(errs),0) |
---|
1059 | errs, inv_errs, conv_dict = self.processor.checkConversion( |
---|
1060 | dict(p_id='nonsense')) |
---|
1061 | self.assertEqual(len(errs),1) |
---|
1062 | timestamp = ("%d" % int(time()*10000))[1:] |
---|
1063 | p_id = "p%s" % timestamp |
---|
1064 | errs, inv_errs, conv_dict = self.processor.checkConversion( |
---|
1065 | dict(p_id=p_id)) |
---|
1066 | self.assertEqual(len(errs),0) |
---|
1067 | |
---|
1068 | def test_import(self): |
---|
1069 | num, num_warns, fin_file, fail_file = self.processor.doImport( |
---|
1070 | self.csv_file, PAYMENT_HEADER_FIELDS,'create') |
---|
1071 | self.assertEqual(num_warns,0) |
---|
1072 | |
---|
1073 | payment = self.processor.getEntry(dict(reg_number='1', |
---|
1074 | p_id='p2907979737440'), self.app) |
---|
1075 | self.assertEqual(payment.p_id, 'p2907979737440') |
---|
1076 | self.assertTrue(payment.p_current) |
---|
1077 | cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S") |
---|
1078 | self.assertEqual(cdate, "2010-11-26 18:59:33") |
---|
1079 | self.assertEqual(str(payment.creation_date.tzinfo),'UTC') |
---|
1080 | |
---|
1081 | payment = self.processor.getEntry(dict(matric_number='100001', |
---|
1082 | p_id='p2907125937570'), self.app) |
---|
1083 | self.assertEqual(payment.p_id, 'p2907125937570') |
---|
1084 | self.assertEqual(payment.amount_auth, 19500.1) |
---|
1085 | self.assertFalse(payment.p_current) |
---|
1086 | cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S") |
---|
1087 | # Ooooh, still the old problem, see |
---|
1088 | # http://mail.dzug.org/mailman/archives/zope/2006-August/001153.html. |
---|
1089 | # WAT is interpreted as GMT-1 and not GMT+1 |
---|
1090 | self.assertEqual(cdate, "2010-11-25 21:16:33") |
---|
1091 | self.assertEqual(str(payment.creation_date.tzinfo),'UTC') |
---|
1092 | |
---|
1093 | payment = self.processor.getEntry(dict(reg_number='3', |
---|
1094 | p_id='ABC-11-1234567890'), self.app) |
---|
1095 | self.assertEqual(payment.amount_auth, 19500.6) |
---|
1096 | |
---|
1097 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
1098 | logcontent = open(self.logfile).read() |
---|
1099 | # Logging message from updateEntry |
---|
1100 | self.assertTrue( |
---|
1101 | 'INFO - system - Student Payment Processor - ' |
---|
1102 | 'sample_payment_data - K1000001 - updated: ' |
---|
1103 | 'p_item=BTECHBDT, creation_date=2010-02-15 13:19:01+00:00, ' |
---|
1104 | 'p_category=schoolfee, amount_auth=19500.0, p_current=True, ' |
---|
1105 | 'p_id=p1266236341955, r_code=00, r_amount_approved=19500.0, ' |
---|
1106 | 'p_state=paid' |
---|
1107 | in logcontent) |
---|
1108 | self.assertTrue( |
---|
1109 | 'INFO - system - Student Payment Processor - ' |
---|
1110 | 'sample_payment_data - K1000001 - updated: ' |
---|
1111 | 'p_item=BTECHBDT, creation_date=2010-02-15 13:19:01+00:00, ' |
---|
1112 | 'p_category=schoolfee, amount_auth=19500.6, p_current=True, ' |
---|
1113 | 'p_id=ABC-11-1234567890, r_code=SC, r_amount_approved=19500.0, ' |
---|
1114 | 'p_state=paid' |
---|
1115 | in logcontent) |
---|
1116 | |
---|
1117 | def test_import_update(self): |
---|
1118 | # We perform the same import twice, |
---|
1119 | # the second time in update mode. The number |
---|
1120 | # of warnings must be the same. |
---|
1121 | num, num_warns, fin_file, fail_file = self.processor.doImport( |
---|
1122 | self.csv_file, PAYMENT_HEADER_FIELDS,'create') |
---|
1123 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
1124 | num, num_warns, fin_file, fail_file = self.processor.doImport( |
---|
1125 | self.csv_file, PAYMENT_HEADER_FIELDS,'update') |
---|
1126 | self.assertEqual(num_warns,0) |
---|
1127 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
1128 | |
---|
1129 | def test_import_remove(self): |
---|
1130 | num, num_warns, fin_file, fail_file = self.processor.doImport( |
---|
1131 | self.csv_file, PAYMENT_HEADER_FIELDS,'create') |
---|
1132 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
1133 | num, num_warns, fin_file, fail_file = self.processor.doImport( |
---|
1134 | self.csv_file, PAYMENT_HEADER_FIELDS,'remove') |
---|
1135 | self.assertEqual(num_warns,0) |
---|
1136 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
1137 | logcontent = open(self.logfile).read() |
---|
1138 | self.assertTrue( |
---|
1139 | 'INFO - system - K1000001 - Payment ticket removed: p1266236341955' |
---|
1140 | in logcontent) |
---|
1141 | |
---|
1142 | def test_import_wo_pid(self): |
---|
1143 | num, num_warns, fin_file, fail_file = self.processor.doImport( |
---|
1144 | self.csv_file2, PAYMENT_CREATE_HEADER_FIELDS,'create') |
---|
1145 | self.assertEqual(num_warns,0) |
---|
1146 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
1147 | self.assertEqual(len(self.app['students']['X666666']['payments']), 50) |
---|
1148 | |
---|
1149 | class StudentVerdictProcessorTest(StudentImportExportSetup): |
---|
1150 | |
---|
1151 | def setUp(self): |
---|
1152 | super(StudentVerdictProcessorTest, self).setUp() |
---|
1153 | |
---|
1154 | # Import students with subobjects |
---|
1155 | student_file = os.path.join(self.workdir, 'sample_student_data.csv') |
---|
1156 | open(student_file, 'wb').write(STUDENT_SAMPLE_DATA) |
---|
1157 | num, num_warns, fin_file, fail_file = StudentProcessor().doImport( |
---|
1158 | student_file, STUDENT_HEADER_FIELDS) |
---|
1159 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
1160 | |
---|
1161 | # Update study courses |
---|
1162 | studycourse_file = os.path.join( |
---|
1163 | self.workdir, 'sample_studycourse_data.csv') |
---|
1164 | open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA) |
---|
1165 | processor = StudentStudyCourseProcessor() |
---|
1166 | num, num_warns, fin_file, fail_file = processor.doImport( |
---|
1167 | studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update') |
---|
1168 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
1169 | # Import study levels |
---|
1170 | self.csv_file = os.path.join( |
---|
1171 | self.workdir, 'sample_studylevel_data.csv') |
---|
1172 | open(self.csv_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA) |
---|
1173 | processor = StudentStudyLevelProcessor() |
---|
1174 | num, num_warns, fin_file, fail_file = processor.doImport( |
---|
1175 | self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create') |
---|
1176 | content = open(fail_file).read() |
---|
1177 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
1178 | |
---|
1179 | self.processor = StudentVerdictProcessor() |
---|
1180 | self.csv_file = os.path.join( |
---|
1181 | self.workdir, 'sample_verdict_data.csv') |
---|
1182 | open(self.csv_file, 'wb').write(VERDICT_SAMPLE_DATA) |
---|
1183 | return |
---|
1184 | |
---|
1185 | def test_import(self): |
---|
1186 | studycourse = self.processor.getEntry(dict(matric_number='100000'), |
---|
1187 | self.app) |
---|
1188 | self.assertEqual(studycourse['200'].level_verdict, None) |
---|
1189 | student = self.processor.getParent( |
---|
1190 | dict(matric_number='100000'), self.app) |
---|
1191 | num, num_warns, fin_file, fail_file = self.processor.doImport( |
---|
1192 | self.csv_file, VERDICT_HEADER_FIELDS,'update') |
---|
1193 | #content = open(fail_file).read() |
---|
1194 | #import pdb; pdb.set_trace() |
---|
1195 | self.assertEqual(num_warns,5) |
---|
1196 | self.assertEqual(studycourse.current_verdict, '0') |
---|
1197 | self.assertEqual(student.state, 'returning') |
---|
1198 | self.assertEqual(studycourse.current_level, 200) |
---|
1199 | self.assertEqual(studycourse['200'].level_verdict, '0') |
---|
1200 | student = self.processor.getParent( |
---|
1201 | dict(matric_number='100005'), self.app) |
---|
1202 | self.assertEqual(student.state, 'returning') |
---|
1203 | self.assertEqual(student['studycourse'].current_verdict, 'A') |
---|
1204 | self.assertEqual(studycourse.current_level, 200) |
---|
1205 | self.assertEqual(student['studycourse']['200'].validated_by, 'System') |
---|
1206 | self.assertTrue(isinstance( |
---|
1207 | student['studycourse']['200'].validation_date, datetime.datetime)) |
---|
1208 | student = self.processor.getParent( |
---|
1209 | dict(matric_number='100008'), self.app) |
---|
1210 | self.assertEqual(student['studycourse']['200'].validated_by, 'Juliana') |
---|
1211 | content = open(fail_file).read() |
---|
1212 | self.assertEqual( |
---|
1213 | content, |
---|
1214 | 'current_session,current_level,bypass_validation,current_verdict,' |
---|
1215 | 'matric_number,validated_by,--ERRORS--\r\n' |
---|
1216 | '2008,100,False,B,100001,<IGNORE>,Current level does not correspond.\r\n' |
---|
1217 | '2007,200,<IGNORE>,C,100002,<IGNORE>,Current session does not correspond.\r\n' |
---|
1218 | '2008,200,<IGNORE>,A,100003,<IGNORE>,Student in wrong state.\r\n' |
---|
1219 | '2008,200,<IGNORE>,<IGNORE>,100004,<IGNORE>,No verdict in import file.\r\n' |
---|
1220 | '2008,200,True,A,100007,<IGNORE>,Study level object is missing.\r\n' |
---|
1221 | ) |
---|
1222 | logcontent = open(self.logfile).read() |
---|
1223 | self.assertMatches( |
---|
1224 | '... INFO - system - Verdict Processor (special processor, ' |
---|
1225 | 'update only) - sample_verdict_data - X666666 - ' |
---|
1226 | 'updated: current_verdict=0...', |
---|
1227 | logcontent) |
---|
1228 | self.assertMatches( |
---|
1229 | '... INFO - system - X666666 - Returned...', |
---|
1230 | logcontent) |
---|
1231 | |
---|
1232 | shutil.rmtree(os.path.dirname(fin_file)) |
---|
1233 | |
---|
1234 | def test_suite(): |
---|
1235 | suite = unittest.TestSuite() |
---|
1236 | for testcase in [ |
---|
1237 | StudentProcessorTest,StudentStudyCourseProcessorTest, |
---|
1238 | StudentStudyLevelProcessorTest,CourseTicketProcessorTest, |
---|
1239 | PaymentProcessorTest,StudentVerdictProcessorTest]: |
---|
1240 | suite.addTest(unittest.TestLoader().loadTestsFromTestCase( |
---|
1241 | testcase |
---|
1242 | ) |
---|
1243 | ) |
---|
1244 | return suite |
---|
1245 | |
---|
1246 | |
---|