Changeset 9266 for main/waeup.kofa/trunk/src/waeup/kofa/accesscodes/tests/test_batching.py

Timestamp:

2 Oct 2012, 06:43:52 (12 years ago)

Author:

Henrik Bettermann

Message:

Repair and optimize AccessCodeProcessor?. AC states and transitions can now be imported in create and update mode.

File:

• main/waeup.kofa/trunk/src/waeup/kofa/accesscodes/tests/test_batching.py (modified) (5 diffs)

main/waeup.kofa/trunk/src/waeup/kofa/accesscodes/tests/test_batching.py

@@ -54 +54 @@
     '\n')[0].split(',')
 
+AC_SAMPLE_DATA_UPDATE = open(
+    os.path.join(os.path.dirname(__file__), 'sample_ac_data_update.csv'),
+    'rb').read()
+
+AC_HEADER_FIELDS_UPDATE = AC_SAMPLE_DATA_UPDATE.split(
+    '\n')[0].split(',')
+
 class ACFullSetup(FunctionalTestCase):
 
@@ -104 +111 @@
         self.assertTrue(
             'system - AccessCodeBatch Processor - CLR-1 - '
-            'Batch updated: num=1, creator=system, entry_num=3, '
-            'creation_date=2012-04-28 07:28:48.719026+00:00, '
+            'Batch updated: num=1, creator=system, used_num=1, '
+            'entry_num=3, creation_date=2012-04-28 07:28:48.719026+00:00, '
             'prefix=CLR, cost=0.0, disabled_num=0'
             in logcontent)
@@ -115 +122 @@
     layer = FunctionalLayer
 
-    def test_import(self):
+    def test_import_create(self):
         self.processor = AccessCodeBatchProcessor()
         self.csv_file = os.path.join(self.workdir, 'sample_batch_data.csv')
@@ -126 +133 @@
         num, num_warns, fin_file, fail_file =  self.processor.doImport(
             self.csv_file, AC_HEADER_FIELDS)
-        self.assertEqual(num_warns,0)
+        self.assertEqual(num_warns,3)
+        fail_file = open(fail_file).read()
+        self.assertTrue(
+            'HOS-1-0007044547,HOS,anything,state: not allowed' in fail_file)
+        self.assertTrue(
+            'HOS-1-555,HOS,anything,workflow: not allowed' in fail_file)
+        self.assertTrue(
+            'HOS-1-666,HOS,anything,transition: not allowed' in fail_file)
+        self.assertEqual(
+            self.app['accesscodes']['HOS-1']['HOS-1-98769876'].state, 'used')
+        self.assertEqual(
+            self.app['accesscodes']['HOS-1']['HOS-1-76254765'].state, 'initialized')
         self.assertEqual(
             self.app['accesscodes']['CLR-1']['CLR-1-1625368961'].batch_serial, 33)
@@ -133 +151 @@
         self.assertEqual(
             self.app['accesscodes']['CLR-1']['CLR-1-1625368961'].state, 'used')
+        self.assertMatches(
+            self.app['accesscodes']['CLR-1']['CLR-1-1625368961'].history,
+            "<YYYY-MM-DD hh:mm:ss> UTC - initialized by system|"
+            "|<YYYY-MM-DD hh:mm:ss> UTC - state 'used' set by system")
+        self.assertMatches(
+            self.app['accesscodes']['HOS-1']['HOS-1-98769876'].history,
+            "<YYYY-MM-DD hh:mm:ss> UTC - initialized by system|"
+            "|<YYYY-MM-DD hh:mm:ss> UTC - used by system")
         logcontent = open(self.logfile).read()
         self.assertTrue(
             'INFO - system - AccessCode Processor - CLR-1-1625368961 - '
-            'AC updated: state=used, history=2012-07-11 14:51:56 UTC - '
-            'initialized by Manager||2012-07-11 14:55:42 UTC - '
-            'invalidated by Manager, batch_serial=33, '
-            'random_num=1625368961, cost=100.0, owner=K1000009'
+            'AC updated: state=used, batch_serial=33, random_num=1625368961, '
+            'cost=100.0, owner=K1000009'
             in logcontent)
         shutil.rmtree(os.path.dirname(fin_file))
         return
+
+    def test_import_update(self):
+        self.processor = AccessCodeBatchProcessor()
+        self.csv_file = os.path.join(self.workdir, 'sample_batch_data.csv')
+        open(self.csv_file, 'wb').write(BATCH_SAMPLE_DATA)
+        self.processor.doImport(self.csv_file, BATCH_HEADER_FIELDS)
+
+        self.processor = AccessCodeProcessor()
+        self.csv_file = os.path.join(self.workdir, 'sample_ac_data.csv')
+        open(self.csv_file, 'wb').write(AC_SAMPLE_DATA)
+        num, num_warns, fin_file, fail_file =  self.processor.doImport(
+            self.csv_file, AC_HEADER_FIELDS)
+
+        self.csv_file = os.path.join(self.workdir, 'sample_ac_data_update.csv')
+        open(self.csv_file, 'wb').write(AC_SAMPLE_DATA_UPDATE)
+        num, num_warns, fin_file, fail_file =  self.processor.doImport(
+            self.csv_file, AC_HEADER_FIELDS_UPDATE, 'update')
+        fin_file = open(fin_file).read()
+        self.assertEqual(num_warns,0)
+        self.assertEqual(
+            self.app['accesscodes']['HOS-1']['HOS-1-98769876'].state, 'disabled')
+        self.assertEqual(
+            self.app['accesscodes']['CLR-1']['CLR-1-1625368961'].state, 'disabled')
+        self.assertMatches(
+            self.app['accesscodes']['CLR-1']['CLR-1-1625368961'].history,
+            "<YYYY-MM-DD hh:mm:ss> UTC - initialized by system|"
+            "|<YYYY-MM-DD hh:mm:ss> UTC - state 'used' set by system|"
+            "|<YYYY-MM-DD hh:mm:ss> UTC - state 'disabled' set by system")
+        self.assertMatches(
+            self.app['accesscodes']['HOS-1']['HOS-1-98769876'].history,
+            "<YYYY-MM-DD hh:mm:ss> UTC - initialized by system|"
+            "|<YYYY-MM-DD hh:mm:ss> UTC - used by system|"
+            "|<YYYY-MM-DD hh:mm:ss> UTC - disabled by system")
+        return