Changeset 11995 for main/waeup.ikoba/trunk/src/waeup/ikoba/customers/tests/test_batching.py

Timestamp:

19 Nov 2014, 16:05:30 (10 years ago)

Author:

Henrik Bettermann

Message:

Add batch processor for customer documents.

File:

• main/waeup.ikoba/trunk/src/waeup/ikoba/customers/tests/test_batching.py (modified) (4 diffs)

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

@@ -2 +2 @@
 ## $Id: test_batching.py 11756 2014-07-09 12:46:08Z henrik $
 ##
-## Copyright (C) 2011 Uli Fouquet & Henrik Bettermann
+## 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
@@ -35 +35 @@
 from waeup.ikoba.app import Company
 from waeup.ikoba.interfaces import IBatchProcessor, FatalCSVError, IUserAccount
-from waeup.ikoba.customers.batching import CustomerProcessor
+from waeup.ikoba.customers.batching import (
+    CustomerProcessor, CustomerDocumentProcessor)
 from waeup.ikoba.customers.customer import Customer
 from waeup.ikoba.testing import FunctionalLayer, FunctionalTestCase
@@ -59 +60 @@
 
 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(',')
 
@@ -238 +246 @@
         shutil.rmtree(os.path.dirname(fin_file))
 
+class CustomerDocumentProcessorTest(CustomerImportExportSetup):
+
+    def setUp(self):
+        super(CustomerDocumentProcessorTest, self).setUp()
+
+        # Add customer with document
+        customer = Customer()
+        customer.firstname = u'Anna'
+        customer.lastname = u'Tester'
+        customer.reg_number = u'123'
+        self.app['customers'].addCustomer(customer)
+        self.customer = self.app['customers'][customer.customer_id]
+        document = createObject(u'waeup.CustomerDocument')
+        document.id = 'd120'
+        self.customer['documents'][document.id] = document
+
+        # 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', id='nonsense'), self.app) is None
+        assert self.processor.getEntry(
+            dict(customer_id=self.customer.customer_id, id='d120'),
+            self.app) is self.customer['documents']['d120']
+
+    def test_delEntry(self):
+        assert self.processor.getEntry(
+            dict(customer_id=self.customer.customer_id, id='d120'),
+            self.app) is self.customer['documents']['d120']
+        self.assertEqual(len(self.customer['documents'].keys()),1)
+        self.processor.delEntry(
+            dict(customer_id=self.customer.customer_id, id='d120'),
+            self.app)
+        assert self.processor.getEntry(
+            dict(customer_id=self.customer.customer_id, id='d120'),
+            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.CustomerDocument')
+        document1.id = 'p234'
+        self.processor.addEntry(
+            document1, dict(customer_id=self.customer.customer_id, id='p234'),
+            self.app)
+        self.assertEqual(len(self.customer['documents'].keys()),2)
+        self.assertEqual(self.customer['documents']['p234'].id, 'p234')
+        document2 = createObject(u'waeup.CustomerDocument')
+        document1.id = 'nonsense'
+
+    def test_checkConversion(self):
+        errs, inv_errs, conv_dict = self.processor.checkConversion(
+            dict(id='d1266236341955'))
+        self.assertEqual(len(errs),0)
+        errs, inv_errs, conv_dict = self.processor.checkConversion(
+            dict(id='nonsense'))
+        self.assertEqual(len(errs),1)
+        timestamp = ("%d" % int(time()*10000))[1:]
+        id = "d%s" % timestamp
+        errs, inv_errs, conv_dict = self.processor.checkConversion(
+            dict(id=id))
+        self.assertEqual(len(errs),0)
+
+    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,0)
+        document = self.processor.getEntry(dict(reg_number='1',
+            id='d1266236341953'), self.app)
+        self.assertEqual(
+            self.app['customers']['X666666']['documents']['d1266236341953'],
+            document)
+        self.assertEqual(document.id, 'd1266236341953')
+        document = self.processor.getEntry(dict(reg_number='3',
+            id='d1266236341955'), 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: '
+            'id=d1266236341953, 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')
+        self.assertEqual(num_warns,1)  # There is one record without id
+        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')
+        self.assertEqual(num_warns,1)  # There is one record without id
+        shutil.rmtree(os.path.dirname(fin_file))
+        logcontent = open(self.logfile).read()
+        self.assertTrue(
+            'INFO - system - K1000001 - Document removed: d1266236341955'
+            in logcontent)