Merge pull request #35 from sgaisser/gaisser_filestore_create_folders

Create folders if they do not exist

Author: robamu

CHANGELOG.md

@@ -12,6 +12,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 - Added `RestrictedFilestore` to limit the file access of the `NativeFilestore` to a specific
   directory.
+- `Filestore` creates a directory if it does not exist when creating a new file.
 
 ## Fixed
 

src/cfdppy/filestore.py

@@ -238,6 +238,8 @@ def create_file(self, file: Path) -> FilestoreResponseStatusCode:
             _LOGGER.warning("File already exists")
             return FilestoreResponseStatusCode.CREATE_NOT_ALLOWED
         try:
+            # Creates subfolders if they do not exist
+            file.parent.mkdir(exist_ok=True, parents=True)
             with open(file, "x"):
                 pass
             return FilestoreResponseStatusCode.CREATE_SUCCESS

tests/test_restricted_filestore.py

@@ -43,6 +43,27 @@ def test_handle_files(self):
             self.assertEqual(result, FilestoreResponseStatusCode.DELETE_SUCCESS)
             self.assertFalse(Path(tempdir).joinpath("renamed_file.txt").exists())
 
+    def test_create_folder_with_file(self):
+        with tempfile.TemporaryDirectory() as tempdir:
+            filestore = RestrictedFilestore(restricted_path=Path(tempdir))
+            new_dir = Path(tempdir).joinpath("new_dir")
+            self.assertFalse(new_dir.exists())
+            result = filestore.create_file(new_dir.joinpath("a_file.txt"))
+            self.assertEqual(FilestoreResponseStatusCode.CREATE_SUCCESS, result)
+            self.assertTrue(new_dir.exists())
+            self.assertTrue(new_dir.joinpath("a_file.txt").exists())
+
+            # Create more than one folder
+            first_folder = Path(tempdir).joinpath("first_folder")
+            second_folder = first_folder.joinpath("second_folder")
+            self.assertFalse(first_folder.exists())
+            self.assertFalse(second_folder.exists())
+            result = filestore.create_file(second_folder.joinpath("a_file.txt"))
+            self.assertEqual(FilestoreResponseStatusCode.CREATE_SUCCESS, result)
+            self.assertTrue(first_folder.exists())
+            self.assertTrue(second_folder.exists())
+            self.assertTrue(second_folder.joinpath("a_file.txt").exists())
+
     def test_handle_directories(self):
         with tempfile.TemporaryDirectory() as tempdir:
             filestore = RestrictedFilestore(restricted_path=Path(tempdir))