Changeset 12124 for main/waeup.ikoba/trunk/src/waeup/ikoba/customers

Timestamp:

3 Dec 2014, 10:49:52 (10 years ago)

Author:

Henrik Bettermann

Message:

Revert changes from last revision. This was an accidental commit.

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('c'):
+            if not contract_id.startswith('a'):
                 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, ContractProcessor)
+    CustomerProcessor, CustomerDocumentProcessor)
 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(',')
 
@@ -90 +83 @@
         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
 
@@ -161 +128 @@
     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')
@@ -280 +255 @@
         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')
@@ -321 +307 @@
         self.assertEqual(len(self.customer['documents'].keys()),1)
         document1 = createObject(u'waeup.CustomerSampleDocument')
-        document1.document_id = 'd234'
+        document1.document_id = 'p234'
         self.processor.addEntry(
-            document1, dict(customer_id=self.customer.customer_id, document_id='d234'),
+            document1, dict(customer_id=self.customer.customer_id, document_id='p234'),
             self.app)
         self.assertEqual(len(self.customer['documents'].keys()),2)
-        self.assertEqual(self.customer['documents']['d234'].document_id, 'd234')
+        self.assertEqual(self.customer['documents']['p234'].document_id, 'p234')
         document2 = createObject(u'waeup.CustomerSampleDocument')
         document1.document_id = 'nonsense'
@@ -332 +318 @@
     def test_checkConversion(self):
         errs, inv_errs, conv_dict = self.processor.checkConversion(
-            dict(document_id='d126', class_name='CustomerSampleDocument'))
+            dict(document_id='d1266236341955', 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'))
+        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'))
         self.assertEqual(len(errs),1)
 
@@ -396 +387 @@
             '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)