Convert phase: Adds saving converted input as a csv

digital_land/phase/convert.py

@@ -83,8 +83,9 @@ def read_excel(path):
     return f
 
 
-def convert_features_to_csv(input_path):
-    output_path = tempfile.NamedTemporaryFile(suffix=".csv").name
+def convert_features_to_csv(input_path, output_path=None):
+    if not output_path:
+        output_path = tempfile.NamedTemporaryFile(suffix=".csv").name
     execute(
         [
             "ogr2ogr",
@@ -114,7 +115,13 @@ def convert_features_to_csv(input_path):
 
 
 class ConvertPhase(Phase):
-    def __init__(self, path=None, dataset_resource_log=None, custom_temp_dir=None):
+    def __init__(
+        self,
+        path=None,
+        dataset_resource_log=None,
+        custom_temp_dir=None,
+        output_path=None,
+    ):
         self.path = path
         self.log = dataset_resource_log
         self.charset = ""
@@ -125,6 +132,12 @@ def __init__(self, path=None, dataset_resource_log=None, custom_temp_dir=None):
         else:
             self.temp_file_extra_kwargs = {}
 
+        self.output_path = output_path
+        if output_path:
+            output_dir = os.path.dirname(output_path)
+            if not os.path.exists(output_dir):
+                os.makedirs(output_dir)
+
     def process(self, stream=None):
         input_path = self.path
 
@@ -161,7 +174,7 @@ def _read_text_file(self, input_path, encoding):
         elif content.lower().startswith(("<?xml ", "<wfs:")):
             logging.debug("%s looks like xml", input_path)
             self.log.mime_type = "application/xml" + self.charset
-            converted_csv_file = convert_features_to_csv(input_path)
+            converted_csv_file = convert_features_to_csv(input_path, self.output_path)
             if not converted_csv_file:
                 f.close()
                 logging.warning("conversion from XML to CSV failed")
@@ -170,7 +183,7 @@ def _read_text_file(self, input_path, encoding):
         elif content.lower().startswith("{"):
             logging.debug("%s looks like json", input_path)
             self.log.mime_type = "application/json" + self.charset
-            converted_csv_file = convert_features_to_csv(input_path)
+            converted_csv_file = convert_features_to_csv(input_path, self.output_path)
 
         if converted_csv_file:
             f.close()
@@ -246,7 +259,7 @@ def _read_binary_file(self, input_path):
                 os.link(input_path, temp_path)
                 zip_path = f"/vsizip/{temp_path}{internal_path}"
                 logging.debug(f"zip_path: {zip_path} mime_type: {mime_type}")
-                csv_path = convert_features_to_csv(zip_path)
+                csv_path = convert_features_to_csv(zip_path, self.output_path)
                 encoding = detect_file_encoding(csv_path)
                 return read_csv(csv_path, encoding)
 
@@ -260,7 +273,7 @@ def _read_binary_file(self, input_path):
         else:
             logging.debug(f"{input_path} looks like SQLite")
             self.log.mime_type = "application/geopackage+sqlite3"
-            csv_path = convert_features_to_csv(input_path)
+            csv_path = convert_features_to_csv(input_path, self.output_path)
             encoding = detect_file_encoding(csv_path)
             return read_csv(csv_path, encoding)
 

tests/unit/phase/test_convert.py

@@ -0,0 +1,58 @@
+#!/usr/bin/env -S py.test -svv
+import os
+import pytest
+
+from digital_land.log import DatasetResourceLog
+from digital_land.phase.convert import ConvertPhase
+
+
+class TestConvertPhase:
+    def test_process_xlsm_is_loaded(self):
+        path = os.path.join(os.getcwd(), "tests/data/brentwood.xlsm")
+        log = DatasetResourceLog()
+        reader = ConvertPhase(path, dataset_resource_log=log).process()
+        block = next(reader)
+        assert block["resource"] == "brentwood"
+        assert block["line-number"] == 1
+        assert "OrganisationURI" in block["line"]
+        assert log.mime_type == "application/vnd.ms-excel"
+
+    @pytest.mark.parametrize(
+        "input_path",
+        [
+            "tests/data/resource_examples/geojson.resource",
+            "tests/data/resource_examples/geopackage.resource",
+            "tests/data/resource_examples/gml.resource",
+            "tests/data/resource_examples/kml.resource",
+            "tests/expectations/resources_to_test_expectations/data_for_url_expect_test.sqlite3",
+        ],
+    )
+    def test_process_file_is_saved_to_output_path(self, input_path, tmp_path):
+        path = os.path.join(os.getcwd(), input_path)
+        output_path = os.path.join(tmp_path, "converted/geojson.csv")
+        log = DatasetResourceLog()
+        ConvertPhase(path, dataset_resource_log=log, output_path=output_path).process()
+
+        assert os.path.isfile(output_path)
+        # os.remove(os.path.join(os.getcwd(), "converted/geojson.csv"))
+
+    def test_process_converted_file_not_saved(self):
+        files_before = []
+        for root, dirs, files in os.walk(os.getcwd()):
+            files_before.extend(files)
+
+        path = os.path.join(
+            os.getcwd(),
+            "tests/expectations/resources_to_test_expectations/data_for_url_expect_test.sqlite3",
+        )
+        log = DatasetResourceLog()
+        ConvertPhase(
+            path,
+            dataset_resource_log=log,
+        ).process()
+
+        files_after = []
+        for root, dirs, files in os.walk(os.getcwd()):
+            files_after.extend(files)
+
+        assert files_before == files_after