source: main/waeup.kofa/trunk/src/waeup/kofa/students/tests/test_batching.py @ 16828

Last change on this file since 16828 was 16828, checked in by Henrik Bettermann, 3 years ago

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