Changeset 12123 for main/waeup.ikoba

Timestamp:

3 Dec 2014, 10:25:50 (10 years ago)

Author:

Henrik Bettermann

Message:

 

Location:

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

Files:

• batching.py (modified) (1 diff)
• tests/test_batching.py (modified) (8 diffs)

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

@@ -555 +555 @@
                 errs.append(('contract_id','id exists'))
         else:
-            if not contract_id.startswith('a'):
+            if not contract_id.startswith('c'):
                 errs.append(('contract_id','invalid format'))
         return errs, inv_errs, conv_dict

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

@@ -36 +36 @@
 from waeup.ikoba.interfaces import IBatchProcessor, FatalCSVError, IUserAccount
 from waeup.ikoba.customers.batching import (
-    CustomerProcessor, CustomerDocumentProcessor)
+    CustomerProcessor, CustomerDocumentProcessor, ContractProcessor)
 from waeup.ikoba.customers.customer import Customer
 from waeup.ikoba.customers.documents import CustomerSampleDocument
@@ -68 +68 @@
 
 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(',')
 
@@ -83 +90 @@
         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)
+
+        # Add customer
+        customer = Customer()
+        customer.firstname = u'Anna'
+        customer.lastname = u'Tester'
+        customer.reg_number = u'123'
+        customer.date_of_birth = datetime.date(1981, 2, 4)
+        customer.email = 'anna@sample.com'
+        customer.phone = u'+234-123-12345'
+        customer.notice = u'Some notice\nin lines.'
+        customer.nationality = u'NG'
+        self.app['customers'].addCustomer(customer)
+        self.customer = customer
+        #notify(grok.ObjectModifiedEvent(customer))
+
+        # Add document
+        document = createObject(u'waeup.CustomerSampleDocument')
+        document.document_id = 'd120'
+        document.title = u'My Document'
+        customer['documents'].addDocument(document)
         return
 
@@ -128 +161 @@
     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')
@@ -255 +280 @@
         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.CustomerSampleDocument')
-        document.document_id = 'd120'
-        self.customer['documents'][document.document_id] = document
-
         # Import customers with subobjects
         customer_file = os.path.join(self.workdir, 'sample_customer_data.csv')
@@ -307 +321 @@
         self.assertEqual(len(self.customer['documents'].keys()),1)
         document1 = createObject(u'waeup.CustomerSampleDocument')
-        document1.document_id = 'p234'
+        document1.document_id = 'd234'
         self.processor.addEntry(
-            document1, dict(customer_id=self.customer.customer_id, document_id='p234'),
+            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']['p234'].document_id, 'p234')
+        self.assertEqual(self.customer['documents']['d234'].document_id, 'd234')
         document2 = createObject(u'waeup.CustomerSampleDocument')
         document1.document_id = 'nonsense'
@@ -318 +332 @@
     def test_checkConversion(self):
         errs, inv_errs, conv_dict = self.processor.checkConversion(
-            dict(document_id='d1266236341955', class_name='CustomerSampleDocument'))
+            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)
-        timestamp = ("%d" % int(time()*10000))[1:]
-        document_id = "d%s" % timestamp
-        errs, inv_errs, conv_dict = self.processor.checkConversion(
-            dict(document_id=document_id, class_name='CustomerSampleDocument'))
-        self.assertEqual(len(errs),0)
-        errs, inv_errs, conv_dict = self.processor.checkConversion(
-            dict(document_id=document_id, class_name='WrongDocument'))
+        errs, inv_errs, conv_dict = self.processor.checkConversion(
+            dict(document_id='d127', class_name='WrongDocument'))
         self.assertEqual(len(errs),1)
 
@@ -387 +396 @@
             'INFO - system - K1000001 - Document removed: d5'
             in logcontent)
+
+
+class ContractProcessorTest(CustomerImportExportSetup):
+
+    def setUp(self):
+        super(ContractProcessorTest, self).setUp()
+
+        # 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='c120'),
+            self.app) is self.customer['contracts']['c120']
+
+    def test_delEntry(self):
+        assert self.processor.getEntry(
+            dict(customer_id=self.customer.customer_id, contract_id='c120'),
+            self.app) is self.customer['contracts']['c120']
+        self.assertEqual(len(self.customer['contracts'].keys()),1)
+        self.processor.delEntry(
+            dict(customer_id=self.customer.customer_id, contract_id='c120'),
+            self.app)
+        assert self.processor.getEntry(
+            dict(customer_id=self.customer.customer_id, contract_id='c120'),
+            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'))
+        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 xxtest_import(self):
+        num, num_warns, fin_file, fail_file = self.processor.doImport(
+            self.csv_file, CONTRACT_HEADER_FIELDS,'create')
+        self.assertEqual(num_warns,1)
+        # contract_id must be unique
+        fail_file = open(fail_file).read()
+        self.assertEqual(fail_file,
+            'class_name,reg_number,contract_id,title,--ERRORS--\r\n'
+            'SampleContract,2,d5,My stolen doc,contract_id: id exists\r\n')
+        contract = self.processor.getEntry(dict(reg_number='1',
+            contract_id='d3'), self.app)
+        self.assertEqual(
+            self.app['customers']['X666666']['contracts']['d3'],
+            contract)
+        self.assertEqual(contract.contract_id, 'd3')
+        contract = self.processor.getEntry(dict(reg_number='3',
+            contract_id='d5'), 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=d3, title=My first doc'
+            in logcontent)
+
+    def xxtest_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 is one record without contract_id and one duplicate
+        self.assertEqual(num_warns,2)
+        shutil.rmtree(os.path.dirname(fin_file))
+
+    def xxtest_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 is one record without contract_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 - Contract removed: d5'
+            in logcontent)