source: main/waeup.kofa/branches/uli-upgrade-beaker/src/waeup/kofa/students/tests/test_batching.py @ 17238

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

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