Changeset 8634 for main/waeup.kofa/trunk/src/waeup/kofa/tests/test_datacenter.py

Timestamp:

6 Jun 2012, 01:40:08 (12 years ago)

Author:

uli

Message:

Let datacenter use the new CSV file merger from helpers.

CAUTION: this is a bit tested, but not in real use. Try
some local imports before using it in production environment!

File:

• main/waeup.kofa/trunk/src/waeup/kofa/tests/test_datacenter.py (modified) (1 diff)

main/waeup.kofa/trunk/src/waeup/kofa/tests/test_datacenter.py

@@ -213 +213 @@
         self.assertEqual(names, expected)
         return
+
+    def test_append_csv_file(self):
+        # we can append CSV files to others
+        datacenter = DataCenter()
+        csv_file1 = os.path.join(self.workdir, 'foo.csv')
+        csv_file2 = os.path.join(self.workdir, 'bar.csv')
+        open(csv_file1, 'wb').write('name,age\nBarney,28')
+        open(csv_file2, 'wb').write('name,age\nManfred,28')
+        datacenter._appendCSVFile(csv_file2, csv_file1)
+        result = open(csv_file1, 'rb').read()
+        self.assertEqual(result, 'age,name\r\n28,Barney\r\n28,Manfred\r\n')
+        # The source is deleted afterwards
+        self.assertEqual(os.path.exists(csv_file2), False)
+
+    def test_append_csv_file_no_dest(self):
+        # a non-existing dest CSV file will result in a simple move
+        datacenter = DataCenter()
+        csv_file1 = os.path.join(self.workdir, 'foo.csv')
+        csv_file2 = os.path.join(self.workdir, 'bar.csv')
+        # csv_file1 does not exist
+        open(csv_file2, 'wb').write('name,age\nManfred,28\n')
+        datacenter._appendCSVFile(csv_file2, csv_file1)
+        result = open(csv_file1, 'rb').read()
+        # raw input, no CSV mangling
+        self.assertEqual(result, 'name,age\nManfred,28\n')
+        # The source is deleted afterwards
+        self.assertEqual(os.path.exists(csv_file2), False)
+
+    def test_append_csv_file_no_source(self):
+        # a non existing source file will mean no changes at all
+        datacenter = DataCenter()
+        csv_file1 = os.path.join(self.workdir, 'foo.csv')
+        csv_file2 = os.path.join(self.workdir, 'bar.csv')
+        open(csv_file1, 'wb').write('name,age\nManfred,28\n')
+        # csv_file2 does not exist
+        datacenter._appendCSVFile(csv_file2, csv_file1)
+        result = open(csv_file1, 'rb').read()
+        # csv_file1 is the same as before
+        self.assertEqual(result, 'name,age\nManfred,28\n')
+
+    def test_append_csv_file_same_src_and_dest(self):
+        # if both csv files are the same file, nothing will be changed
+        datacenter = DataCenter()
+        csv_file1 = os.path.join(self.workdir, 'foo.csv')
+        csv_file2 = csv_file1
+        open(csv_file1, 'wb').write('name,age\nManfred,28\n')
+        # csv_file2 does not exist
+        datacenter._appendCSVFile(csv_file2, csv_file1)
+        result = open(csv_file1, 'rb').read()
+        # csv_file1 is the same as before
+        self.assertEqual(result, 'name,age\nManfred,28\n')
+        self.assertEqual(os.path.exists(csv_file2), True)