source: main/waeup.ikoba/trunk/src/waeup/ikoba/customers/tests/test_batching.py @ 11962

Last change on this file since 11962 was 11958, checked in by Henrik Bettermann, 10 years ago

Add components for customer management. Some tests are still missing.
File size: 9.5 KB
Line 
1# -*- coding: utf-8 -*-
2## $Id: test_batching.py 11756 2014-07-09 12:46:08Z 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 customer-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.ikoba.app import Company
36from waeup.ikoba.interfaces import IBatchProcessor, FatalCSVError, IUserAccount
37from waeup.ikoba.customers.batching import CustomerProcessor
38from waeup.ikoba.customers.customer import Customer
39from waeup.ikoba.testing import FunctionalLayer, FunctionalTestCase
40
41CUSTOMER_SAMPLE_DATA = open(
42    os.path.join(os.path.dirname(__file__), 'sample_customer_data.csv'),
43    'rb').read()
44
45CUSTOMER_HEADER_FIELDS = CUSTOMER_SAMPLE_DATA.split(
46    '\n')[0].split(',')
47
48CUSTOMER_SAMPLE_DATA_UPDATE = open(
49    os.path.join(os.path.dirname(__file__), 'sample_customer_data_update.csv'),
50    'rb').read()
51
52CUSTOMER_HEADER_FIELDS_UPDATE = CUSTOMER_SAMPLE_DATA_UPDATE.split(
53    '\n')[0].split(',')
54
55CUSTOMER_SAMPLE_DATA_DUPLICATES = open(
56    os.path.join(os.path.dirname(__file__),
57                 'sample_customer_data_duplicates.csv'),
58    'rb').read()
59
60CUSTOMER_HEADER_FIELDS_DUPLICATES = CUSTOMER_SAMPLE_DATA_DUPLICATES.split(
61    '\n')[0].split(',')
62
63class CustomerImportExportSetup(FunctionalTestCase):
64
65    layer = FunctionalLayer
66
67    def setUp(self):
68        super(CustomerImportExportSetup, self).setUp()
69        self.dc_root = tempfile.mkdtemp()
70        self.workdir = tempfile.mkdtemp()
71        app = Company()
72        app['datacenter'].setStoragePath(self.dc_root)
73        self.getRootFolder()['app'] = app
74        self.app = self.getRootFolder()['app']
75        setSite(app)
76
77        self.logfile = os.path.join(
78            self.app['datacenter'].storage, 'logs', 'customers.log')
79        return
80
81    def tearDown(self):
82        super(CustomerImportExportSetup, self).tearDown()
83        shutil.rmtree(self.workdir)
84        shutil.rmtree(self.dc_root)
85        clearSite()
86        return
87
88    def setup_for_export(self):
89        customer = Customer()
90        customer.customer_id = u'A111111'
91        self.app['customers'][customer.customer_id] = self.customer = customer
92        self.outfile = os.path.join(self.workdir, 'myoutput.csv')
93        return
94
95    def setup_customer(self, customer):
96        # set predictable values for `customer`
97        customer.matric_number = u'234'
98        customer.perm_address = u'Customerroad 21\nLagos 123456\n'
99        customer.reg_number = u'123'
100        customer.firstname = u'Anna'
101        customer.lastname = u'Tester'
102        customer.middlename = u'M.'
103        customer.date_of_birth = datetime.date(1981, 2, 4)
104        #customer.sex = 'f'
105        customer.email = 'anna@sample.com'
106        customer.phone = u'+234-123-12345'
107        customer.notice = u'Some notice\nin lines.'
108        customer.nationality = u'NG'
109
110        return customer
111
112class CustomerProcessorTest(CustomerImportExportSetup):
113
114    layer = FunctionalLayer
115
116    def setUp(self):
117        super(CustomerProcessorTest, self).setUp()
118
119        # Add customer with subobjects
120        customer = Customer()
121        self.app['customers'].addCustomer(customer)
122        customer = self.setup_customer(customer)
123        notify(grok.ObjectModifiedEvent(customer))
124        self.customer = self.app['customers'][customer.customer_id]
125
126        self.processor = CustomerProcessor()
127        self.csv_file = os.path.join(self.workdir, 'sample_customer_data.csv')
128        self.csv_file_update = os.path.join(
129            self.workdir, 'sample_customer_data_update.csv')
130        self.csv_file_duplicates = os.path.join(
131            self.workdir, 'sample_customer_data_duplicates.csv')
132        open(self.csv_file, 'wb').write(CUSTOMER_SAMPLE_DATA)
133        open(self.csv_file_update, 'wb').write(CUSTOMER_SAMPLE_DATA_UPDATE)
134        open(self.csv_file_duplicates, 'wb').write(CUSTOMER_SAMPLE_DATA_DUPLICATES)
135
136    def test_interface(self):
137        # Make sure we fulfill the interface contracts.
138        assert verifyObject(IBatchProcessor, self.processor) is True
139        assert verifyClass(
140            IBatchProcessor, CustomerProcessor) is True
141
142    def test_parentsExist(self):
143        self.assertFalse(self.processor.parentsExist(None, dict()))
144        self.assertTrue(self.processor.parentsExist(None, self.app))
145
146    def test_entryExists(self):
147        assert self.processor.entryExists(
148            dict(customer_id='ID_NONE'), self.app) is False
149        assert self.processor.entryExists(
150            dict(reg_number='123'), self.app) is True
151
152    def test_getParent(self):
153        parent = self.processor.getParent(None, self.app)
154        assert parent is self.app['customers']
155
156    def test_getEntry(self):
157        assert self.processor.getEntry(
158            dict(customer_id='ID_NONE'), self.app) is None
159        assert self.processor.getEntry(
160            dict(customer_id=self.customer.customer_id), self.app) is self.customer
161
162    def test_addEntry(self):
163        new_customer = Customer()
164        self.processor.addEntry(
165            new_customer, dict(), self.app)
166        assert len(self.app['customers'].keys()) == 2
167
168    def test_checkConversion(self):
169        # Make sure we can check conversions and that the cust_id
170        # counter is not raised during such checks.
171        initial_cust_id = self.app['customers']._curr_cust_id
172        errs, inv_errs, conv_dict = self.processor.checkConversion(
173            dict(reg_number='1', state='admitted'))
174        self.assertEqual(len(errs),0)
175        # Empty state is allowed
176        errs, inv_errs, conv_dict = self.processor.checkConversion(
177            dict(reg_number='1', state=''))
178        self.assertEqual(len(errs),0)
179        #self.assertTrue(('state', 'no value provided') in errs)
180        errs, inv_errs, conv_dict = self.processor.checkConversion(
181            dict(reg_number='1', state='nonsense'))
182        self.assertEqual(len(errs),1)
183        self.assertTrue(('state', 'not allowed') in errs)
184        new_cust_id = self.app['customers']._curr_cust_id
185        self.assertEqual(initial_cust_id, new_cust_id)
186        return
187
188    def test_checkUpdateRequirements(self):
189        # Make sure that pg customers can't be updated with wrong transition.
190        err = self.processor.checkUpdateRequirements(self.customer,
191            dict(reg_number='1', state='returning'), self.app)
192        self.assertTrue(err is None)
193
194    def test_delEntry(self):
195        assert self.customer.customer_id in self.app['customers'].keys()
196        self.processor.delEntry(
197            dict(reg_number=self.customer.reg_number), self.app)
198        assert self.customer.customer_id not in self.app['customers'].keys()
199
200    def test_import(self):
201        self.assertEqual(self.app['customers']._curr_cust_id, 1000001)
202        num, num_warns, fin_file, fail_file = self.processor.doImport(
203            self.csv_file, CUSTOMER_HEADER_FIELDS)
204        self.assertEqual(num_warns,0)
205        self.assertEqual(len(self.app['customers']), 10)
206        self.assertEqual(self.app['customers']['X666666'].reg_number,'1')
207        self.assertEqual(
208            self.app['customers']['X666666'].state, 'courses validated')
209        # Two new customer_ids have been created.
210        self.assertEqual(self.app['customers']._curr_cust_id, 1000003)
211        shutil.rmtree(os.path.dirname(fin_file))
212
213    def test_import_update(self):
214        num, num_warns, fin_file, fail_file = self.processor.doImport(
215            self.csv_file, CUSTOMER_HEADER_FIELDS)
216        shutil.rmtree(os.path.dirname(fin_file))
217        num, num_warns, fin_file, fail_file = self.processor.doImport(
218            self.csv_file_update, CUSTOMER_HEADER_FIELDS_UPDATE, 'update')
219        self.assertEqual(num_warns,0)
220        # state has changed
221        self.assertEqual(self.app['customers']['X666666'].state,'admitted')
222        # state has not changed
223        self.assertEqual(self.app['customers']['Y777777'].state,
224                         'courses validated')
225        shutil.rmtree(os.path.dirname(fin_file))
226
227    def test_import_duplicate_data(self):
228        num, num_warns, fin_file, fail_file = self.processor.doImport(
229            self.csv_file_duplicates, CUSTOMER_HEADER_FIELDS_DUPLICATES)
230        content = open(fail_file).read()
231        self.assertEqual(num_warns,2)
232        self.assertEqual(
233            content,
234            'reg_number,password,firstname,lastname,sex,phone,state,date_of_birth,customer_id,email,--ERRORS--\r\n'
235            '1,,Frank,Meyer,m,1234,,1990-01-06,F123456,aa@aa.ng,reg_number: reg_number\r\n'
236            '3,,Uli,Schulz,m,1234,,1990-01-07,A123456,aa@aa.ng,This object already exists. Skipping.\r\n'
237
238            )
239        shutil.rmtree(os.path.dirname(fin_file))
240
Note: See TracBrowser for help on using the repository browser.