Changeset 12054

Timestamp:

25 Nov 2014, 11:29:42 (10 years ago)

Author:

Henrik Bettermann

Message:

Use classname from import file to check if the right processor has been selected.

Location:

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

Files:

• batching.py (modified) (2 diffs)
• tests/sample_document_data.csv (modified) (1 diff)
• tests/test_batching.py (modified) (2 diffs)

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

@@ -379 +379 @@
 
     location_fields = []
-    additional_fields = ['document_id']
-    additional_headers = []
+    additional_fields = ['document_id', 'classname']
+    additional_headers = ['classname']
 
     def checkHeaders(self, headerfields, mode='ignore'):
@@ -437 +437 @@
         errs, inv_errs, conv_dict = super(
             CustomerDocumentProcessor, self).checkConversion(row, mode=mode)
-
+        # We need to check if the classname corresponds with the
+        # processor chosen. This is to avoid accidentally wrong imports.
+        if mode != 'remove':
+            classname = row.get('classname', None)
+            if classname != self.factory_name.strip('waeup.'):
+                errs.append(('classname','wrong processor'))
         # We have to check document_id.
         document_id = row.get('document_id', None)
-        if not document_id:
-            document_id = generate_document_id()
-            conv_dict['document_id'] = document_id
-            return errs, inv_errs, conv_dict
-        # document_id must not exist.
         if mode == 'create':
+            if not document_id:
+                document_id = generate_document_id()
+                conv_dict['document_id'] = document_id
+                return errs, inv_errs, conv_dict
             cat = queryUtility(ICatalog, name='documents_catalog')
             results = list(
                 cat.searchResults(document_id=(document_id, document_id)))
             if results:
+                # document_id must not exist.
                 errs.append(('document_id','id exists'))
-        if not document_id.startswith('d'):
-            errs.append(('document_id','invalid format'))
+        else:
+            if not document_id.startswith('d'):
+                errs.append(('document_id','invalid format'))
         return errs, inv_errs, conv_dict

main/waeup.ikoba/trunk/src/waeup/ikoba/customers/tests/sample_document_data.csv

@@ -1 @@
-document_id,reg_number,title
-d3,1,My first doc
-d4,2,My second doc
-d5,3,My third doc
-,1,My 4th doc
-d5,2,My stolen doc
+classname,document_id,reg_number,title
+CustomerDocument,d3,1,My first doc
+CustomerDocument,d4,2,My second doc
+CustomerDocument,d5,3,My third doc
+CustomerDocument,,1,My 4th doc
+CustomerDocument,d5,2,My stolen doc

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

@@ -318 +318 @@
     def test_checkConversion(self):
         errs, inv_errs, conv_dict = self.processor.checkConversion(
-            dict(document_id='d1266236341955'))
+            dict(document_id='d1266236341955', classname='CustomerDocument'))
         self.assertEqual(len(errs),0)
         errs, inv_errs, conv_dict = self.processor.checkConversion(
-            dict(document_id='nonsense'))
+            dict(document_id='nonsense', classname='CustomerDocument'))
         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))
+            dict(document_id=document_id, classname='CustomerDocument'))
         self.assertEqual(len(errs),0)
+        errs, inv_errs, conv_dict = self.processor.checkConversion(
+            dict(document_id=document_id, classname='WrongDocument'))
+        self.assertEqual(len(errs),1)
 
     def test_import(self):
@@ -336 +339 @@
         fail_file = open(fail_file).read()
         self.assertEqual(fail_file,
-            'reg_number,document_id,title,--ERRORS--\r\n'
-            '2,d5,My stolen doc,document_id: id exists\r\n')
+            'classname,reg_number,document_id,title,--ERRORS--\r\n'
+            'CustomerDocument,2,d5,My stolen doc,document_id: id exists\r\n')
         document = self.processor.getEntry(dict(reg_number='1',
             document_id='d3'), self.app)