Changeset 11995 for main

Timestamp:

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

Author:

Henrik Bettermann

Message:

Add batch processor for customer documents.

Location:

main/waeup.ikoba/trunk/src/waeup/ikoba/customers

Files:

• batching.py (modified) (4 diffs)
• tests/sample_document_data.csv (added)
• tests/test_batching.py (modified) (4 diffs)

main/waeup.ikoba/trunk/src/waeup/ikoba/customers/batching.py

@@ -41 +41 @@
 from waeup.ikoba.interfaces import MessageFactory as _
 from waeup.ikoba.customers.interfaces import (
-    ICustomer, ICustomerUpdateByRegNo)
+    ICustomer, ICustomerUpdateByRegNo,
+    ICustomerDocument)
 from waeup.ikoba.customers.workflow import  (
     IMPORTABLE_STATES, IMPORTABLE_TRANSITIONS)
@@ -278 +279 @@
 
 class CustomerProcessorBase(BatchProcessor):
-    """A base for customer subitem processor.
+    """A base for customer subitem processors.
 
     Helps reducing redundancy.
@@ -354 +355 @@
         raw_header = reader.next()
         for num, field in enumerate(headerfields):
-            if field not in ['customer_id', 'reg_number', 'p_id', 'code', 'level'
+            if field not in ['customer_id', 'reg_number', 'id', 'code', 'level'
                              ] and mode == 'remove':
                 continue
@@ -362 +363 @@
             result[raw_header[num]] = field
         return result
+
+class CustomerDocumentProcessor(CustomerProcessorBase):
+    """A batch processor for ICustomerDocument objects.
+    """
+    grok.implements(IBatchProcessor)
+    grok.provides(IBatchProcessor)
+    grok.context(Interface)
+    util_name = 'customerdocumentprocessor'
+    grok.name(util_name)
+
+    name = _('CustomerDocument Processor')
+    iface = ICustomerDocument
+    factory_name = 'waeup.CustomerDocument'
+
+    location_fields = []
+    additional_fields = ['id']
+    additional_headers = []
+
+    def checkHeaders(self, headerfields, mode='ignore'):
+        super(CustomerDocumentProcessor, self).checkHeaders(headerfields)
+        if mode in ('update', 'remove') and not 'id' in headerfields:
+            raise FatalCSVError(
+                "Need id for import in update and remove modes!")
+        return True
+
+    def getParent(self, row, site):
+        customer = self._getCustomer(row, site)
+        if customer is None:
+            return None
+        return customer['documents']
+
+    def getEntry(self, row, site):
+        documents = self.getParent(row, site)
+        if documents is None:
+            return None
+        id = row.get('id', None)
+        if id is None:
+            return None
+        entry = documents.get(id)
+        return entry
+
+    def updateEntry(self, obj, row, site, filename):
+        """Update obj to the values given in row.
+        """
+        items_changed = super(CustomerDocumentProcessor, self).updateEntry(
+            obj, row, site, filename)
+        customer = self.getParent(row, site).__parent__
+        customer.__parent__.logger.info(
+            '%s - %s - %s - updated: %s'
+            % (self.name, filename, customer.customer_id, items_changed))
+        return
+
+    def addEntry(self, obj, row, site):
+        parent = self.getParent(row, site)
+        id = row['id'].strip('#')
+        parent[id] = obj
+        return
+
+    def delEntry(self, row, site):
+        document = self.getEntry(row, site)
+        parent = self.getParent(row, site)
+        if document is not None:
+            customer = self._getCustomer(row, site)
+            customer.__parent__.logger.info('%s - Document removed: %s'
+                % (customer.customer_id, document.id))
+            del parent[document.id]
+        return
+
+    def checkConversion(self, row, mode='ignore'):
+        """Validates all values in row.
+        """
+        errs, inv_errs, conv_dict = super(
+            CustomerDocumentProcessor, self).checkConversion(row, mode=mode)
+
+        # We have to check id.
+        id = row.get('id', None)
+        if not id:
+            timestamp = ("%d" % int(time()*10000))[1:]
+            id = "d%s" % timestamp
+            conv_dict['id'] = id
+            return errs, inv_errs, conv_dict
+        if not id.startswith('d') or len(id) != 14:
+            errs.append(('id','invalid format'))
+        return errs, inv_errs, conv_dict

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)