diff --git a/src/specify_cli/extensions/__init__.py b/src/specify_cli/extensions/__init__.py index b84b836545..6caad2ae74 100644 --- a/src/specify_cli/extensions/__init__.py +++ b/src/specify_cli/extensions/__init__.py @@ -1397,12 +1397,68 @@ def install_from_directory( backup_config_dir.unlink() did_remove = self.remove(manifest.id) + # Rescue any config files left behind by a prior `remove --keep-config`. + # When an extension is removed with --keep-config, it is no longer in + # the registry but its config files remain in dest_dir. A subsequent + # plain (non-force) install would delete that directory unconditionally, + # silently discarding the preserved config. We read those files into + # memory now and write them back after copytree so the user's values + # always win over the packaged defaults. + stranded_configs: dict[str, tuple[bytes, int]] = {} + if dest_dir.exists() and not self.registry.is_installed(manifest.id): + for cfg_file in ( + list(dest_dir.glob("*-config.yml")) + + list(dest_dir.glob("*-config.local.yml")) + ): + if cfg_file.is_file() and not cfg_file.is_symlink(): + stranded_configs[cfg_file.name] = ( + cfg_file.read_bytes(), + cfg_file.stat().st_mode, + ) + # Install extension (dest_dir computed above during self-install guard) if dest_dir.exists(): shutil.rmtree(dest_dir) ignore_fn = self._load_extensionignore(source_dir) - shutil.copytree(source_dir, dest_dir, ignore=ignore_fn) + + def _restore_stranded_config_file( + target: Path, content: bytes, preserved_mode: int + ) -> None: + tmp_path: Path | None = None + try: + with tempfile.NamedTemporaryFile( + mode="wb", + dir=target.parent, + prefix=f".{target.name}.", + delete=False, + ) as tmp: + tmp_path = Path(tmp.name) + tmp.write(content) + tmp_path.chmod(preserved_mode) + os.replace(tmp_path, target) + except BaseException: + if tmp_path is not None and tmp_path.exists(): + tmp_path.unlink() + raise + + try: + shutil.copytree(source_dir, dest_dir, ignore=ignore_fn) + except BaseException: + # copytree failed — dest_dir may be absent or only partially + # created. Write the rescued configs back now so they are not + # permanently lost even though the install did not complete. + if stranded_configs: + dest_dir.mkdir(parents=True, exist_ok=True) + for filename, (content, mode) in stranded_configs.items(): + target = dest_dir / filename + _restore_stranded_config_file(target, content, mode) + raise + + # Restore stranded configs rescued before the rmtree above. + for filename, (content, mode) in stranded_configs.items(): + target = dest_dir / filename + _restore_stranded_config_file(target, content, mode) # Register commands with AI agents registered_commands = {} diff --git a/tests/test_extensions.py b/tests/test_extensions.py index 2885180360..093b0ffe77 100644 --- a/tests/test_extensions.py +++ b/tests/test_extensions.py @@ -20,6 +20,7 @@ from pathlib import Path from datetime import datetime, timezone from unittest.mock import MagicMock +import specify_cli.extensions as _ext_module from tests.conftest import strip_ansi from specify_cli.extensions import ( @@ -1226,6 +1227,133 @@ def test_install_force_config_preserved(self, extension_dir, project_dir): assert new_config.exists() assert new_config.read_text() == "test: config" + def test_reinstall_after_keep_config_preserves_config( + self, extension_dir, project_dir + ): + """Reinstalling after `remove --keep-config` must not overwrite preserved config.""" + manager = ExtensionManager(project_dir) + + # Add a packaged default config so the reinstall has a file to overwrite. + # Without the fix, the packaged default would silently win on reinstall. + packaged_config = extension_dir / "test-ext-config.yml" + packaged_config.write_text("model: default-model\nmax_iterations: 1\n") + + # Install once (packaged default is copied into the installed directory) + manager.install_from_directory( + extension_dir, "0.1.0", register_commands=False + ) + + # Overwrite the installed config with user-customized values + ext_dir = project_dir / ".specify" / "extensions" / "test-ext" + config_file = ext_dir / "test-ext-config.yml" + config_file.write_text("model: custom-model\nmax_iterations: 99\n") + + # Remove while preserving config + manager.remove("test-ext", keep_config=True) + assert not manager.registry.is_installed("test-ext") + assert config_file.exists() + assert "custom-model" in config_file.read_text() + + # Plain reinstall (no --force) — packaged default is still present in + # extension_dir, so a naive implementation would overwrite the custom values. + manager.install_from_directory( + extension_dir, "0.1.0", register_commands=False + ) + + # Preserved config must survive the reinstall and beat the packaged default + assert config_file.exists() + assert "custom-model" in config_file.read_text() + assert "99" in config_file.read_text() + assert "default-model" not in config_file.read_text() + + def test_reinstall_after_keep_config_preserves_local_config( + self, extension_dir, project_dir + ): + """Local config override files (*-config.local.yml) are also rescued on reinstall.""" + manager = ExtensionManager(project_dir) + + manager.install_from_directory( + extension_dir, "0.1.0", register_commands=False + ) + + ext_dir = project_dir / ".specify" / "extensions" / "test-ext" + local_cfg = ext_dir / "test-ext-config.local.yml" + local_cfg.write_text("local_override: true\n") + + manager.remove("test-ext", keep_config=True) + assert local_cfg.exists() + + manager.install_from_directory( + extension_dir, "0.1.0", register_commands=False + ) + + assert local_cfg.exists() + assert "local_override: true" in local_cfg.read_text() + + def test_copytree_failure_restores_stranded_config( + self, extension_dir, project_dir, monkeypatch + ): + """A copytree failure must not permanently lose a preserved config. + + When copytree raises after the existing directory has been removed, the + rollback path must write the rescued bytes back to dest_dir and restore + the original file mode, while leaving the extension unregistered. + """ + import stat + + manager = ExtensionManager(project_dir) + + # Add a packaged default config so copytree would overwrite it on success. + packaged_config = extension_dir / "test-ext-config.yml" + packaged_config.write_text("model: default-model\n") + + # Install once so the extension is on disk. + manager.install_from_directory( + extension_dir, "0.1.0", register_commands=False + ) + + ext_dir = project_dir / ".specify" / "extensions" / "test-ext" + config_file = ext_dir / "test-ext-config.yml" + config_file.write_text("model: custom-model\nmax_iterations: 99\n") + + # Set a known, non-default mode so we can assert it survives the rollback. + if platform.system() != "Windows": + config_file.chmod(0o640) + original_bytes = config_file.read_bytes() + original_mode = config_file.stat().st_mode + + # Remove while preserving the config — it is now a stranded file. + manager.remove("test-ext", keep_config=True) + assert not manager.registry.is_installed("test-ext") + assert config_file.exists() + + # Make copytree create a partial destination then raise so the rollback + # path is exercised. + + def failing_copytree(src, dst, **kwargs): + Path(dst).mkdir(parents=True, exist_ok=True) + (Path(dst) / "_partial.txt").write_text("partial") + raise OSError("simulated disk full") + + monkeypatch.setattr(_ext_module.shutil, "copytree", failing_copytree) + + with pytest.raises(OSError, match="simulated disk full"): + manager.install_from_directory( + extension_dir, "0.1.0", register_commands=False + ) + + # The preserved config must have been written back by the rollback path. + assert config_file.exists(), "rollback must recreate the config file" + assert config_file.read_bytes() == original_bytes + + # On POSIX, the original file mode must be faithfully restored. + if platform.system() != "Windows": + restored_mode = config_file.stat().st_mode + assert stat.S_IMODE(restored_mode) == stat.S_IMODE(original_mode) + + # The extension must remain unregistered after the failed install. + assert not manager.registry.is_installed("test-ext") + def test_install_force_without_existing(self, extension_dir, project_dir): """Test force-install when extension is NOT already installed (works normally).""" manager = ExtensionManager(project_dir)