diff --git a/src/ucode/agents/__init__.py b/src/ucode/agents/__init__.py index 0785342..17d8d30 100644 --- a/src/ucode/agents/__init__.py +++ b/src/ucode/agents/__init__.py @@ -25,6 +25,7 @@ map_claude_family_models, resolve_provider_service, ) +from ucode.managed_files import restore_managed_file from ucode.state import get_provider_service, load_state, save_state from ucode.telemetry import agent_version from ucode.ui import ( @@ -89,6 +90,32 @@ } +def _os_managed_file_path(tool: str): + if tool == "claude": + return claude._managed_settings_path() + if tool == "codex": + return codex._managed_config_path() + return None + + +def reconcile_global_settings(managed: dict | None) -> None: + """Undo ucode's OS-managed-file write for each global-settings agent the current config no longer + marks global, so switching workspaces stops a prior one's settings applying to a bare agent. + + Reconciles both agents, not just the launched one; an agent still marked global is left for its + writer to (re)write. Idempotent, drift-suppressed, and nonfatal. ``managed_use_as_global_settings`` + is imported locally to avoid a cycle (managed_resolve imports ``GLOBAL_SETTINGS_AGENTS`` from here). + """ + from ucode.managed_resolve import managed_use_as_global_settings + + for tool in GLOBAL_SETTINGS_AGENTS: + if managed_use_as_global_settings(managed or {}, tool): + continue + path = _os_managed_file_path(tool) + if path is not None: + restore_managed_file(path, display=TOOL_SPECS[tool]["display"]) + + def install_databricks_ai_tools_for_agents(tools: list[str], state: dict) -> None: """Install Databricks AI Tools for the coding agents that support them (gemini/pi have no ``aitools`` support and are dropped).""" diff --git a/src/ucode/cli.py b/src/ucode/cli.py index ca71dd6..307fb27 100644 --- a/src/ucode/cli.py +++ b/src/ucode/cli.py @@ -22,6 +22,7 @@ install_tool_binary, normalize_tool, provider_permission_error, + reconcile_global_settings, resolve_launch_model, resolve_provider_models, validate_all_tools, @@ -282,6 +283,7 @@ def _resolve_workspace_then_maybe_reject( managed, coding_agent_config_feature_disabled = refresh_managed_config( {"workspace": workspace, "profile": profile} ) + reconcile_global_settings(managed) if not managed and not coding_agent_config_feature_disabled: _maybe_offer_admin_setup(workspace, profile) if not managed: @@ -1910,6 +1912,8 @@ def _launch_tool( f"which overrides `--model {model}` — Claude Code will launch on the pinned " "model instead. Edit or remove that file to use --model." ) + if managed_agent_config_enabled(): + reconcile_global_settings(managed) state = configure_tool( tool, state, @@ -2091,6 +2095,7 @@ def _launch_managed_default( else: with spinner("Loading..."): managed, coding_agent_config_feature_disabled = refresh_managed_config(state) + reconcile_global_settings(managed) if not managed and not coding_agent_config_feature_disabled: _print_no_managed_config_guidance(current, state.get("profile")) if not managed: diff --git a/src/ucode/managed_files.py b/src/ucode/managed_files.py index 40e5880..54ae965 100644 --- a/src/ucode/managed_files.py +++ b/src/ucode/managed_files.py @@ -17,14 +17,17 @@ from __future__ import annotations +import json import os import shlex import subprocess import sys import tempfile +from collections.abc import Callable from enum import Enum from pathlib import Path +from ucode import config_io from ucode.config_io import is_dry_run from ucode.ui import console, print_err, print_warning @@ -89,6 +92,7 @@ def write_managed_file(path: Path, desired_text: str, *, display: str) -> str: if is_dry_run(): console.print(f"\n[bold]\\[dry run] {path} (via sudo)[/bold]\n{desired_text}") return "written" + snapshot = _snapshot_original(path) try: _sudo_replace(path, desired_text) except PermissionError as exc: @@ -100,6 +104,7 @@ def write_managed_file(path: Path, desired_text: str, *, display: str) -> str: except subprocess.CalledProcessError as exc: _report_sudo_failure(path, display, exc) return "skipped" + _record_original(path, snapshot) return "written" @@ -186,3 +191,156 @@ def _report_sudo_failure(path: Path, display: str, exc: subprocess.CalledProcess ) else: print_err(f"{display}: failed to write managed settings at {path}: {stderr or exc}") + + +def _ledger_path() -> Path: + """Machine-global record of each OS-managed file's pre-ucode state (OS files are machine-wide, + unlike per-workspace ``state.json``), so reconcile can restore the original instead of pruning keys.""" + return config_io.APP_DIR / "managed-os-backups.json" + + +def _read_ledger() -> dict: + path = _ledger_path() + try: + if not path.exists(): + return {} + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return {} + return data if isinstance(data, dict) else {} + + +def _write_ledger(ledger: dict) -> None: + try: + config_io.APP_DIR.mkdir(parents=True, exist_ok=True) + _ledger_path().write_text(json.dumps(ledger, indent=2), encoding="utf-8") + except OSError: + pass + + +def _snapshot_original(path: Path) -> dict: + """Read ``path``'s pre-ucode state into a ledger entry. + + ``contents`` is the exact text for a readable file, ``None`` for one absent or unreadable (kept + distinct from a genuinely empty ``""``). When the unprivileged stat or read fails (a root-locked + parent, or a root-only file under a readable dir) it falls back to a privileged probe via + :func:`_sudo_snapshot`, because guessing here would either leave a ucode-created file uncleaned or + delete a pre-existing one. Runs only on the write path, so the fallback adds no extra sudo prompt. + """ + try: + existed = path.exists() + except OSError: + return _sudo_snapshot(path) + if not existed: + return {"existed": False, "contents": None} + try: + return {"existed": True, "contents": path.read_text(encoding="utf-8")} + except OSError: + return _sudo_snapshot(path) + + +def _sudo_snapshot(path: Path) -> dict: + """Determine ``path``'s pre-ucode state with sudo, for a root-locked parent or root-only file. + + A file sudo confirms absent is recorded ``existed=False`` so cleanup removes the file ucode is + about to create; one readable only as root is captured verbatim; one unreadable even as root stays + ``existed=True/contents=None`` (left untouched on cleanup). A sudo failure is harmless: the write + that follows fails the same way, so this snapshot is never recorded. + """ + present = subprocess.run([_SUDO, "test", "-e", str(path)], capture_output=True, check=False) + if present.returncode != 0: + return {"existed": False, "contents": None} + read = subprocess.run([_SUDO, "cat", str(path)], capture_output=True, text=True, check=False) + if read.returncode == 0: + return {"existed": True, "contents": read.stdout} + return {"existed": True, "contents": None} + + +def _record_original(path: Path, snapshot: dict) -> None: + """Persist ``snapshot`` to the ledger under ``path``, once (call only after a successful write, so + a failed sudo leaves no false ownership record). No-op if already recorded.""" + key = str(path) + ledger = _read_ledger() + if key in ledger: + return + ledger[key] = snapshot + _write_ledger(ledger) + + +def _forget_original(key: str) -> None: + if is_dry_run(): + return + ledger = _read_ledger() + if key in ledger: + del ledger[key] + _write_ledger(ledger) + + +def restore_managed_file(path: Path, *, display: str) -> str: + """Undo ucode's write to ``path``, restoring the pre-ucode original or removing a ucode-created + file. Returns ``"restored"``, ``"removed"``, ``"unchanged"``, or ``"skipped"``. + + No ledger entry means ucode never wrote it, so nothing happens. Drift-suppressed (an already-matching + or already-absent file makes no sudo call) and never raises: sudo failures warn and return ``"skipped"``. + """ + if not managed_files_supported(): + return "unchanged" + key = str(path) + entry = _read_ledger().get(key) + if not isinstance(entry, dict): + return "unchanged" + original = entry.get("contents") + if entry.get("existed"): + if not isinstance(original, str): + _forget_original(key) + return "unchanged" + if _read_existing(path) == original: + _forget_original(key) + return "unchanged" + if is_dry_run(): + console.print(f"\n[bold]\\[dry run] restore {path} (via sudo)[/bold]\n{original}") + return "restored" + if not _sudo_managed_op(lambda: _sudo_replace(path, original), path, display): + return "skipped" + _forget_original(key) + return "restored" + if not _path_present(path): + _forget_original(key) + return "unchanged" + if is_dry_run(): + console.print(f"\n[bold]\\[dry run] remove {path} (via sudo)[/bold]") + return "removed" + if not _sudo_managed_op(lambda: _sudo_remove(path), path, display): + return "skipped" + _forget_original(key) + return "removed" + + +def _path_present(path: Path) -> bool: + """Whether ``path`` exists; a root-locked parent we can't stat is assumed present (let sudo try).""" + try: + return path.exists() + except OSError: + return True + + +def _sudo_managed_op(op: Callable[[], None], path: Path, display: str) -> bool: + """Run a privileged managed-file op, turning its failures into a warning. Returns success.""" + try: + op() + except PermissionError as exc: + print_err( + f"{display}: cannot update {path} without root ({exc}). Re-run with `sudo ucode ...` to " + "reconcile the machine-wide config." + ) + return False + except subprocess.CalledProcessError as exc: + _report_sudo_failure(path, display, exc) + return False + return True + + +def _sudo_remove(path: Path) -> None: + """Remove a root-owned managed file via sudo, clearing any immutable flag first.""" + _clear_immutable(path) + subprocess.run([_SUDO, "rm", "-f", str(path)], capture_output=True, text=True, check=True) diff --git a/tests/test_agent_claude.py b/tests/test_agent_claude.py index 6b49782..80a08df 100644 --- a/tests/test_agent_claude.py +++ b/tests/test_agent_claude.py @@ -565,6 +565,56 @@ def test_relayed_skips_managed_write(self, monkeypatch): assert any("bare `claude`" in w for w in warns) +class TestClaudeManagedSettingsLifecycle: + """Workspace A (global settings) -> B (none): the real capture+restore path, sudo mocked.""" + + def _patch(self, tmp_path, monkeypatch): + import ucode.managed_files as mf + + managed_path = tmp_path / "etc-claude-code" / "managed-settings.json" + monkeypatch.setattr(claude, "_managed_settings_path", lambda: managed_path) + monkeypatch.setattr(claude, "write_json_file", lambda path, payload: None) + monkeypatch.setattr(claude, "save_state", lambda state: None) + monkeypatch.setattr(claude, "backup_existing_file", lambda *a, **kw: True) + monkeypatch.setattr(claude, "_register_web_search_mcp", lambda *a, **kw: True) + monkeypatch.setattr(mf, "managed_files_supported", lambda: True) + + def _fake_replace(path, text): + p = Path(path) + p.parent.mkdir(parents=True, exist_ok=True) + p.write_text(text, encoding="utf-8") + + monkeypatch.setattr(mf, "_sudo_replace", _fake_replace) + monkeypatch.setattr(mf, "_sudo_remove", lambda path: Path(path).unlink()) + return managed_path + + def test_switch_away_restores_it_authored_file(self, tmp_path, monkeypatch): + import ucode.managed_files as mf + + managed_path = self._patch(tmp_path, monkeypatch) + managed_path.parent.mkdir(parents=True, exist_ok=True) + managed_path.write_text('{"env": {"IT_KEY": "keep"}}', encoding="utf-8") + + state = {"workspace": WS, "codex_models": [], "write_managed_config": True} + claude.write_tool_config(state, "databricks-claude-sonnet-4") + written = json.loads(managed_path.read_text(encoding="utf-8")) + assert written["env"]["IT_KEY"] == "keep" # IT key preserved through the merge + assert written["env"]["ANTHROPIC_BASE_URL"] # ucode keys applied + + assert mf.restore_managed_file(managed_path, display="Claude Code") == "restored" + assert json.loads(managed_path.read_text(encoding="utf-8")) == {"env": {"IT_KEY": "keep"}} + + def test_switch_away_removes_ucode_created_file(self, tmp_path, monkeypatch): + import ucode.managed_files as mf + + managed_path = self._patch(tmp_path, monkeypatch) # no file on disk before ucode + state = {"workspace": WS, "codex_models": [], "write_managed_config": True} + claude.write_tool_config(state, "databricks-claude-sonnet-4") + assert managed_path.exists() + assert mf.restore_managed_file(managed_path, display="Claude Code") == "removed" + assert not managed_path.exists() + + class TestRegisterWebSearchMcp: def test_clears_existing_then_adds(self, monkeypatch): import ucode.mcp as mcp_mod diff --git a/tests/test_agent_codex.py b/tests/test_agent_codex.py index 54fe1aa..91879a0 100644 --- a/tests/test_agent_codex.py +++ b/tests/test_agent_codex.py @@ -718,3 +718,54 @@ def test_no_managed_write_by_default(self, tmp_path, monkeypatch): state = {"workspace": WS, "codex_models": ["gpt-5"]} codex.write_tool_config(state) assert not managed_path.exists() + + +class TestCodexManagedConfigLifecycle: + """Workspace A (global settings) -> B (none): the real capture+restore path, sudo mocked.""" + + def _patch(self, tmp_path, monkeypatch): + import ucode.managed_files as mf + + config_path = tmp_path / ".codex" / "ucode.config.toml" + managed_path = tmp_path / "etc-codex" / "managed_config.toml" + monkeypatch.setattr(codex, "CODEX_CONFIG_PATH", config_path) + monkeypatch.setattr(codex, "CODEX_BACKUP_PATH", tmp_path / "codex-ucode-config.backup.toml") + monkeypatch.setattr(codex, "agent_version", lambda binary: "0.134.0") + monkeypatch.setattr(codex, "save_state", lambda state: None) + monkeypatch.setattr(codex, "_managed_config_path", lambda: managed_path) + monkeypatch.setattr(mf, "managed_files_supported", lambda: True) + + def _fake_replace(path, text): + p = Path(path) + p.parent.mkdir(parents=True, exist_ok=True) + p.write_text(text, encoding="utf-8") + + monkeypatch.setattr(mf, "_sudo_replace", _fake_replace) + monkeypatch.setattr(mf, "_sudo_remove", lambda path: Path(path).unlink()) + return managed_path + + def test_switch_away_restores_it_authored_file(self, tmp_path, monkeypatch): + import ucode.managed_files as mf + + managed_path = self._patch(tmp_path, monkeypatch) + managed_path.parent.mkdir(parents=True, exist_ok=True) + managed_path.write_text('approval_policy = "on-request"\n', encoding="utf-8") + + state = {"workspace": WS, "codex_models": ["gpt-5"], "write_managed_config": True} + codex.write_tool_config(state) + doc = read_toml_safe(managed_path) + assert doc["approval_policy"] == "on-request" # IT key preserved through the merge + assert doc["model"] == "gpt-5" # ucode keys applied + + assert mf.restore_managed_file(managed_path, display="Codex") == "restored" + assert managed_path.read_text(encoding="utf-8") == 'approval_policy = "on-request"\n' + + def test_switch_away_removes_ucode_created_file(self, tmp_path, monkeypatch): + import ucode.managed_files as mf + + managed_path = self._patch(tmp_path, monkeypatch) # no file on disk before ucode + state = {"workspace": WS, "codex_models": ["gpt-5"], "write_managed_config": True} + codex.write_tool_config(state) + assert managed_path.exists() + assert mf.restore_managed_file(managed_path, display="Codex") == "removed" + assert not managed_path.exists() diff --git a/tests/test_agents_init.py b/tests/test_agents_init.py index f346af9..5f0606e 100644 --- a/tests/test_agents_init.py +++ b/tests/test_agents_init.py @@ -3,6 +3,7 @@ from __future__ import annotations import subprocess +from pathlib import Path import pytest @@ -705,3 +706,48 @@ def fail_run(cmd, **kwargs): assert ok is True assert err == "" + + +class TestReconcileGlobalSettings: + """`reconcile_global_settings` undoes the OS-managed write for any global-settings agent the + current config no longer marks global — for BOTH agents, whichever one is being launched.""" + + def _record(self, monkeypatch): + calls: list[str] = [] + monkeypatch.setattr( + agents_mod, "restore_managed_file", lambda path, *, display: calls.append(str(path)) + ) + monkeypatch.setattr(agents_mod.claude, "_managed_settings_path", lambda: Path("/x/claude")) + monkeypatch.setattr(agents_mod.codex, "_managed_config_path", lambda: Path("/x/codex")) + return calls + + def test_launching_claude_reconciles_stale_codex(self, monkeypatch): + calls = self._record(monkeypatch) + agents_mod.reconcile_global_settings( + {"enabled_agents": {"claude": {"use_as_global_settings": True}}} + ) + assert calls == ["/x/codex"] + + def test_launching_codex_reconciles_stale_claude(self, monkeypatch): + calls = self._record(monkeypatch) + agents_mod.reconcile_global_settings( + {"enabled_agents": {"codex": {"use_as_global_settings": True}}} + ) + assert calls == ["/x/claude"] + + def test_no_managed_config_reconciles_both(self, monkeypatch): + calls = self._record(monkeypatch) + agents_mod.reconcile_global_settings(None) + assert sorted(calls) == ["/x/claude", "/x/codex"] + + def test_both_global_reconciles_neither(self, monkeypatch): + calls = self._record(monkeypatch) + agents_mod.reconcile_global_settings( + { + "enabled_agents": { + "claude": {"use_as_global_settings": True}, + "codex": {"use_as_global_settings": True}, + } + } + ) + assert calls == [] diff --git a/tests/test_cli.py b/tests/test_cli.py index 99b99dc..b5c98fb 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -2558,6 +2558,96 @@ def test_a_removed_model_list_no_longer_skips_discovery(self, monkeypatch): assert mock_shared.call_args.kwargs["skip_model_discovery"] is False +class TestReconcileWiring: + """Both the launch and the configure entry points reconcile the OS managed files, so switching + workspaces tears down a prior workspace's global settings.""" + + def test_launch_reconciles_with_the_resolved_config(self, monkeypatch): + seen: list = [] + monkeypatch.setenv("ENABLE_MANAGED_AGENT_CONFIG", "1") + managed = {"enabled_agents": {"claude": {}}} + monkeypatch.setattr("ucode.cli.refresh_managed_config", lambda state: (managed, False)) + monkeypatch.setattr("ucode.cli.reconcile_global_settings", lambda m: seen.append(m)) + state = dict(MINIMAL_STATE) + with ( + patch("ucode.cli.normalize_tool", return_value="claude"), + patch("ucode.cli.load_state", return_value=state), + patch("ucode.cli.apply_pat_environment"), + patch("ucode.cli.ensure_bootstrap_dependencies"), + patch("ucode.cli.ensure_provider_state", return_value=state), + patch("ucode.cli.configure_shared_state", return_value=state), + patch("ucode.cli.configure_tool", return_value=state), + patch("ucode.cli.launch_agent"), + ): + result = runner.invoke(app, ["claude"]) + assert result.exit_code == 0, result.output + assert seen == [managed] + + def test_launch_does_not_reconcile_when_feature_disabled(self, monkeypatch): + seen: list = [] + monkeypatch.delenv("ENABLE_MANAGED_AGENT_CONFIG", raising=False) + monkeypatch.setattr("ucode.cli.reconcile_global_settings", lambda m: seen.append(m)) + state = dict(MINIMAL_STATE) + with ( + patch("ucode.cli.normalize_tool", return_value="claude"), + patch("ucode.cli.load_state", return_value=state), + patch("ucode.cli.apply_pat_environment"), + patch("ucode.cli.ensure_bootstrap_dependencies"), + patch("ucode.cli.ensure_provider_state", return_value=state), + patch("ucode.cli.configure_shared_state", return_value=state), + patch("ucode.cli.configure_tool", return_value=state), + patch("ucode.cli.launch_agent"), + ): + result = runner.invoke(app, ["claude"]) + assert result.exit_code == 0, result.output + assert seen == [] + + def test_configure_reconciles_when_a_managed_config_exists(self, monkeypatch): + import typer + + seen: list = [] + monkeypatch.setenv("ENABLE_MANAGED_AGENT_CONFIG", "1") + monkeypatch.setattr("ucode.cli.set_current_workspace", lambda ws: None) + monkeypatch.setattr("ucode.cli.load_state", lambda: {"workspace": "https://w"}) + managed = {"enabled_agents": {"claude": {}}} + monkeypatch.setattr("ucode.cli.refresh_managed_config", lambda state: (managed, False)) + monkeypatch.setattr("ucode.cli.reconcile_global_settings", lambda m: seen.append(m)) + import ucode.cli as cli_mod + + with pytest.raises(typer.Exit): + cli_mod._resolve_workspace_then_maybe_reject([("https://w", None)]) + assert seen == [managed] + + def test_configure_reconciles_when_no_managed_config(self, monkeypatch): + seen: list = [] + monkeypatch.setenv("ENABLE_MANAGED_AGENT_CONFIG", "1") + monkeypatch.setattr("ucode.cli.set_current_workspace", lambda ws: None) + monkeypatch.setattr("ucode.cli.load_state", lambda: {"workspace": "https://w"}) + monkeypatch.setattr("ucode.cli.refresh_managed_config", lambda state: (None, False)) + monkeypatch.setattr("ucode.cli.get_databricks_token", lambda ws, profile=None: "tok") + monkeypatch.setattr("ucode.cli.is_workspace_admin", lambda ws, tok: False) + monkeypatch.setattr("ucode.cli.reconcile_global_settings", lambda m: seen.append(m)) + import ucode.cli as cli_mod + + entries = cli_mod._resolve_workspace_then_maybe_reject([("https://w", None)]) + assert entries == [("https://w", None)] + assert seen == [None] + + def test_bare_ucode_reconciles_before_the_no_config_early_return(self, monkeypatch): + seen: list = [] + monkeypatch.setenv("ENABLE_MANAGED_AGENT_CONFIG", "1") + monkeypatch.setattr("ucode.cli.install_databricks_cli", lambda *a, **k: None) + monkeypatch.setattr("ucode.cli.apply_pat_environment", lambda *a, **k: None) + monkeypatch.setattr("ucode.cli.load_state", lambda: {"workspace": "https://w"}) + monkeypatch.setattr("ucode.cli.refresh_managed_config", lambda state: (None, False)) + monkeypatch.setattr("ucode.cli.get_databricks_token", lambda *a, **k: "tok") + monkeypatch.setattr("ucode.cli.is_workspace_admin", lambda *a, **k: False) + monkeypatch.setattr("ucode.cli.reconcile_global_settings", lambda m: seen.append(m)) + result = runner.invoke(app, []) + assert result.exit_code == 0, result.output + assert seen == [None] + + class TestConfigureDeprecation: """`ucode configure` resolves the target workspace first, then short-circuits once a managed config exists for it, since the admin's wins anyway.""" diff --git a/tests/test_managed_files.py b/tests/test_managed_files.py index dbb14f7..8ea0fdf 100644 --- a/tests/test_managed_files.py +++ b/tests/test_managed_files.py @@ -104,3 +104,258 @@ def exists(self): managed_files.subprocess, "run", lambda *a, **k: pytest.fail("should not shell out") ) assert managed_files._clear_immutable(_StatDenied()) is False + + +def _fake_sudo(monkeypatch): + """Mock the privileged primitives to act on the tmp file directly (no real sudo/`/etc`). + + Records each op so a test can assert whether a privileged call happened at all. + """ + calls: list = [] + + def _replace(path, text): + calls.append(("write", str(path), text)) + p = managed_files.Path(path) + p.parent.mkdir(parents=True, exist_ok=True) + p.write_text(text, encoding="utf-8") + + def _remove(path): + calls.append(("remove", str(path))) + p = managed_files.Path(path) + if p.exists(): + p.unlink() + + monkeypatch.setattr(managed_files, "_sudo_replace", _replace) + monkeypatch.setattr(managed_files, "_sudo_remove", _remove) + return calls + + +class TestCaptureOnWrite: + """`write_managed_file` records the pre-ucode file, once, only on a real change.""" + + def test_absent_file_captured_as_not_existing(self, tmp_path, monkeypatch): + path = tmp_path / "managed.json" + _fake_sudo(monkeypatch) + managed_files.write_managed_file(path, "new", display="X") + assert managed_files._read_ledger()[str(path)] == {"existed": False, "contents": None} + + def test_existing_file_captures_original_contents(self, tmp_path, monkeypatch): + path = tmp_path / "managed.json" + path.write_text("orig", encoding="utf-8") + _fake_sudo(monkeypatch) + managed_files.write_managed_file(path, "new", display="X") + assert managed_files._read_ledger()[str(path)] == {"existed": True, "contents": "orig"} + + def test_capture_is_idempotent_across_writes(self, tmp_path, monkeypatch): + path = tmp_path / "managed.json" + path.write_text("orig", encoding="utf-8") + _fake_sudo(monkeypatch) + managed_files.write_managed_file(path, "first", display="X") + managed_files.write_managed_file(path, "second", display="X") + assert managed_files._read_ledger()[str(path)] == {"existed": True, "contents": "orig"} + + def test_unchanged_write_does_not_capture(self, tmp_path, monkeypatch): + path = tmp_path / "managed.json" + path.write_text("same", encoding="utf-8") + _fake_sudo(monkeypatch) + managed_files.write_managed_file(path, "same", display="X") + assert managed_files._read_ledger() == {} + + def test_dry_run_does_not_capture(self, tmp_path, monkeypatch): + path = tmp_path / "managed.json" + path.write_text("orig", encoding="utf-8") + _fake_sudo(monkeypatch) + config_io.set_dry_run(True) + managed_files.write_managed_file(path, "new", display="X") + assert managed_files._read_ledger() == {} + + def test_failed_sudo_records_no_ledger_entry(self, tmp_path, monkeypatch): + path = tmp_path / "managed.json" + path.write_text("orig", encoding="utf-8") + + def boom(path, text): + raise PermissionError("no root") + + monkeypatch.setattr(managed_files, "_sudo_replace", boom) + assert managed_files.write_managed_file(path, "new", display="X") == "skipped" + assert managed_files._read_ledger() == {} + + +class TestSnapshotOriginal: + """The captured pre-ucode state distinguishes absent / empty / readable / unreadable.""" + + def test_absent_file(self, tmp_path): + snap = managed_files._snapshot_original(tmp_path / "nope.json") + assert snap == {"existed": False, "contents": None} + + def test_readable_file(self, tmp_path): + path = tmp_path / "m.json" + path.write_text("orig", encoding="utf-8") + assert managed_files._snapshot_original(path) == {"existed": True, "contents": "orig"} + + def test_empty_file_is_empty_string_not_none(self, tmp_path): + path = tmp_path / "m.json" + path.write_text("", encoding="utf-8") + assert managed_files._snapshot_original(path) == {"existed": True, "contents": ""} + + def test_read_failure_routes_to_sudo_snapshot(self, tmp_path, monkeypatch): + path = tmp_path / "m.json" + path.write_text("secret IT config", encoding="utf-8") + orig_read = managed_files.Path.read_text + + def fake_read(self, *args, **kwargs): + if self.name == "m.json": + raise PermissionError("denied") + return orig_read(self, *args, **kwargs) + + monkeypatch.setattr(managed_files.Path, "read_text", fake_read) + monkeypatch.setattr( + managed_files, "_sudo_snapshot", lambda p: {"existed": True, "contents": "via sudo"} + ) + assert managed_files._snapshot_original(path) == {"existed": True, "contents": "via sudo"} + + def test_locked_parent_routes_to_sudo_snapshot(self, tmp_path, monkeypatch): + path = tmp_path / "m.json" + orig_exists = managed_files.Path.exists + + def fake_exists(self, *args, **kwargs): + if self.name == "m.json": + raise PermissionError("denied") + return orig_exists(self, *args, **kwargs) + + monkeypatch.setattr(managed_files.Path, "exists", fake_exists) + monkeypatch.setattr( + managed_files, "_sudo_snapshot", lambda p: {"existed": False, "contents": None} + ) + assert managed_files._snapshot_original(path) == {"existed": False, "contents": None} + + +class TestSudoSnapshot: + """The privileged fallback for a root-locked parent or root-only file.""" + + def _fake_run(self, monkeypatch, *, present, cat_rc=0, cat_out=""): + def run(cmd, *args, **kwargs): + if cmd[:2] == [managed_files._SUDO, "test"]: + return subprocess.CompletedProcess(cmd, 0 if present else 1) + if cmd[:2] == [managed_files._SUDO, "cat"]: + return subprocess.CompletedProcess(cmd, cat_rc, stdout=cat_out) + raise AssertionError(f"unexpected command {cmd}") + + monkeypatch.setattr(managed_files.subprocess, "run", run) + + def test_sudo_absent_is_existed_false(self, tmp_path, monkeypatch): + self._fake_run(monkeypatch, present=False) + assert managed_files._sudo_snapshot(tmp_path / "m") == {"existed": False, "contents": None} + + def test_root_only_readable_is_captured(self, tmp_path, monkeypatch): + self._fake_run(monkeypatch, present=True, cat_rc=0, cat_out="it config") + snap = managed_files._sudo_snapshot(tmp_path / "m") + assert snap == {"existed": True, "contents": "it config"} + + def test_unreadable_even_as_root_is_none(self, tmp_path, monkeypatch): + self._fake_run(monkeypatch, present=True, cat_rc=1) + assert managed_files._sudo_snapshot(tmp_path / "m") == {"existed": True, "contents": None} + + +class TestRestoreManagedFile: + """The inverse of `write_managed_file`: put the pre-ucode file back, or remove one ucode made.""" + + def test_workspace_a_to_b_restores_original_and_preserves_unrelated( + self, tmp_path, monkeypatch + ): + path = tmp_path / "managed.json" + path.write_text('{"itKey": "keep"}', encoding="utf-8") + _fake_sudo(monkeypatch) + assert ( + managed_files.write_managed_file( + path, '{"itKey": "keep", "ucode": 1}', display="Claude Code" + ) + == "written" + ) + assert managed_files.restore_managed_file(path, display="Claude Code") == "restored" + assert path.read_text(encoding="utf-8") == '{"itKey": "keep"}' + + def test_created_file_with_no_original_is_removed(self, tmp_path, monkeypatch): + path = tmp_path / "managed.json" # absent before ucode + _fake_sudo(monkeypatch) + managed_files.write_managed_file(path, "ucode", display="Codex") + assert path.exists() + assert managed_files.restore_managed_file(path, display="Codex") == "removed" + assert not path.exists() + + def test_second_restore_is_a_noop_without_privilege(self, tmp_path, monkeypatch): + path = tmp_path / "managed.json" + path.write_text("orig", encoding="utf-8") + calls = _fake_sudo(monkeypatch) + managed_files.write_managed_file(path, "ucode", display="X") + assert managed_files.restore_managed_file(path, display="X") == "restored" + before = len(calls) + assert managed_files.restore_managed_file(path, display="X") == "unchanged" + assert len(calls) == before + + def test_no_ledger_entry_is_a_noop(self, tmp_path, monkeypatch): + path = tmp_path / "managed.json" + path.write_text("some IT file", encoding="utf-8") + calls = _fake_sudo(monkeypatch) + assert managed_files.restore_managed_file(path, display="X") == "unchanged" + assert calls == [] + assert path.read_text(encoding="utf-8") == "some IT file" + + def test_missing_created_file_is_a_noop(self, tmp_path, monkeypatch): + path = tmp_path / "managed.json" + managed_files._write_ledger({str(path): {"existed": False, "contents": None}}) + calls = _fake_sudo(monkeypatch) + assert managed_files.restore_managed_file(path, display="X") == "unchanged" + assert calls == [] + + def test_preexisting_but_unreadable_original_is_left_untouched(self, tmp_path, monkeypatch): + path = tmp_path / "managed.json" + path.write_text("some IT file", encoding="utf-8") + managed_files._write_ledger({str(path): {"existed": True, "contents": None}}) + calls = _fake_sudo(monkeypatch) + assert managed_files.restore_managed_file(path, display="X") == "unchanged" + assert calls == [] + assert path.exists() + assert path.read_text(encoding="utf-8") == "some IT file" + + def test_unsupported_platform_is_safe(self, tmp_path, monkeypatch): + monkeypatch.setattr(managed_files, "managed_files_supported", lambda: False) + path = tmp_path / "managed.json" + managed_files._write_ledger({str(path): {"existed": True, "contents": "orig"}}) + calls = _fake_sudo(monkeypatch) + assert managed_files.restore_managed_file(path, display="X") == "unchanged" + assert calls == [] + + def test_permission_error_is_skipped_not_raised(self, tmp_path, monkeypatch): + path = tmp_path / "managed.json" + path.write_text("ucode", encoding="utf-8") + managed_files._write_ledger({str(path): {"existed": True, "contents": "orig"}}) + + def boom(path, text): + raise PermissionError("no root") + + monkeypatch.setattr(managed_files, "_sudo_replace", boom) + assert managed_files.restore_managed_file(path, display="X") == "skipped" + assert str(path) in managed_files._read_ledger() + + def test_sudo_failure_is_skipped_not_raised(self, tmp_path, monkeypatch): + path = tmp_path / "managed.json" + managed_files._write_ledger({str(path): {"existed": False, "contents": None}}) + path.write_text("ucode", encoding="utf-8") + + def boom(path): + raise subprocess.CalledProcessError(1, ["/usr/bin/sudo", "rm"], stderr="denied") + + monkeypatch.setattr(managed_files, "_sudo_remove", boom) + assert managed_files.restore_managed_file(path, display="X") == "skipped" + + def test_dry_run_restore_makes_no_privileged_call(self, tmp_path, monkeypatch): + path = tmp_path / "managed.json" + path.write_text("ucode", encoding="utf-8") + managed_files._write_ledger({str(path): {"existed": True, "contents": "orig"}}) + calls = _fake_sudo(monkeypatch) + config_io.set_dry_run(True) + assert managed_files.restore_managed_file(path, display="X") == "restored" + assert calls == [] + assert path.read_text(encoding="utf-8") == "ucode" + assert str(path) in managed_files._read_ledger()