Cover onboarding with atomic config writes
This commit is contained in:
parent
828b244031
commit
d5904283ab
2 changed files with 60 additions and 3 deletions
|
|
@ -28,6 +28,7 @@ from api.config import (
|
|||
save_settings,
|
||||
verify_hermes_imports,
|
||||
)
|
||||
from api.paths import _atomic_write_text
|
||||
from api.providers import _write_env_file # shared impl with _ENV_LOCK (#1164)
|
||||
from api.workspace import get_last_workspace, load_workspaces
|
||||
|
||||
|
|
@ -251,7 +252,8 @@ def _save_yaml_config(config_path: Path, config: dict) -> None:
|
|||
raise RuntimeError("PyYAML is required to write Hermes config.yaml") from exc
|
||||
|
||||
config_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
config_path.write_text(
|
||||
_atomic_write_text(
|
||||
config_path,
|
||||
_yaml.safe_dump(config, sort_keys=False, allow_unicode=True),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
|
|
|||
|
|
@ -98,6 +98,52 @@ def test_atomic_write_preserves_existing_group(tmp_path: Path) -> None:
|
|||
assert os.stat(target).st_gid == expected_gid
|
||||
|
||||
|
||||
@pytest.mark.skipif(not hasattr(os, "fchown"), reason="POSIX ownership semantics")
|
||||
def test_fchown_denial_falls_back_without_temp_debris(
|
||||
tmp_path: Path, monkeypatch
|
||||
) -> None:
|
||||
"""A writer unable to transfer ownership must keep the old inode contract."""
|
||||
target = tmp_path / "config.yaml"
|
||||
target.write_text("old: true\n", encoding="utf-8")
|
||||
|
||||
def _deny_fchown(*_args) -> None:
|
||||
raise PermissionError("simulated ownership denial")
|
||||
|
||||
monkeypatch.setattr(os, "fchown", _deny_fchown)
|
||||
|
||||
_atomic_write_text(target, "new: true\n")
|
||||
|
||||
assert target.read_text(encoding="utf-8") == "new: true\n"
|
||||
assert [p.name for p in tmp_path.iterdir()] == ["config.yaml"]
|
||||
|
||||
|
||||
def test_atomic_write_without_posix_fd_metadata_helpers(
|
||||
tmp_path: Path, monkeypatch
|
||||
) -> None:
|
||||
"""Windows-like platforms still use same-directory atomic replacement."""
|
||||
target = tmp_path / "config.yaml"
|
||||
target.write_text("old: true\n", encoding="utf-8")
|
||||
replace_calls: list[tuple[Path, Path]] = []
|
||||
real_replace = os.replace
|
||||
|
||||
monkeypatch.delattr(os, "fchown", raising=False)
|
||||
monkeypatch.delattr(os, "fchmod", raising=False)
|
||||
|
||||
def _recording_replace(src, dst) -> None:
|
||||
replace_calls.append((Path(src), Path(dst)))
|
||||
real_replace(src, dst)
|
||||
|
||||
monkeypatch.setattr(os, "replace", _recording_replace)
|
||||
|
||||
_atomic_write_text(target, "new: true\n")
|
||||
|
||||
assert target.read_text(encoding="utf-8") == "new: true\n"
|
||||
assert len(replace_calls) == 1
|
||||
assert replace_calls[0][0].parent == target.parent
|
||||
assert replace_calls[0][1] == target
|
||||
assert [p.name for p in tmp_path.iterdir()] == ["config.yaml"]
|
||||
|
||||
|
||||
def test_atomic_write_fsyncs_file_and_parent_directory(
|
||||
tmp_path: Path, monkeypatch
|
||||
) -> None:
|
||||
|
|
@ -393,7 +439,9 @@ def test_failed_write_through_symlink_leaves_link_and_target_intact(
|
|||
assert [p.name for p in target_dir.iterdir()] == ["config.yaml"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("writer", ["main", "profile_endpoint", "profile_defaults"])
|
||||
@pytest.mark.parametrize(
|
||||
"writer", ["main", "onboarding", "profile_endpoint", "profile_defaults"]
|
||||
)
|
||||
def test_each_config_writer_preserves_old_bytes_when_replace_fails(
|
||||
tmp_path: Path, monkeypatch, writer: str
|
||||
) -> None:
|
||||
|
|
@ -412,6 +460,12 @@ def test_each_config_writer_preserves_old_bytes_when_replace_fails(
|
|||
from api import config
|
||||
|
||||
config._save_yaml_config_file(target, {"model": {"default": "new"}})
|
||||
elif writer == "onboarding":
|
||||
from api import onboarding
|
||||
|
||||
onboarding._save_yaml_config(
|
||||
target, {"model": {"default": "replacement-model"}}
|
||||
)
|
||||
else:
|
||||
from api import profiles
|
||||
|
||||
|
|
@ -426,7 +480,8 @@ def test_each_config_writer_preserves_old_bytes_when_replace_fails(
|
|||
|
||||
|
||||
_ROOT_SKIP = pytest.mark.skipif(
|
||||
os.geteuid() == 0, reason="root bypasses directory permission bits"
|
||||
getattr(os, "geteuid", lambda: -1)() == 0,
|
||||
reason="root bypasses directory permission bits",
|
||||
)
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue