--- a/src/osforge/transition.py +++ b/src/osforge/transition.py @@ -47,12 +47,15 @@ os.close(fd) -def atomic_text(path: Path, text: str, *, exclusive: bool = False, mode: int = 0o644) -> None: +def atomic_text(path: Path, text: str, *, exclusive: bool = False, mode: int = 0o644, + owner: tuple[int, int] | None = None) -> None: path.parent.mkdir(parents=True, exist_ok=True) fd, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) temp = Path(temporary) try: with os.fdopen(fd, "w", encoding="utf-8") as handle: + if owner is not None: + os.fchown(handle.fileno(), *owner) os.fchmod(handle.fileno(), mode) handle.write(text) handle.flush() --- a/src/osforge/repository_activation.py +++ b/src/osforge/repository_activation.py @@ -139,7 +139,9 @@ if os.readlink(alias) != str(home / "current"): switch_pointer(alias, home / "current") if text != rendered: - atomic_text(config, rendered, mode=stat.S_IMODE(config.stat().st_mode)) + info = config.stat() + atomic_text(config, rendered, mode=stat.S_IMODE(info.st_mode), + owner=(info.st_uid, info.st_gid)) return {"portage_config": str(config), "local_link": str(alias), "release": str(release)} --- a/tests/test_repository_activation.py +++ b/tests/test_repository_activation.py @@ -66,6 +66,14 @@ self.assertEqual(self.alias.resolve(), self.location()) self.publish() # Idempotent publication also checks the activation. + @unittest.skipUnless(os.geteuid() == 0, 'changing file owner requires root') + def test_registration_preserves_existing_owner_and_group(self): + os.chown(self.config, 65534, 65534) + candidate = self.update() + self.publish(candidate) + info = self.config.stat() + self.assertEqual((info.st_uid, info.st_gid), (65534, 65534)) + def test_interrupted_activation_reconciles_without_duplicate_commit(self): candidate = self.update() with patch('osforge.repository_activation.atomic_text', side_effect=OSError('disk failure')): --- a/tests/test_merge.py +++ b/tests/test_merge.py @@ -147,7 +147,7 @@ def test_merge_plan_writes_ledger_for_valid_configs(self) -> None: with TemporaryDirectory() as tmp: root = Path(tmp) - config = OsforgeConfig(path=root / "osforge.toml", state_root=root / "state") + config = OsforgeConfig(path=root / "osforge.toml", state_root=root / "state", merge_output_repo=root / "output") result = build_merge_plan( config,