File size: 6,696 Bytes
c61f7ed
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
"""``atomic_write_text``'s opt-in metadata preservation (mode + owner).

``os.replace`` swaps mkstemp's 0600 temp file (owned by the writing user)
onto the target, so a bare atomic rewrite of an existing user-authored file
tightens its permission bits and — for root-run callers on Docker/NAS
volumes — flips its ownership.  ``preserve_mode=True`` carries both across
the replace, exactly like ``atomic_yaml_write`` does unconditionally;
``create_mode=`` sets the bits when the target does not exist yet.

These guard the follow-up to PR #79323, which collapsed three hand-rolled
stat/write/chmod blocks (xai migration, uninstaller shell-rc rewrite,
dashboard SOUL.md editor) into these kwargs.
"""

from __future__ import annotations

import os
import stat
import sys
from pathlib import Path

import pytest

from utils import atomic_write_text, atomic_yaml_write


pytestmark = pytest.mark.skipif(
    sys.platform == "win32", reason="POSIX permission bits"
)


class TestPreserveMode:
    def test_existing_mode_survives_the_rewrite(self, tmp_path: Path) -> None:
        """A 0640 managed config must not tighten to mkstemp's 0600."""
        target = tmp_path / "config.yaml"
        target.write_text("old: true\n", encoding="utf-8")
        os.chmod(target, 0o640)

        atomic_write_text(target, "new: true\n", preserve_mode=True)

        assert target.read_text(encoding="utf-8") == "new: true\n"
        assert stat.S_IMODE(target.stat().st_mode) == 0o640

    def test_default_still_leaves_mkstemp_mode(self, tmp_path: Path) -> None:
        """Without opt-in, behavior is unchanged: the file lands 0600."""
        target = tmp_path / "notes.md"
        target.write_text("old\n", encoding="utf-8")
        os.chmod(target, 0o644)

        atomic_write_text(target, "new\n")

        assert stat.S_IMODE(target.stat().st_mode) == 0o600

    def test_mode_is_applied_before_the_replace(
        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        """The temp fd gets fchmod'd, so the target never transits 0600."""
        target = tmp_path / "config.yaml"
        target.write_text("old\n", encoding="utf-8")
        os.chmod(target, 0o640)

        import utils as utils_mod

        real_replace = utils_mod.atomic_replace
        seen: list[int] = []

        def spying_replace(tmp, dst):
            seen.append(stat.S_IMODE(os.stat(tmp).st_mode))
            return real_replace(tmp, dst)

        monkeypatch.setattr(utils_mod, "atomic_replace", spying_replace)
        atomic_write_text(target, "new\n", preserve_mode=True)

        assert seen == [0o640]

    def test_owner_is_restored_on_the_real_symlink_target(
        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        """Root-run rewrites of a user-owned file must not flip ownership.

        Mirrors test_atomic_yaml_write_restores_owner_on_real_symlink_target:
        forces a preserved uid/gid so the test does not need root.
        """
        real = tmp_path / "zshrc"
        link = tmp_path / ".zshrc"
        real.write_text("export A=1\n", encoding="utf-8")
        link.symlink_to(real)

        chown_calls: list[tuple[Path, int, int]] = []
        monkeypatch.setattr("utils._preserve_file_owner", lambda _p: (123, 456))
        monkeypatch.setattr(
            "utils.os.chown",
            lambda path, uid, gid: chown_calls.append((Path(path), uid, gid)),
        )

        atomic_write_text(link, "export B=2\n", preserve_mode=True)

        assert chown_calls == [(real, 123, 456)]
        assert link.is_symlink()
        assert real.read_text(encoding="utf-8") == "export B=2\n"

    def test_no_owner_calls_without_opt_in(
        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        target = tmp_path / "mem.md"
        target.write_text("old\n", encoding="utf-8")

        chown_calls: list[tuple] = []
        monkeypatch.setattr("utils._preserve_file_owner", lambda _p: (123, 456))
        monkeypatch.setattr(
            "utils.os.chown", lambda *a: chown_calls.append(a)
        )

        atomic_write_text(target, "new\n")

        assert chown_calls == []


class TestCreateMode:
    def test_create_mode_applies_when_target_is_new(self, tmp_path: Path) -> None:
        target = tmp_path / "SOUL.md"
        assert not target.exists()

        atomic_write_text(
            target, "# Persona\n", preserve_mode=True, create_mode=0o644
        )

        assert stat.S_IMODE(target.stat().st_mode) == 0o644

    def test_existing_mode_beats_create_mode(self, tmp_path: Path) -> None:
        target = tmp_path / "SOUL.md"
        target.write_text("old\n", encoding="utf-8")
        os.chmod(target, 0o600)

        atomic_write_text(
            target, "new\n", preserve_mode=True, create_mode=0o644
        )

        assert stat.S_IMODE(target.stat().st_mode) == 0o600

    def test_create_mode_never_rewrites_an_existing_file(
        self, tmp_path: Path
    ) -> None:
        """create_mode without preserve_mode must not chmod an existing file."""
        target = tmp_path / "notes.md"
        target.write_text("old\n", encoding="utf-8")
        os.chmod(target, 0o640)

        atomic_write_text(target, "new\n", create_mode=0o644)

        # The write is a plain (non-preserving) atomic rewrite: mkstemp 0600.
        assert stat.S_IMODE(target.stat().st_mode) == 0o600

    def test_windows_fallback_branch_applies_mode_after_replace(
        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        """Without os.fchmod (Windows), the mode is applied post-replace."""
        target = tmp_path / "config.yaml"
        target.write_text("old\n", encoding="utf-8")
        os.chmod(target, 0o640)

        monkeypatch.delattr(os, "fchmod")
        atomic_write_text(target, "new\n", preserve_mode=True)

        assert target.read_text(encoding="utf-8") == "new\n"
        assert stat.S_IMODE(target.stat().st_mode) == 0o640

    def test_atomic_yaml_write_create_mode(self, tmp_path: Path) -> None:
        """write_manifest's create path: new file lands 0644, not 0600."""
        target = tmp_path / "distribution.yaml"
        assert not target.exists()

        atomic_yaml_write(target, {"name": "t"}, create_mode=0o644)

        assert stat.S_IMODE(target.stat().st_mode) == 0o644

    def test_atomic_yaml_write_existing_mode_beats_create_mode(
        self, tmp_path: Path
    ) -> None:
        target = tmp_path / "distribution.yaml"
        target.write_text("name: old\n", encoding="utf-8")
        os.chmod(target, 0o600)

        atomic_yaml_write(target, {"name": "new"}, create_mode=0o644)

        assert stat.S_IMODE(target.stat().st_mode) == 0o600