source: main/waeup.kofa/branches/uli-py3/src/waeup/kofa/applicants/tests/test_batching.py @ 17650

Last change on this file since 17650 was 15553, checked in by Henrik Bettermann, 5 years ago

Successful applicant payments do trigger 'approve' transition only
if applicant is in state 'started' and either p_category is 'application' or
applicant is special.

Add 'app_balance' payment category.

  • Property svn:keywords set to Id
File size: 25.1 KB
Line 
1## $Id: test_batching.py 15553 2019-08-19 19:32:09Z henrik $
2##
3## Copyright (C) 2011 Uli Fouquet & Henrik Bettermann
4## This program is free software; you can redistribute it and/or modify
5## it under the terms of the GNU General Public License as published by
6## the Free Software Foundation; either version 2 of the License, or
7## (at your option) any later version.
8##
9## This program is distributed in the hope that it will be useful,
10## but WITHOUT ANY WARRANTY; without even the implied warranty of
11## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
12## GNU General Public License for more details.
13##
14## You should have received a copy of the GNU General Public License
15## along with this program; if not, write to the Free Software
16## Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17##
18"""Unit tests for applicants-related data processors.
19"""
20import datetime
21import os
22import pytz
23import shutil
24import tempfile
25import unittest
26import grok
27from time import time
28from hurry.workflow.interfaces import IWorkflowState
29from zope.component.hooks import setSite, clearSite
30from zope.component import createObject
31from zope.interface.verify import verifyClass, verifyObject
32from zope.event import notify
33
34from waeup.kofa.app import University
35from waeup.kofa.applicants.batching import (
36    ApplicantsContainerProcessor, ApplicantProcessor,
37    ApplicantOnlinePaymentProcessor)
38from waeup.kofa.applicants.container import ApplicantsContainer
39from waeup.kofa.applicants.applicant import Applicant
40from waeup.kofa.university.faculty import Faculty
41from waeup.kofa.university.department import Department
42from waeup.kofa.testing import FunctionalLayer, FunctionalTestCase
43from waeup.kofa.interfaces import (
44    IBatchProcessor, IUserAccount, DuplicationError)
45from waeup.kofa.applicants.workflow import CREATED
46
47
48# Sample data we can use in tests...
49APPS_CONTAINER_SAMPLE_DATA = open(
50    os.path.join(os.path.dirname(__file__), 'sample_container_data.csv'),
51    'rb').read()
52
53# The header fields of the above CSV snippet
54APPS_CONTAINER_HEADER_FIELDS = APPS_CONTAINER_SAMPLE_DATA.split(
55    '\n')[0].split(',')
56
57# The same for applicants
58APPLICANT_SAMPLE_DATA = open(
59    os.path.join(os.path.dirname(__file__), 'sample_applicant_data.csv'),
60    'rb').read()
61FAULTY_APPLICANT_SAMPLE_DATA = open(
62    os.path.join(os.path.dirname(__file__),
63                 'sample_faulty_applicant_data.csv'), 'rb').read()
64
65APPLICANT_HEADER_FIELDS = APPLICANT_SAMPLE_DATA.split(
66    '\n')[0].split(',')
67
68APPLICANT_SAMPLE_DATA_UPDATE = open(
69    os.path.join(os.path.dirname(__file__),
70                 'sample_applicant_data_update.csv'), 'rb').read()
71
72APPLICANT_SAMPLE_DATA_UPDATE2 = open(
73    os.path.join(os.path.dirname(__file__),
74                 'sample_applicant_data_update2.csv'), 'rb').read()
75
76APPLICANT_HEADER_FIELDS_UPDATE = APPLICANT_SAMPLE_DATA_UPDATE.split(
77    '\n')[0].split(',')
78
79APPLICANT_HEADER_FIELDS_UPDATE2 = APPLICANT_SAMPLE_DATA_UPDATE2.split(
80    '\n')[0].split(',')
81
82PAYMENT_SAMPLE_DATA = open(
83    os.path.join(os.path.dirname(__file__), 'sample_payment_data.csv'),
84    'rb').read()
85
86PAYMENT_HEADER_FIELDS = PAYMENT_SAMPLE_DATA.split(
87    '\n')[0].split(',')
88
89class ApplicantsContainerProcessorTest(FunctionalTestCase):
90
91    layer = FunctionalLayer
92
93    def setUp(self):
94        super(ApplicantsContainerProcessorTest, self).setUp()
95
96        # Setup a sample site for each test
97        app = University()
98        self.dc_root = tempfile.mkdtemp()
99        app['datacenter'].setStoragePath(self.dc_root)
100
101        # Prepopulate the ZODB...
102        self.getRootFolder()['app'] = app
103        self.app = self.getRootFolder()['app']
104        self.container = ApplicantsContainer()
105        self.container.code = u'dp2011'
106        self.app['applicants']['dp2011'] = self.container
107
108        self.processor = ApplicantsContainerProcessor()
109        self.workdir = tempfile.mkdtemp()
110        self.csv_file = os.path.join(self.workdir, 'sampledata.csv')
111        open(self.csv_file, 'wb').write(APPS_CONTAINER_SAMPLE_DATA)
112        setSite(self.app)
113        return
114
115    def tearDown(self):
116        super(ApplicantsContainerProcessorTest, self).tearDown()
117        shutil.rmtree(self.workdir)
118        shutil.rmtree(self.dc_root)
119        clearSite()
120        return
121
122    def test_interface(self):
123        # Make sure we fulfill the interface contracts.
124        assert verifyObject(IBatchProcessor, self.processor) is True
125        assert verifyClass(
126            IBatchProcessor, ApplicantsContainerProcessor) is True
127
128    def test_parentsExist(self):
129        assert self.processor.parentsExist(None, dict()) is False
130        assert self.processor.parentsExist(None, self.app) is True
131
132    def test_entryExists(self):
133        assert self.processor.entryExists(
134            dict(code='REG_NONE'), self.app) is False
135        assert self.processor.entryExists(
136            dict(code='dp2011'), self.app) is True
137
138    def test_getParent(self):
139        parent = self.processor.getParent(None, self.app)
140        assert parent is self.app['applicants']
141
142    def test_getEntry(self):
143        assert self.processor.getEntry(
144            dict(code='REG_NONE'), self.app) is None
145        assert self.processor.getEntry(
146            dict(code='dp2011'), self.app) is self.container
147
148    def test_addEntry(self):
149        self.processor.addEntry(
150            'New application', dict(code='dp2012'), self.app)
151        assert self.app['applicants']['dp2012'] == 'New application'
152
153    def test_delEntry(self):
154        self.processor.delEntry(dict(code='dp2011'), self.app)
155        assert 'dp2011' not in self.app['applicants'].keys()
156
157    def test_import(self):
158        # Do a real import
159        # see local sample_container.csv file for input
160        num, num_warns, fin_file, fail_file = self.processor.doImport(
161            self.csv_file, APPS_CONTAINER_HEADER_FIELDS)
162        avail_containers = [x for x in self.app['applicants'].keys()]
163        container = self.app['applicants'].get('app2017', None)
164        container2 = self.app['applicants'].get('app2018', None)
165        self.assertTrue(container is not None)
166        self.assertTrue(container2 is not None)
167
168        # check attributes
169        self.assertEqual(container.code, u'app2017')
170        self.assertEqual(container.title, u'General Studies')
171        self.assertEqual(container.prefix, u'app')
172        self.assertEqual(container.year, 2017)
173        self.assertEqual(container.application_category, 'basic')
174        self.assertEqual(
175            container.description,
176            u'This text can been seen by anonymous users.\n'
177            u'>>de<<\nDieser Text kann von anonymen Benutzern '
178            u'gelesen werden.')
179        self.assertEqual(container.startdate,
180                         datetime.datetime(2012, 3, 1, 0, 0, tzinfo=pytz.utc))
181        self.assertEqual(container.enddate,
182                         datetime.datetime(2012, 4, 25, 0, 0, tzinfo=pytz.utc))
183        shutil.rmtree(os.path.dirname(fin_file))
184
185class ApplicantImportExportSetup(FunctionalTestCase):
186
187    layer = FunctionalLayer
188
189    def setUp(self):
190        super(ApplicantImportExportSetup, self).setUp()
191        # Setup a sample site for each test
192        app = University()
193        self.dc_root = tempfile.mkdtemp()
194        app['datacenter'].setStoragePath(self.dc_root)
195
196        # Prepopulate the ZODB...
197        self.getRootFolder()['app'] = app
198        # we add the site immediately after creation to the
199        # ZODB. Catalogs and other local utilities are not setup
200        # before that step.
201        self.app = self.getRootFolder()['app']
202        # Set site here. Some of the following setup code might need
203        # to access grok.getSite() and should get our new app then
204        setSite(app)
205
206        # Add an applicants container
207        self.container = ApplicantsContainer()
208        self.container.code = u'dp2011'
209        self.container.application_category = u'basic'
210        self.app['applicants']['dp2011'] = self.container
211
212        # Populate university
213        self.certificate = createObject('waeup.Certificate')
214        self.certificate.code = 'CERT1'
215        self.certificate.application_category = 'basic'
216        self.certificate.start_level = 100
217        self.certificate.end_level = 500
218        self.app['faculties']['fac1'] = Faculty()
219        self.app['faculties']['fac1']['dep1'] = Department()
220        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
221            self.certificate)
222        self.certificate2 = createObject('waeup.Certificate')
223        self.certificate2.code = 'CERT2'
224        self.certificate2.application_category = 'xyz'
225        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
226            self.certificate2)
227
228        # Add applicant with subobjects
229        applicant = Applicant()
230        applicant.firstname = u'Anna'
231        applicant.lastname = u'Tester'
232        self.app['applicants']['dp2011'].addApplicant(applicant)
233        self.application_number = applicant.application_number
234        self.applicant = self.app['applicants']['dp2011'][
235            self.application_number]
236        self.workdir = tempfile.mkdtemp()
237
238        self.logfile = os.path.join(
239            self.app['datacenter'].storage, 'logs', 'applicants.log')
240        return
241
242    def tearDown(self):
243        super(ApplicantImportExportSetup, self).tearDown()
244        shutil.rmtree(self.workdir)
245        shutil.rmtree(self.dc_root)
246        clearSite()
247        return
248
249class ApplicantProcessorTest(ApplicantImportExportSetup):
250
251    layer = FunctionalLayer
252
253    def setUp(self):
254        super(ApplicantProcessorTest, self).setUp()
255        self.processor = ApplicantProcessor()
256        self.csv_file = os.path.join(self.workdir, 'sample_applicant_data.csv')
257        self.csv_file_faulty = os.path.join(self.workdir,
258                                            'faulty_applicant_data.csv')
259        self.csv_file_update = os.path.join(
260            self.workdir, 'sample_applicant_data_update.csv')
261        self.csv_file_update2 = os.path.join(
262            self.workdir, 'sample_applicant_data_update2.csv')
263        open(self.csv_file, 'wb').write(APPLICANT_SAMPLE_DATA)
264        open(self.csv_file_faulty, 'wb').write(FAULTY_APPLICANT_SAMPLE_DATA)
265        open(self.csv_file_update, 'wb').write(APPLICANT_SAMPLE_DATA_UPDATE)
266        open(self.csv_file_update2, 'wb').write(APPLICANT_SAMPLE_DATA_UPDATE2)
267
268    def test_interface(self):
269        # Make sure we fulfill the interface contracts.
270        assert verifyObject(IBatchProcessor, self.processor) is True
271        assert verifyClass(
272            IBatchProcessor, ApplicantProcessor) is True
273
274    def test_entryExists(self):
275        assert self.processor.entryExists(
276            dict(container_code='dp2011', application_number='999'),
277            self.app) is False
278
279    def test_getEntry(self):
280        applicant = self.processor.getEntry(
281            dict(container_code='dp2011',
282                 application_number=self.application_number), self.app)
283        self.assertEqual(applicant.applicant_id, self.applicant.applicant_id)
284
285    def test_addEntry(self):
286        new_applicant = Applicant()
287        self.processor.addEntry(
288            new_applicant, dict(container_code='dp2011'), self.app)
289        assert len(self.app['applicants']['dp2011'].keys()) == 2
290
291    def test_delEntry(self):
292        assert self.application_number in self.app[
293            'applicants']['dp2011'].keys()
294        self.processor.delEntry(
295            dict(container_code='dp2011',
296                application_number=self.application_number), self.app)
297        assert self.application_number not in self.app[
298            'applicants']['dp2011'].keys()
299
300    def test_import(self):
301        num, num_warns, fin_file, fail_file = self.processor.doImport(
302            self.csv_file, APPLICANT_HEADER_FIELDS)
303        self.assertEqual(num_warns,0)
304        keys = self.app['applicants']['dp2011'].keys()
305        assert len(keys) == 5
306        container = self.app['applicants']['dp2011']
307        assert  container.__implemented__.__name__ == (
308            'waeup.kofa.applicants.container.ApplicantsContainer')
309        applicant = container[keys[0]]
310        assert applicant.__implemented__.__name__ == (
311            'waeup.kofa.applicants.applicant.Applicant')
312        logcontent = open(self.logfile).read()
313        # Logging message from updateEntry,
314        # create applicant with given application_number
315        self.assertTrue(
316            'Applicant Processor - sample_applicant_data - imported: '
317            'applicant_id=dp2011_1234, password=mypwd1, '
318            'reg_number=1001, firstname=Aaren, middlename=Peter, lastname=Pieri, '
319            'sex=m, course1=CERT1, date_of_birth=1990-01-02, email=xx@yy.zz' in
320            logcontent)
321        # create applicant with random application_number which is
322        # not shown in the log file
323        self.assertTrue(
324            'Applicant Processor - sample_applicant_data - imported: '
325            'reg_number=1003, firstname=Aaren, '
326            'middlename=Alfons, lastname=Berson, sex=m, course1=CERT1, '
327            'date_of_birth=1990-01-04, email=xx@yy.zz' in
328            logcontent)
329        # Logging message from handle_applicant_transition_event
330        self.assertTrue(
331            'dp2011_1234 - Application initialized' in
332            logcontent)
333        shutil.rmtree(os.path.dirname(fin_file))
334
335    def test_import_faulty(self):
336        num, num_warns, fin_file, fail_file = self.processor.doImport(
337            self.csv_file_faulty, APPLICANT_HEADER_FIELDS)
338        # we cannot import data with faulty dates. A date is faulty
339        # when in format xx/yy/zzzz as we cannot say whether it is
340        # meant as dd/mm/yyyy or mm/dd/yyyy. We therefore require yyyy-mm-dd
341        for applicant in self.app['applicants']['dp2011'].values():
342            if applicant.date_of_birth == datetime.date(1990, 1, 2):
343                self.fail(
344                    'Wrong birthdate of imported applicant '
345                    '(1990-01-02, should be: 1990-02-01)')
346        self.assertEqual(num_warns,4)
347        fail_contents = open(fail_file, 'rb').read()
348        # CERT2 is in wrong category ...
349        self.assertTrue('course1: wrong application category' in fail_contents)
350        # ... and CERT3 does not exist.
351        self.assertTrue('course1: Invalid value' in fail_contents)
352        # Conversion checking already fails because Mister or Miss No's
353        # container does not exist.
354        self.assertTrue('no@yy.zz,container: not found' in fail_contents)
355        self.assertTrue('nobody@yy.zz,container: not found' in fail_contents)
356        shutil.rmtree(os.path.dirname(fail_file))
357        return
358
359    def test_import_update(self):
360        num, num_warns, fin_file, fail_file = self.processor.doImport(
361            self.csv_file, APPLICANT_HEADER_FIELDS)
362        shutil.rmtree(os.path.dirname(fin_file))
363        num, num_warns, fin_file, fail_file = self.processor.doImport(
364            self.csv_file_update, APPLICANT_HEADER_FIELDS_UPDATE, 'update')
365        self.assertEqual(num_warns,0)
366        # The middlename import value was None.
367        # Confirm that middlename has not been deleted.
368        container = self.app['applicants']['dp2011']
369        self.assertEqual(container['1234'].middlename, 'Peter')
370        # state of Pieri has not changed
371        self.assertEqual(container['1234'].state,'initialized')
372        # state of Finau has changed
373        self.assertEqual(container['2345'].state,'admitted')
374        # password of Pieri has been set
375        self.assertTrue(IUserAccount(container['1234']).checkPassword('mypwd1'))
376        # password of Finau is still unset
377        self.assertEqual(IUserAccount(container['2345']).password,None)
378        # password of Simon was encrypted already
379        self.assertTrue(
380            IUserAccount(container['4567']).checkPassword('mypwd1'))
381        # reg_number of Finau has changed
382        self.assertEqual(container['2345'].reg_number, '6666')
383        logcontent = open(self.logfile).read()
384        # Logging message from updateEntry,
385        # reg_number is locator
386        self.assertTrue(
387            'Applicant Processor - sample_applicant_data_update - updated: '
388            'reg_number=1001, firstname=Aaren' in
389            logcontent)
390        # applicant_id is locator
391        self.assertTrue(
392            'Applicant Processor - sample_applicant_data_update - updated: '
393            'state=admitted, reg_number=6666, '
394            'firstname=Alfons, applicant_id=dp2011_2345' in
395            logcontent)
396        shutil.rmtree(os.path.dirname(fin_file))
397
398        # Now we import another file which clears all middlename attributes
399        # and uses the new reg_number as locator. This test also checks
400        # if the catalog has been informed about the reg_no change and if
401        # applicants in state created are really blocked.
402        IWorkflowState(container['4567']).setState(CREATED)
403        num, num_warns, fin_file, fail_file = self.processor.doImport(
404            self.csv_file_update2, APPLICANT_HEADER_FIELDS_UPDATE2, 'update')
405        failcontent = open(fail_file).read()
406        self.assertTrue('Applicant is blocked' in failcontent)
407        self.assertEqual(num_warns,1)
408        # Middlename is cleared.
409        assert container['1234'].middlename is None
410        # Firstname of applicant in state created isn't changed.
411        self.assertEqual(container['4567'].firstname, 'Simon')
412        shutil.rmtree(os.path.dirname(fin_file))
413
414    def test_import_remove(self):
415        num, num_warns, fin_file, fail_file = self.processor.doImport(
416            self.csv_file, APPLICANT_HEADER_FIELDS)
417        shutil.rmtree(os.path.dirname(fin_file))
418        num, num_warns, fin_file, fail_file = self.processor.doImport(
419            self.csv_file_update, APPLICANT_HEADER_FIELDS_UPDATE, 'remove')
420        self.assertEqual(num_warns,0)
421        logcontent = open(self.logfile).read()
422        # Logging message from handle_applicant_transition_event
423        self.assertTrue(
424            'dp2011_1234 - Application record removed' in
425            logcontent)
426        shutil.rmtree(os.path.dirname(fin_file))
427
428class PaymentProcessorTest(ApplicantImportExportSetup):
429
430    def setUp(self):
431        super(PaymentProcessorTest, self).setUp()
432
433        applicant = Applicant()
434        applicant.firstname = u'Anna2'
435        applicant.lastname = u'Tester'
436        applicant.applicant_id = u'dp2011_1234'
437        self.app['applicants']['dp2011'].addApplicant(applicant)
438        payment = createObject(u'waeup.ApplicantOnlinePayment')
439        payment.p_id = 'p120'
440        payment.p_session = 2012
441        payment.p_category = 'application'
442        payment.p_state = 'paid'
443        applicant['p120'] = payment
444        self.applicant2 = applicant
445        self.processor = ApplicantOnlinePaymentProcessor()
446        self.csv_file = os.path.join(
447            self.workdir, 'sample_payment_data.csv')
448        open(self.csv_file, 'wb').write(PAYMENT_SAMPLE_DATA)
449
450    def test_interface(self):
451        # Make sure we fulfill the interface contracts.
452        assert verifyObject(IBatchProcessor, self.processor) is True
453        assert verifyClass(
454            IBatchProcessor, ApplicantOnlinePaymentProcessor) is True
455
456    def test_getEntry(self):
457        assert self.processor.getEntry(
458            dict(applicant_id='ID_NONE', p_id='nonsense'), self.app) is None
459        assert self.processor.getEntry(
460            dict(applicant_id=self.applicant2.applicant_id, p_id='p120'),
461            self.app) is self.applicant2['p120']
462        assert self.processor.getEntry(
463            dict(applicant_id=self.applicant2.applicant_id, p_id='p120#'),
464            self.app) is self.applicant2['p120']
465
466    def test_delEntry(self):
467        assert self.processor.getEntry(
468            dict(applicant_id=self.applicant2.applicant_id, p_id='p120'),
469            self.app) is self.applicant2['p120']
470        self.assertEqual(len(self.applicant2.keys()),1)
471        self.processor.delEntry(
472            dict(applicant_id=self.applicant2.applicant_id, p_id='p120'),
473            self.app)
474        assert self.processor.getEntry(
475            dict(applicant_id=self.applicant2.applicant_id, p_id='p120'),
476            self.app) is None
477        self.assertEqual(len(self.applicant.keys()),0)
478
479    def test_addEntry(self):
480        self.assertEqual(len(self.applicant2.keys()),1)
481        self.processor.delEntry(
482            dict(applicant_id=self.applicant2.applicant_id, p_id='p120'),
483            self.app)
484        payment1 = createObject(u'waeup.ApplicantOnlinePayment')
485        payment1.p_category = 'application'
486        payment1.p_id = 'p234'
487        self.processor.addEntry(
488            payment1, dict(applicant_id=self.applicant2.applicant_id, p_id='p234'),
489            self.app)
490        self.assertEqual(len(self.applicant2.keys()),1)
491        self.assertEqual(self.applicant2['p234'].p_id, 'p234')
492        # Same payment must not exist.
493        payment1.p_state = 'paid'
494        payment2 = createObject(u'waeup.ApplicantOnlinePayment')
495        payment2.p_id = 'p456'
496        payment2.p_category = 'application'
497        self.assertRaises(
498            DuplicationError, self.processor.addEntry, payment2,
499            dict(applicant_id=self.applicant2.applicant_id, p_id='p456'), self.app)
500        # But we can add a ticket with another p_category.
501        payment2.p_category = 'app_balance'
502        self.processor.addEntry(
503            payment2, dict(applicant_id=self.applicant2.applicant_id, p_id='p456'),
504            self.app)
505        self.assertEqual(len(self.applicant2.keys()),2)
506        self.assertEqual(self.applicant2['p456'].p_id, 'p456')
507
508    def test_checkConversion(self):
509        errs, inv_errs, conv_dict = self.processor.checkConversion(
510            dict(p_id='<IGNORE>'), mode='create')
511        self.assertEqual(len(errs),0)
512        errs, inv_errs, conv_dict = self.processor.checkConversion(
513            dict(p_id='<IGNORE>'), mode='update')
514        self.assertEqual(len(errs),1)
515        self.assertEqual(errs[0], ('p_id', u'missing'))
516        errs, inv_errs, conv_dict = self.processor.checkConversion(
517            dict(p_id='p1266236341955'))
518        self.assertEqual(len(errs),0)
519        errs, inv_errs, conv_dict = self.processor.checkConversion(
520            dict(p_id='nonsense'))
521        self.assertEqual(len(errs),1)
522        self.assertEqual(errs[0], ('p_id', u'invalid length'))
523        timestamp = ("%d" % int(time()*10000))[1:]
524        p_id = "p%s" % timestamp
525        errs, inv_errs, conv_dict = self.processor.checkConversion(
526            dict(p_id=p_id))
527        self.assertEqual(len(errs),0)
528        dup_payment = createObject(u'waeup.ApplicantOnlinePayment')
529        dup_payment.p_id = 'p1266236341955'
530        self.applicant2[dup_payment.p_id] = dup_payment
531        errs, inv_errs, conv_dict = self.processor.checkConversion(
532            dict(p_id='p1266236341955'), mode='create')
533        self.assertEqual(len(errs),1)
534        self.assertEqual(errs[0], ('p_id', u'p_id exists in dp2011_1234 '))
535
536    def test_import(self):
537        num, num_warns, fin_file, fail_file = self.processor.doImport(
538            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
539        self.assertEqual(num_warns,2)
540        fail_file = open(fail_file).read()
541        self.assertTrue('Payment has already been made' in fail_file)
542        self.processor.delEntry(
543            dict(applicant_id=self.applicant2.applicant_id, p_id='p120'),
544            self.app)
545        num, num_warns, fin_file, fail_file = self.processor.doImport(
546            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
547        # Both records can be imported now.
548        self.assertEqual(num_warns,0)
549        # One with imported and known p_id ...
550        payment = self.processor.getEntry(dict(applicant_id='dp2011_1234',
551            p_id='p1266236341955'), self.app)
552        self.assertEqual(payment.p_id, 'p1266236341955')
553        cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S")
554        # Ooooh, still the old problem, see
555        # http://mail.dzug.org/mailman/archives/zope/2006-August/001153.html.
556        # WAT is interpreted as GMT-1 and not GMT+1
557        self.assertEqual(cdate, '2010-11-25 21:16:33')
558        self.assertEqual(str(payment.creation_date.tzinfo),'UTC')
559        # ... the other one with generated p_id.
560        p_id_failed = self.applicant2.keys()[1]
561        payment_failed = self.processor.getEntry(dict(applicant_id='dp2011_1234',
562            p_id=p_id_failed), self.app)
563        self.assertEqual(payment_failed.p_state, 'failed')
564        self.assertEqual(payment_failed.amount_auth, 10500.1)
565        shutil.rmtree(os.path.dirname(fin_file))
566        logcontent = open(self.logfile).read()
567        # Logging message from updateEntry
568        self.assertTrue(
569            'INFO - system - ApplicantOnlinePayment Processor - dp2011_1234 - '
570            'previous update cancelled' in logcontent)
571        self.assertTrue(
572            'INFO - system - ApplicantOnlinePayment Processor - '
573            'sample_payment_data - dp2011_1234 - updated: p_id=p1266236341955, '
574            'creation_date=2010-11-25 21:16:33.757000+00:00, '
575            'r_amount_approved=19500.1, p_category=application, '
576            'amount_auth=19500.1, p_session=2015, p_state=paid' in logcontent)
577        self.assertTrue(
578            'INFO - system - ApplicantOnlinePayment Processor - '
579            'sample_payment_data - dp2011_1234 - updated: p_id=%s, '
580            'creation_date=2011-11-25 21:16:33.757000+00:00, '
581            'r_amount_approved=19500.1, p_category=application, '
582            'amount_auth=10500.1, p_session=2016, p_state=failed'
583            % p_id_failed in logcontent)
Note: See TracBrowser for help on using the repository browser.