Changeset 12291 for main

Timestamp:

21 Dec 2014, 22:43:47 (10 years ago)

Author:

Henrik Bettermann

Message:

Add two more export-import tests.

File:

• main/ikobacustom.skeleton/trunk/src/ikobacustom/skeleton/customers/tests/test_browser.py (modified) (6 diffs)

main/ikobacustom.skeleton/trunk/src/ikobacustom/skeleton/customers/tests/test_browser.py

@@ -26 +26 @@
 from zope.event import notify
 from zope.interface.verify import verifyObject, verifyClass
+from zope.testbrowser.testing import Browser
+from hurry.workflow.interfaces import IWorkflowState
 from waeup.ikoba.customers.tests.test_batching import CustomerImportExportSetup
 from ikobacustom.skeleton.customers.documents import SkeletonCustomerDocument
-from ikobacustom.skeleton.customers.export import SkeletonCustomerDocumentExporter
-from ikobacustom.skeleton.customers.batching import SkeletonCustomerDocumentProcessor
+from ikobacustom.skeleton.customers.export import (
+    SkeletonCustomerExporter,
+    SkeletonCustomerDocumentExporter,
+    SkeletonContractExporter)
+from ikobacustom.skeleton.customers.batching import (
+    SkeletonCustomerProcessor,
+    SkeletonCustomerDocumentProcessor,
+    SkeletonContractProcessor)
 from ikobacustom.skeleton.testing import FunctionalLayer
 
@@ -39 +47 @@
     def setup_for_export(self):
         customer = createObject(u'waeup.Customer')
+        customer.firstname = u'Beate'
+        customer.lastname = u'Mueller'
+        customer.reg_number = u'123'
+        customer.sex = u'f'
+        IWorkflowState(customer).setState('started')
         self.app['customers'].addCustomer(customer)
         document = createObject(u'waeup.SkeletonCustomerDocument')
         document.title = u'My first document'
         customer['documents'].addDocument(document)
+        contract = createObject(u'waeup.SkeletonContract')
+        contract .title = u'My first contract'
+        customer['contracts'].addContract(contract)
         self.customer = customer
         self.document = document
+        self.contract = contract
         self.outfile = os.path.join(self.workdir, 'myoutput.csv')
         return
 
-    def test_export_reimport_all(self):
+    def test_export_reimport_customers(self):
+        # we can export all customers in a portal
+        # set values we can expect in export file
+        self.setup_for_export()
+        exporter = SkeletonCustomerExporter()
+        exporter.export_all(self.app, self.outfile)
+        result = open(self.outfile, 'rb').read()
+        self.assertEqual(result,
+            'customer_id,email,firstname,lastname,middlename,phone,'
+            'reg_number,sex,suspended,suspended_comment,password,state,history\r\n'
+            'K1000000,,Beate,Mueller,,,123,f,0,,,started,[]\r\n')
+        # We can reimport the file ...
+        processor = SkeletonCustomerProcessor()
+        result = processor.doImport(
+            self.outfile,
+            ['customer_id','email','firstname','lastname','middlename','phone',
+            'reg_number','sex','suspended','suspended_comment','password','state'],
+            mode='create')
+        num, num_fail, finished_path, failed_path = result
+        self.assertEqual(num_fail,1)
+        # ... if we remove the original customer.
+        del self.app['customers']['K1000000']
+        result = processor.doImport(
+            self.outfile,
+            ['customer_id','email','firstname','lastname','middlename','phone',
+            'reg_number','sex','suspended','suspended_comment','password','state'],
+            mode='create')
+        num_succ, num_fail, finished_path, failed_path = result
+        self.assertEqual(num_fail,0)
+        # We can import the same file in update mode if we ignore the reg_number.
+        result = processor.doImport(
+            self.outfile,
+            ['customer_id','email','firstname','lastname','middlename','phone',
+            'xx_reg_number','sex','suspended','suspended_comment','password','state'],
+            mode='update')
+        num_succ, num_fail, finished_path, failed_path = result
+        self.assertEqual(num_succ,1)
+        self.assertEqual(num_fail,0)
+        return
+
+    def test_export_reimport_documents(self):
         # we can export all documents in a portal
         # set values we can expect in export file
@@ -61 +118 @@
             'created,My first document,K1000000\r\n'
             % self.document.document_id)
-        # We can import the file if we change the header (user_id -> customer_id)
+        # We can reimport the file if we change the header (user_id -> customer_id)
         processor = SkeletonCustomerDocumentProcessor()
         open(self.outfile, 'wb').write(
@@ -72 +129 @@
             mode='create')
         num, num_fail, finished_path, failed_path = result
-        #content = open(failed_path).read()
         # The object exists.
         self.assertEqual(num_fail,1)
@@ -84 +140 @@
         self.assertEqual(num_fail,0)
         # We can import the same file in update mode.
-        processor = SkeletonCustomerDocumentProcessor()
         result = processor.doImport(
             self.outfile,
@@ -93 +148 @@
         self.assertEqual(num_fail,0)
         return
+
+    def test_export_reimport_contracts(self):
+        # we can export all contracts in a portal
+        # set values we can expect in export file
+        self.setup_for_export()
+        exporter = SkeletonContractExporter()
+        exporter.export_all(self.app, self.outfile)
+        result = open(self.outfile, 'rb').read()
+        self.assertMatches(result,
+            'class_name,contract_category,contract_id,document_object,'
+            'history,last_product_id,product_object,state,title,user_id\r\n'
+            'SkeletonContract,sample,%s,,'
+            '[u\'2014-12-21 22:26:00 WAT - Contract created by system\']'
+            ',,,created,My first contract,K1000000\r\n'
+            % self.contract.contract_id)
+        # We can reimport the file if we change the header (user_id -> customer_id)
+        processor = SkeletonContractProcessor()
+        open(self.outfile, 'wb').write(
+            'class_name,contract_category,contract_id,document_object,'
+            'history,last_product_id,product_object,state,title,customer_id\r\n'
+            'SkeletonContract,sample,%s,,'
+            '[u\'2014-12-21 22:26:00 WAT - Contract created by system\']'
+            ',,,created,My first contract,K1000000\r\n'
+            % self.contract.contract_id)
+        result = processor.doImport(
+            self.outfile,
+            ['class_name','contract_category','contract_id','document_object',
+            'history','last_product_id','product_object','state','title','customer_id'],
+            mode='create')
+        num, num_fail, finished_path, failed_path = result
+        # The object exists.
+        self.assertEqual(num_fail,1)
+        # We remove the original contract.
+        del self.customer['contracts'][self.contract.contract_id]
+        result = processor.doImport(
+            self.outfile,
+            ['class_name','contract_category','contract_id','document_object',
+            'history','last_product_id','product_object','state','title','customer_id'],
+            mode='create')
+        num_succ, num_fail, finished_path, failed_path = result
+        self.assertEqual(num_fail,0)
+        # We can import the same file in update mode.
+        result = processor.doImport(
+            self.outfile,
+            ['class_name','contract_category','contract_id','document_object',
+            'history','last_product_id','product_object','state','title','customer_id'],
+            mode='update')
+        num_succ, num_fail, finished_path, failed_path = result
+        self.assertEqual(num_succ,1)
+        self.assertEqual(num_fail,0)
+        return