diff --git a/src/openlifu/db/database.py b/src/openlifu/db/database.py index f82dfa48..a747c1a5 100644 --- a/src/openlifu/db/database.py +++ b/src/openlifu/db/database.py @@ -414,11 +414,17 @@ def write_volume(self, subject_id, volume_id, volume_name, volume_data_filepath, try: volume_ids = self.get_volume_ids(subject_id) + old_volume_data_filepath = None if volume_id in volume_ids: if on_conflict == OnConflictOpts.ERROR: raise ValueError(f"Volume with ID {volume_id} already exists for subject {subject_id}.") elif on_conflict == OnConflictOpts.OVERWRITE: self.logger.info(f"Overwriting volume with ID {volume_id} for subject {subject_id}.") + volume_metadata_filepath = self.get_volume_metadata_filepath(subject_id, volume_id) + if volume_metadata_filepath.exists(): + with open(volume_metadata_filepath) as file: + old_volume_metadata = json.load(file) + old_volume_data_filepath = volume_metadata_filepath.parent / old_volume_metadata["data_filename"] elif on_conflict == OnConflictOpts.SKIP: self.logger.info(f"Skipping volume with ID {volume_id} for subject {subject_id} as it already exists.") return @@ -435,6 +441,14 @@ def write_volume(self, subject_id, volume_id, volume_name, volume_data_filepath, file.write(volume_metadata_json) shutil.copy(Path(volume_data_filepath), Path(volume_metadata_filepath).parent) + new_volume_data_filepath = Path(volume_metadata_filepath).parent / Path(volume_data_filepath).name + if ( + old_volume_data_filepath is not None + and old_volume_data_filepath != new_volume_data_filepath + and old_volume_data_filepath.exists() + ): + old_volume_data_filepath.unlink() + if volume_id not in volume_ids: volume_ids.append(volume_id) self.write_volume_ids(subject_id, volume_ids) diff --git a/tests/test_database.py b/tests/test_database.py index 7530abe7..b070b51b 100644 --- a/tests/test_database.py +++ b/tests/test_database.py @@ -581,6 +581,34 @@ def test_write_volume(example_database:Database, tmp_path:Path): assert(volume_filepath.name == "example_volume_2.json") assert((volume_filepath.parent/"example_volume_2.nii").exists()) +def test_issue_136(example_database:Database, tmp_path:Path): + subject_id = "example_subject" + volume_id = "a_volume_id" + volume_name = "a_volume_name" + volume_data_path = Path(tmp_path/'test_db_files/a_volume_filename.nii') + volume_data_path.parent.mkdir(parents=True, exist_ok=True) + volume_data_path.touch() + volume_data_path2 = Path(tmp_path/'test_db_files/a_different_volume_filename.nii') + volume_data_path2.parent.mkdir(parents=True, exist_ok=True) + volume_data_path2.touch() + + # write volume with the data path volume_data_path + example_database.write_volume(subject_id, volume_id, volume_name, volume_data_path) + + # check that the file is there in the DB with the expected name + volume_filepath = example_database.get_volume_metadata_filepath(subject_id, volume_id) + assert((volume_filepath.parent/volume_data_path.name).exists()) + + # now overwrite using a different volume data path volume_data_path2, with a different filename + example_database.write_volume(subject_id, volume_id, volume_name, volume_data_path2, on_conflict=OnConflictOpts.OVERWRITE) + + # check that the file is there in the DB with the expected name + volume_filepath = example_database.get_volume_metadata_filepath(subject_id, volume_id) + assert((volume_filepath.parent/volume_data_path2.name).exists()) + + # check that the old file was removed + assert(not (volume_filepath.parent/volume_data_path.name).exists()) + def test_load_solution(example_database:Database, example_session:Session): with pytest.raises(FileNotFoundError,match="Solution file not found"): example_database.load_solution(example_session, "bogus_solution_id")