# -*- coding: utf-8 -*- ## $Id: test_batching.py 12143 2014-12-04 12:22:03Z 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 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, CustomerDocumentProcessor, ContractProcessor) from waeup.ikoba.customers.customer import Customer from waeup.ikoba.customers.documents import CustomerSampleDocument 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(self): super(CustomerImportExportSetup, self).setUp() 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 Samle Product' self.product.contract_category = u'sample' 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.date_of_birth = datetime.date(1981, 2, 4) #customer.sex = 'f' customer.email = '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' customer['documents'].addDocument(document) IWorkflowState(document).setState(VERIFIED) self.document = document # Add contract contract = createObject('waeup.SampleContract') contract.title = u'My Contract' contract.product_object = self.product contract.document_object = self.document customer['contracts'].addContract(contract) self.contract = contract return customer class CustomerProcessorTest(CustomerImportExportSetup): layer = FunctionalLayer def setUp(self): super(CustomerProcessorTest, self).setUp() # Add customer with subobjects 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) 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,2) 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,aa@aa.ng,reg_number: reg_number\r\n' '3,,Uli,Schulz,m,1234,,1990-01-07,A123456,aa@aa.ng,This object already exists. Skipping.\r\n' ) shutil.rmtree(os.path.dirname(fin_file)) class CustomerDocumentProcessorTest(CustomerImportExportSetup): def setUp(self): super(CustomerDocumentProcessorTest, 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 = CustomerDocumentProcessor() 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, CustomerDocumentProcessor) 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='d101'), self.app) is self.customer['documents']['d101'] def test_delEntry(self): assert self.processor.getEntry( dict(customer_id=self.customer.customer_id, document_id='d101'), self.app) is self.customer['documents']['d101'] self.assertEqual(len(self.customer['documents'].keys()),1) self.processor.delEntry( dict(customer_id=self.customer.customer_id, document_id='d101'), self.app) assert self.processor.getEntry( dict(customer_id=self.customer.customer_id, document_id='d101'), 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 = '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') document2 = createObject(u'waeup.CustomerSampleDocument') document1.document_id = 'nonsense' def test_checkConversion(self): errs, inv_errs, conv_dict = self.processor.checkConversion( dict(document_id='d126', class_name='CustomerSampleDocument')) self.assertEqual(len(errs),0) errs, inv_errs, conv_dict = self.processor.checkConversion( dict(document_id='nonsense', class_name='CustomerSampleDocument')) self.assertEqual(len(errs),1) errs, inv_errs, conv_dict = self.processor.checkConversion( dict(document_id='d127', class_name='WrongDocument')) self.assertEqual(len(errs),1) errs, inv_errs, conv_dict = self.processor.checkConversion( dict(document_id='', class_name='CustomerSampleDocument', mode='update')) 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') self.assertEqual(num_warns,1) # document_id must be unique fail_file = open(fail_file).read() self.assertEqual(fail_file, 'class_name,reg_number,document_id,title,--ERRORS--\r\n' 'CustomerSampleDocument,2,d5,My stolen doc,document_id: id exists\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() # Logging message from updateEntry self.assertTrue( 'INFO - system - CustomerDocument Processor - ' 'sample_document_data - X666666 - updated: ' 'document_id=d3, title=My first doc' 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 and one duplicate self.assertEqual(num_warns,2) 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 ContractProcessorTest(CustomerImportExportSetup): def setUp(self): super(ContractProcessorTest, 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 = ContractProcessor() 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, ContractProcessor) 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='c101'), self.app) is self.customer['contracts']['c101'] def test_delEntry(self): assert self.processor.getEntry( dict(customer_id=self.customer.customer_id, contract_id='c101'), self.app) is self.customer['contracts']['c101'] self.assertEqual(len(self.customer['contracts'].keys()),1) self.processor.delEntry( dict(customer_id=self.customer.customer_id, contract_id='c101'), self.app) assert self.processor.getEntry( dict(customer_id=self.customer.customer_id, contract_id='c101'), 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 = 'c234' self.processor.addEntry( contract1, dict(customer_id=self.customer.customer_id, contract_id='c234'), self.app) self.assertEqual(len(self.customer['contracts'].keys()),2) self.assertEqual(self.customer['contracts']['c234'].contract_id, 'c234') contract2 = createObject(u'waeup.SampleContract') contract1.contract_id = 'nonsense' def test_checkConversion(self): errs, inv_errs, conv_dict = self.processor.checkConversion( dict(contract_id='c126', class_name='SampleContract', document_object='d101', product_object='SAM')) self.assertEqual(len(errs),0) errs, inv_errs, conv_dict = self.processor.checkConversion( dict(contract_id='nonsense', class_name='SampleContract')) self.assertEqual(len(errs),1) errs, inv_errs, conv_dict = self.processor.checkConversion( dict(contract_id='c127', 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,contract_id,title,class_name,product_object,' 'document_object,--ERRORS--\r\n' '2,c5,My stolen contract,SampleContract,SAM,d101,' 'contract_id: id exists\r\n' '3,c6,My 5th contract,SampleContract,SAM,NONEXISTENT,' 'document_object: Invalid value\r\n' '3,c7,My sixt contract,SampleContract,NONEXISTENT,d101,' '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.contract_id, 'c3') 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 self.assertTrue( 'INFO - system - Contract Processor - ' 'sample_contract_data - X666666 - updated: ' 'contract_id=c3, title=My first contract' 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, one duplicate and two # records with wrong object identifier. self.assertEqual(num_warns,4) 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)