diff --git a/CHANGELOG.md b/CHANGELOG.md index 1ad69a3..4900967 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/src/cfdppy/filestore.py b/src/cfdppy/filestore.py index ea007b6..5257f4a 100644 --- a/src/cfdppy/filestore.py +++ b/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 diff --git a/tests/test_restricted_filestore.py b/tests/test_restricted_filestore.py index b546518..460a059 100644 --- a/tests/test_restricted_filestore.py +++ b/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))