1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
|
--- 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,
|
