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

# -*- coding: utf-8 -*-
## $Id: test_batching.py 14202 2016-09-29 06:51:42Z 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'
        customer['contracts'].addContract(contract)
        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)
        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,<IGNORE>,Frank,Meyer,m,1234,<IGNORE>,1990-01-06,F123456,cc@aa.ng,reg_number: Invalid input\r\n'
            '3,<IGNORE>,Uli,Schulz,m,1234,<IGNORE>,1990-01-07,A123456,dd@aa.ng,This object already exists. Skipping.\r\n'
            '4,<IGNORE>,Berta,Pieri,m,1234,<IGNORE>,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,<IGNORE>,c5,<IGNORE>,SampleContract,<IGNORE>,SAM,<IGNORE>,DOC1,<IGNORE>,This object already exists. Skipping.\r\n'
            '3,<IGNORE>,c6,<IGNORE>,SampleContract,<IGNORE>,SAM,<IGNORE>,NONEXISTENT,<IGNORE>,document_object: Invalid value\r\n'
            '3,<IGNORE>,c7,<IGNORE>,SampleContract,<IGNORE>,NONEXISTENT,<IGNORE>,DOC1,<IGNORE>,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: product_object=SAM, document_object=DOC1'
            % conid
            in logcontent)

    def test_import_update(self):
        # We perform the same import twice,
        # the second time in update mode. No record can be imported.
        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.
        # The other 4 records cannot be imported because DOC1 does
        # not belong to these customers.
        self.assertEqual(num_warns,7)
        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)