# -*- coding: utf-8 -*- ## $Id: test_batching.py 12812 2015-03-23 11:32:12Z henrik $ ## ## Copyright (C) 2014 Uli Fouquet & Henrik Bettermann ## This program is free software; you can redistribute it and/or modify ## it under the terms of the GNU General Public License as published by ## the Free Software Foundation; either version 2 of the License, or ## (at your option) any later version. ## ## This program is distributed in the hope that it will be useful, ## but WITHOUT ANY WARRANTY; without even the implied warranty of ## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the ## GNU General Public License for more details. ## ## You should have received a copy of the GNU General Public License ## along with this program; if not, write to the Free Software ## Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA ## """Unit tests for customer-related data processors. """ import os import shutil import tempfile import unittest import datetime import grok from decimal import Decimal from time import time from zope.event import notify from zope.component import createObject, queryUtility from zope.component.hooks import setSite, clearSite from zope.catalog.interfaces import ICatalog from zope.interface.verify import verifyClass, verifyObject from hurry.workflow.interfaces import IWorkflowState from waeup.ikoba.app import Company from waeup.ikoba.interfaces import ( IBatchProcessor, FatalCSVError, IUserAccount, VERIFIED) from waeup.ikoba.customers.batching import ( CustomerProcessor, CustomerSampleDocumentProcessor, SampleContractProcessor) from waeup.ikoba.customers.customer import Customer from waeup.ikoba.customers.documents import CustomerSampleDocument from waeup.ikoba.products.productoptions import ProductOption from waeup.ikoba.testing import FunctionalLayer, FunctionalTestCase CUSTOMER_SAMPLE_DATA = open( os.path.join(os.path.dirname(__file__), 'sample_customer_data.csv'), 'rb').read() CUSTOMER_HEADER_FIELDS = CUSTOMER_SAMPLE_DATA.split( '\n')[0].split(',') CUSTOMER_SAMPLE_DATA_UPDATE = open( os.path.join(os.path.dirname(__file__), 'sample_customer_data_update.csv'), 'rb').read() CUSTOMER_HEADER_FIELDS_UPDATE = CUSTOMER_SAMPLE_DATA_UPDATE.split( '\n')[0].split(',') CUSTOMER_SAMPLE_DATA_DUPLICATES = open( os.path.join(os.path.dirname(__file__), 'sample_customer_data_duplicates.csv'), 'rb').read() CUSTOMER_HEADER_FIELDS_DUPLICATES = CUSTOMER_SAMPLE_DATA_DUPLICATES.split( '\n')[0].split(',') DOCUMENT_SAMPLE_DATA = open( os.path.join(os.path.dirname(__file__), 'sample_document_data.csv'), 'rb').read() DOCUMENT_HEADER_FIELDS = DOCUMENT_SAMPLE_DATA.split( '\n')[0].split(',') CONTRACT_SAMPLE_DATA = open( os.path.join(os.path.dirname(__file__), 'sample_contract_data.csv'), 'rb').read() CONTRACT_HEADER_FIELDS = CONTRACT_SAMPLE_DATA.split( '\n')[0].split(',') class CustomerImportExportSetup(FunctionalTestCase): layer = FunctionalLayer def setup_customizable_params(self): self._contract_category = u'sample' return def setUp(self): super(CustomerImportExportSetup, self).setUp() self.setup_customizable_params() self.dc_root = tempfile.mkdtemp() self.workdir = tempfile.mkdtemp() app = Company() app['datacenter'].setStoragePath(self.dc_root) self.getRootFolder()['app'] = app self.app = self.getRootFolder()['app'] setSite(app) self.logfile = os.path.join( self.app['datacenter'].storage, 'logs', 'customers.log') #Add some products self.product = createObject('waeup.Product') self.product.product_id = u'SAM' self.product.title = u'Our Sample Product' self.product.contract_category = self._contract_category self.product.valid_from = datetime.date(2015, 12, 4) self.product.tc_dict = {'en': u'Hello world'} option = ProductOption() option.title = u'Base Fee' option.fee = Decimal('800.6') option.currency = 'USD' self.product.options = [option,] self.app['products'].addProduct(self.product) return def tearDown(self): super(CustomerImportExportSetup, self).tearDown() shutil.rmtree(self.workdir) shutil.rmtree(self.dc_root) clearSite() return def setup_for_export(self): customer = Customer() customer.customer_id = u'A111111' self.app['customers'][customer.customer_id] = self.customer = customer self.outfile = os.path.join(self.workdir, 'myoutput.csv') return def setup_customer(self, customer): customer.matric_number = u'234' customer.perm_address = u'Customerroad 21\nLagos 123456\n' customer.reg_number = u'123' customer.firstname = u'Anna' customer.lastname = u'Tester' customer.middlename = u'M.' customer.email = u'bb@bb.bb' customer.date_of_birth = datetime.date(1981, 2, 4) #customer.sex = 'f' customer.email = u'anna@sample.com' customer.phone = u'+234-123-12345' customer.notice = u'Some notice\nin lines.' customer.nationality = u'NG' # Add document document = createObject('waeup.CustomerSampleDocument') document.title = u'My Document' document.document_id = u'DOC1' customer['documents'].addDocument(document) IWorkflowState(document).setState(VERIFIED) self.document = document # Add contract contract = createObject('waeup.SampleContract') contract.contract_id = u'CON1' contract.product_object = self.product contract.document_object = self.document contract.product_options = self.product.options contract.tc_dict = {'en': u'Hello world'} contract.valid_from = datetime.date(2014, 2, 4) contract.valid_to = datetime.date(2014, 12, 4) customer['contracts'].addContract(contract) self.contract = contract return customer class CustomerProcessorTest(CustomerImportExportSetup): layer = FunctionalLayer def setUp(self): super(CustomerProcessorTest, self).setUp() # Add customer customer = Customer() self.app['customers'].addCustomer(customer) customer = self.setup_customer(customer) notify(grok.ObjectModifiedEvent(customer)) self.customer = self.app['customers'][customer.customer_id] self.processor = CustomerProcessor() self.csv_file = os.path.join(self.workdir, 'sample_customer_data.csv') self.csv_file_update = os.path.join( self.workdir, 'sample_customer_data_update.csv') self.csv_file_duplicates = os.path.join( self.workdir, 'sample_customer_data_duplicates.csv') open(self.csv_file, 'wb').write(CUSTOMER_SAMPLE_DATA) open(self.csv_file_update, 'wb').write(CUSTOMER_SAMPLE_DATA_UPDATE) open(self.csv_file_duplicates, 'wb').write(CUSTOMER_SAMPLE_DATA_DUPLICATES) def test_interface(self): # Make sure we fulfill the interface contracts. assert verifyObject(IBatchProcessor, self.processor) is True assert verifyClass( IBatchProcessor, CustomerProcessor) is True def test_parentsExist(self): self.assertFalse(self.processor.parentsExist(None, dict())) self.assertTrue(self.processor.parentsExist(None, self.app)) def test_entryExists(self): assert self.processor.entryExists( dict(customer_id='ID_NONE'), self.app) is False assert self.processor.entryExists( dict(reg_number='123'), self.app) is True def test_getParent(self): parent = self.processor.getParent(None, self.app) assert parent is self.app['customers'] def test_getEntry(self): assert self.processor.getEntry( dict(customer_id='ID_NONE'), self.app) is None assert self.processor.getEntry( dict(customer_id=self.customer.customer_id), self.app) is self.customer def test_addEntry(self): new_customer = Customer() self.processor.addEntry( new_customer, dict(), self.app) assert len(self.app['customers'].keys()) == 2 def test_checkConversion(self): # Make sure we can check conversions and that the cust_id # counter is not raised during such checks. initial_cust_id = self.app['customers']._curr_cust_id errs, inv_errs, conv_dict = self.processor.checkConversion( dict(reg_number='1', state='approved')) self.assertEqual(len(errs),0) # Empty state is allowed errs, inv_errs, conv_dict = self.processor.checkConversion( dict(reg_number='1', state='')) self.assertEqual(len(errs),0) #self.assertTrue(('state', 'no value provided') in errs) errs, inv_errs, conv_dict = self.processor.checkConversion( dict(reg_number='1', state='nonsense')) self.assertEqual(len(errs),1) self.assertTrue(('state', 'not allowed') in errs) new_cust_id = self.app['customers']._curr_cust_id self.assertEqual(initial_cust_id, new_cust_id) return def test_checkUpdateRequirements(self): # Make sure that pg customers can't be updated with wrong transition. err = self.processor.checkUpdateRequirements(self.customer, dict(reg_number='1', state='returning'), self.app) self.assertTrue(err is None) def test_delEntry(self): assert self.customer.customer_id in self.app['customers'].keys() self.processor.delEntry( dict(reg_number=self.customer.reg_number), self.app) assert self.customer.customer_id not in self.app['customers'].keys() def test_import(self): self.assertEqual(self.app['customers']._curr_cust_id, 1000001) num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file, CUSTOMER_HEADER_FIELDS) self.assertEqual(num_warns,0) self.assertEqual(len(self.app['customers']), 10) # Three empty rows have been skipped. self.assertEqual(num, 12) self.assertEqual(self.app['customers']['X666666'].reg_number,'1') self.assertEqual( self.app['customers']['X666666'].state, 'requested') # Two new customer_ids have been created. self.assertEqual(self.app['customers']._curr_cust_id, 1000003) shutil.rmtree(os.path.dirname(fin_file)) def test_import_update(self): num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file, CUSTOMER_HEADER_FIELDS) shutil.rmtree(os.path.dirname(fin_file)) num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file_update, CUSTOMER_HEADER_FIELDS_UPDATE, 'update') self.assertEqual(num_warns,0) # state has changed self.assertEqual(self.app['customers']['X666666'].state,'approved') # state has not changed self.assertEqual(self.app['customers']['Y777777'].state, 'requested') shutil.rmtree(os.path.dirname(fin_file)) def test_import_duplicate_data(self): num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file_duplicates, CUSTOMER_HEADER_FIELDS_DUPLICATES) content = open(fail_file).read() self.assertEqual(num_warns,3) self.assertEqual( content, 'reg_number,password,firstname,lastname,sex,phone,state,date_of_birth,customer_id,email,--ERRORS--\r\n' '1,,Frank,Meyer,m,1234,,1990-01-06,F123456,cc@aa.ng,reg_number: Invalid input\r\n' '3,,Uli,Schulz,m,1234,,1990-01-07,A123456,dd@aa.ng,This object already exists. Skipping.\r\n' '4,,Berta,Pieri,m,1234,,1991-01-07,A234567,aa@aa.ng,email: Invalid input\r\n' ) shutil.rmtree(os.path.dirname(fin_file)) class CustomerSampleDocumentProcessorTest(CustomerImportExportSetup): def setUp(self): super(CustomerSampleDocumentProcessorTest, self).setUp() customer = Customer() self.app['customers'].addCustomer(customer) customer = self.setup_customer(customer) notify(grok.ObjectModifiedEvent(customer)) self.customer = self.app['customers'][customer.customer_id] # Import customers with subobjects customer_file = os.path.join(self.workdir, 'sample_customer_data.csv') open(customer_file, 'wb').write(CUSTOMER_SAMPLE_DATA) num, num_warns, fin_file, fail_file = CustomerProcessor().doImport( customer_file, CUSTOMER_HEADER_FIELDS) shutil.rmtree(os.path.dirname(fin_file)) self.processor = CustomerSampleDocumentProcessor() self.csv_file = os.path.join( self.workdir, 'sample_document_data.csv') open(self.csv_file, 'wb').write(DOCUMENT_SAMPLE_DATA) def test_interface(self): # Make sure we fulfill the interface contracts. assert verifyObject(IBatchProcessor, self.processor) is True assert verifyClass( IBatchProcessor, CustomerSampleDocumentProcessor) is True def test_getEntry(self): assert self.processor.getEntry( dict(customer_id='ID_NONE', document_id='nonsense'), self.app) is None assert self.processor.getEntry( dict(customer_id=self.customer.customer_id, document_id='DOC1'), self.app) is self.customer['documents']['DOC1'] def test_delEntry(self): assert self.processor.getEntry( dict(customer_id=self.customer.customer_id, document_id='DOC1'), self.app) is self.customer['documents']['DOC1'] self.assertEqual(len(self.customer['documents'].keys()),1) self.processor.delEntry( dict(customer_id=self.customer.customer_id, document_id='DOC1'), self.app) assert self.processor.getEntry( dict(customer_id=self.customer.customer_id, document_id='DOC1'), self.app) is None self.assertEqual(len(self.customer['documents'].keys()),0) def test_addEntry(self): self.assertEqual(len(self.customer['documents'].keys()),1) document1 = createObject(u'waeup.CustomerSampleDocument') document1.document_id = u'd234' self.processor.addEntry( document1, dict(customer_id=self.customer.customer_id, document_id='d234'), self.app) self.assertEqual(len(self.customer['documents'].keys()),2) self.assertEqual(self.customer['documents']['d234'].document_id, 'd234') def test_checkConversion(self): errs, inv_errs, conv_dict = self.processor.checkConversion( dict(document_id='DOC', class_name='CustomerSampleDocument')) self.assertEqual(len(errs),0) errs, inv_errs, conv_dict = self.processor.checkConversion( dict(document_id='id with spaces', class_name='CustomerSampleDocument')) self.assertEqual(len(errs),1) errs, inv_errs, conv_dict = self.processor.checkConversion( dict(document_id='DOC', class_name='WrongDocument')) self.assertEqual(len(errs),1) def test_import(self): num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file, DOCUMENT_HEADER_FIELDS,'create') fail_file = open(fail_file).read() self.assertEqual(num_warns,1) # document_id must be unique self.assertEqual(fail_file, 'class_name,reg_number,document_id,title,--ERRORS--\r\n' 'CustomerSampleDocument,2,d5,My stolen doc,' 'This object already exists. Skipping.\r\n') document = self.processor.getEntry(dict(reg_number='1', document_id='d3'), self.app) self.assertEqual( self.app['customers']['X666666']['documents']['d3'], document) self.assertEqual(document.document_id, 'd3') document = self.processor.getEntry(dict(reg_number='3', document_id='d5'), self.app) shutil.rmtree(os.path.dirname(fin_file)) logcontent = open(self.logfile).read() # Fetch uuid docid = [i for i in self.app['customers']['X666666']['documents'].keys() if len(i) > 10][0] # Logging message from updateEntry self.assertTrue( 'INFO - system - Customer Sample Document Processor - ' 'sample_document_data - X666666 - %s - updated: title=My 4th doc' % docid in logcontent) def test_import_update(self): # We perform the same import twice, # the second time in update mode. The number # of warnings must be the same. num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file, DOCUMENT_HEADER_FIELDS,'create') shutil.rmtree(os.path.dirname(fin_file)) num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file, DOCUMENT_HEADER_FIELDS,'update') # There is one record without document_id self.assertEqual(num_warns,1) shutil.rmtree(os.path.dirname(fin_file)) def test_import_remove(self): # We perform the same import twice, # the second time in remove mode. The number # of warnings must be the same. num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file, DOCUMENT_HEADER_FIELDS,'create') shutil.rmtree(os.path.dirname(fin_file)) num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file, DOCUMENT_HEADER_FIELDS,'remove') # There is one record without document_id and one duplicate self.assertEqual(num_warns,2) shutil.rmtree(os.path.dirname(fin_file)) logcontent = open(self.logfile).read() self.assertTrue( 'INFO - system - K1000001 - Document removed: d5' in logcontent) class SampleContractProcessorTest(CustomerImportExportSetup): def setUp(self): super(SampleContractProcessorTest, self).setUp() customer = Customer() self.app['customers'].addCustomer(customer) customer = self.setup_customer(customer) notify(grok.ObjectModifiedEvent(customer)) self.customer = self.app['customers'][customer.customer_id] # Import customers with subobjects customer_file = os.path.join(self.workdir, 'sample_customer_data.csv') open(customer_file, 'wb').write(CUSTOMER_SAMPLE_DATA) num, num_warns, fin_file, fail_file = CustomerProcessor().doImport( customer_file, CUSTOMER_HEADER_FIELDS) shutil.rmtree(os.path.dirname(fin_file)) self.processor = SampleContractProcessor() self.csv_file = os.path.join( self.workdir, 'sample_contract_data.csv') open(self.csv_file, 'wb').write(CONTRACT_SAMPLE_DATA) def test_interface(self): # Make sure we fulfill the interface contracts. assert verifyObject(IBatchProcessor, self.processor) is True assert verifyClass( IBatchProcessor, SampleContractProcessor) is True def test_getEntry(self): assert self.processor.getEntry( dict(customer_id='ID_NONE', contract_id='nonsense'), self.app) is None assert self.processor.getEntry( dict(customer_id=self.customer.customer_id, contract_id='CON1'), self.app) is self.customer['contracts']['CON1'] def test_delEntry(self): assert self.processor.getEntry( dict(customer_id=self.customer.customer_id, contract_id='CON1'), self.app) is self.customer['contracts']['CON1'] self.assertEqual(len(self.customer['contracts'].keys()),1) self.processor.delEntry( dict(customer_id=self.customer.customer_id, contract_id='CON1'), self.app) assert self.processor.getEntry( dict(customer_id=self.customer.customer_id, contract_id='CON1'), self.app) is None self.assertEqual(len(self.customer['contracts'].keys()),0) def test_addEntry(self): self.assertEqual(len(self.customer['contracts'].keys()),1) contract1 = createObject(u'waeup.SampleContract') contract1.contract_id = u'CON2' self.processor.addEntry( contract1, dict(customer_id=self.customer.customer_id, contract_id='CON2'), self.app) self.assertEqual(len(self.customer['contracts'].keys()),2) self.assertEqual(self.customer['contracts']['CON2'].contract_id, 'CON2') def test_checkConversion(self): errs, inv_errs, conv_dict = self.processor.checkConversion( dict(contract_id='CON3', class_name='SampleContract', document_object='DOC1', product_object='SAM')) self.assertEqual(len(errs),0) errs, inv_errs, conv_dict = self.processor.checkConversion( dict(contract_id='id with spaces', class_name='SampleContract')) self.assertEqual(len(errs),1) errs, inv_errs, conv_dict = self.processor.checkConversion( dict(contract_id='CON3', class_name='WrongContract')) self.assertEqual(len(errs),1) def test_import(self): num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file, CONTRACT_HEADER_FIELDS,'create') self.assertEqual(num_warns,3) # contract_id must be unique fail_file = open(fail_file).read() self.assertEqual(fail_file, 'reg_number,valid_from,contract_id,title,class_name,valid_to,product_object,tc_dict,document_object,product_options,--ERRORS--\r\n' '2,,c5,,SampleContract,,SAM,,DOC1,[],This object already exists. Skipping.\r\n' '3,,c6,,SampleContract,,SAM,,NONEXISTENT,[],document_object: Invalid value\r\n' '3,,c7,,SampleContract,,NONEXISTENT,,DOC1,[],product_object: Invalid value\r\n' ) contract = self.processor.getEntry(dict(reg_number='1', contract_id='c3'), self.app) self.assertEqual( self.app['customers']['X666666']['contracts']['c3'], contract) self.assertEqual(contract.product_options[0].title, 'Base Fee') self.assertEqual(contract.product_options[0].fee, Decimal('800.6') ) self.assertEqual(contract.product_options[0].currency, 'USD') self.assertEqual(contract.contract_id, 'c3') self.assertEqual(contract.title, u'Nice Contract') self.assertEqual(contract.tc_dict, {'en':'Hello World'}) contract = self.processor.getEntry(dict(reg_number='3', contract_id='c5'), self.app) shutil.rmtree(os.path.dirname(fin_file)) logcontent = open(self.logfile).read() # Logging message from updateEntry # Fetch uuid conid = [i for i in self.app['customers']['X666666']['contracts'].keys() if len(i) > 10][0] self.assertTrue( 'INFO - system - Customer Sample Contract Processor - sample_contract_data - ' 'X666666 - %s - updated: valid_from=, title=, valid_to=, product_object=SAM, tc_dict=, document_object=DOC1, ' 'product_options=[]' % conid in logcontent) def test_import_update(self): # We perform the same import twice, # the second time in update mode. The number # of warnings must be the same. num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file, CONTRACT_HEADER_FIELDS,'create') shutil.rmtree(os.path.dirname(fin_file)) num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file, CONTRACT_HEADER_FIELDS,'update') # There are one record without contract_id and two # records with wrong object identifier. self.assertEqual(num_warns,3) shutil.rmtree(os.path.dirname(fin_file)) def test_import_remove(self): # We perform the same import twice, # the second time in remove mode. The number # of warnings must be the same. num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file, CONTRACT_HEADER_FIELDS,'create') shutil.rmtree(os.path.dirname(fin_file)) num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file, CONTRACT_HEADER_FIELDS,'remove') # There are one record without contract_id, one duplicate and two # records with wrong object identifier. self.assertEqual(num_warns,4) shutil.rmtree(os.path.dirname(fin_file)) logcontent = open(self.logfile).read() self.assertTrue( 'INFO - system - K1000001 - Contract removed: c5' in logcontent)