diff --git a/AGENTS.md b/AGENTS.md index 804b5ce6f..f172c826b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -49,6 +49,14 @@ The two columns show what `ug` does without and with a [managed config](https:// This is the intended behavior; we are still working to make the code match it in every case. In particular, `ug` today also clears some fields that this list leaves alone: the `env.ANTHROPIC_DEFAULT_*_MODEL_NAME` companions, `env.CLAUDE_CODE_ENABLE_GATEWAY_MODEL_DISCOVERY`, `env.CLAUDE_CODE_DISABLE_EXPERIMENTAL_BETAS`, and Codex `model_reasoning_effort`. Codex `[otel]` is never written to a file. +### How `ug` tracks what it wrote (provenance) + +`ug` shares these files with other writers (an admin, Isaac, MDM, or the developer), so when it cleans up a field it previously set it must not delete a value someone else owns at the same key. For that, `ug` keeps a provenance record at `~/.ucode/managed-backups/provenance.json`: for each agent and destination file, the key paths `ug` wrote and the exact value it wrote at each. + +- `ug` removes or restores a field only while the file still holds the value `ug` last wrote there. If anyone else has since changed that value, `ug` gives up the claim and leaves the field alone. +- The record is refreshed after each confirmed write by `ug configure` (including the auto-configure a launch runs first), and `ug revert` clears it. +- It is not a mirror of this whole table. It holds only the keys a cleanup path actually consults (the fields `ug` conditionally removes), not every field `ug` writes. Values other writers set are never recorded. +
Claude Code diff --git a/src/ucode/cli.py b/src/ucode/cli.py index 656abd746..a5a72850a 100644 --- a/src/ucode/cli.py +++ b/src/ucode/cli.py @@ -125,6 +125,7 @@ revert_mcp_configs, ) from ucode.os_compatibility import subprocess_cross_os +from ucode.provenance import clear_provenance from ucode.skills_download import ( configure_location_skills_download_command, configure_selected_skills_download_command, @@ -1279,6 +1280,7 @@ def revert() -> int: # place; restoring the per-profile file above does not undo that. legacy_codex_stripped = revert_legacy_shared_config() clear_state() + clear_provenance() print_heading("Revert") print_kv("Workspace", state.get("workspace") or "none") diff --git a/src/ucode/provenance.py b/src/ucode/provenance.py new file mode 100644 index 000000000..dfc0def93 --- /dev/null +++ b/src/ucode/provenance.py @@ -0,0 +1,202 @@ +"""Per-file record of the settings key paths ug wrote, and the exact values it wrote there. + +Later reconciles use the record to tell ug's own keys apart from a user's, so ug only ever +restores or deletes values it can prove it wrote. ``ug revert`` clears the store. +""" + +from __future__ import annotations + +import copy +import json +import os +from collections.abc import Iterable, Iterator +from contextlib import contextmanager +from pathlib import Path +from typing import Any + +from ucode import managed_files +from ucode.config_io import is_dry_run +from ucode.os_compatibility.file_lock_cross_os import ( + acquire_exclusive_file_lock, + release_file_lock, +) + +KeyPath = tuple[str, ...] + + +def load_provenance(agent: str, path: Path) -> dict[KeyPath, Any] | None: + """Return the key paths ug recorded for this file, or None when there is no record yet.""" + store = _load_store() + if store is None: + return None + return _parse_entries(store["files"].get(_file_key(agent, path))) + + +def save_provenance(agent: str, path: Path, owned: dict[KeyPath, Any]) -> None: + """Replace the recorded key paths and values ug wrote to this file.""" + if is_dry_run(): + return + with _store_lock(): + store = _load_store() or {"files": {}} + store["files"][_file_key(agent, path)] = [ + {"path": list(key_path), "value": value} for key_path, value in sorted(owned.items()) + ] + managed_files._write_private_file(_store_path(), json.dumps(store, indent=2) + "\n") + + +def clear_provenance() -> None: + """Delete the whole store, as part of ``ug revert``; a missing store is fine.""" + if is_dry_run() or not _store_path().exists(): + return + try: + with _store_lock(): + _store_path().unlink(missing_ok=True) + except OSError as exc: + raise RuntimeError(f"Could not remove the provenance store: {exc}") from exc + + +def owned_after_write( + owned: dict[KeyPath, Any], + generated: dict[KeyPath, Any], + final: dict, + before: dict | None = None, +) -> dict[KeyPath, Any]: + """Keep generated paths that landed in ``final`` and prior owned paths still intact there. + + With ``before`` (the file before this write), a generated value the file already held is + adopted only if that held value was still ug's own, so an equal value someone else wrote, or an + edit that happens to match, stays theirs. + """ + result: dict[KeyPath, Any] = { + key_path: copy.deepcopy(value) + for key_path, value in generated.items() + if _is_intact(final, key_path, value) + and ( + before is None + or not _is_intact(before, key_path, value) + or (key_path in owned and _is_intact(before, key_path, owned[key_path])) + ) + } + for key_path, value in owned.items(): + if key_path not in result and _is_intact(final, key_path, value): + result[key_path] = copy.deepcopy(value) + return result + + +def values_at(key_paths: list[list[str]], doc: dict | None) -> dict[KeyPath, Any]: + """The values ``doc`` holds at ``key_paths``, skipping paths it lacks; None means no document.""" + if doc is None: + return {} + owned: dict[KeyPath, Any] = {} + for key_path in map(tuple, key_paths): + value = _value_at(doc, key_path) + if value is not managed_files._MISSING: + owned[key_path] = copy.deepcopy(value) + return owned + + +def retire( + doc: dict, owned: dict[KeyPath, Any], paths: Iterable[KeyPath], baseline: dict | None +) -> None: + """Restore each owned, unedited path to its baseline value, or delete it when none exists.""" + for key_path in paths: + if key_path in owned and _is_intact(doc, key_path, owned[key_path]): + _restore(doc, key_path, baseline) + + +def retire_group( + doc: dict, owned: dict[KeyPath, Any], paths: Iterable[KeyPath], baseline: dict | None +) -> None: + """Like :func:`retire`, but all or nothing. + + A member that ug owns but was edited or deleted, or a present member that ug does not own and + that differs from the baseline, vetoes the group; the veto drops ug's ownership of every member, + so a later run never retires a remnant of someone else's group. Unowned members still at their + baseline value stay. + """ + paths = list(paths) + present = [ + key_path for key_path in paths if _value_at(doc, key_path) is not managed_files._MISSING + ] + + def vetoes(key_path: KeyPath) -> bool: + if key_path in owned: + return not _is_intact(doc, key_path, owned[key_path]) + return key_path in present and ( + baseline is None or not _is_intact(baseline, key_path, _value_at(doc, key_path)) + ) + + if any(vetoes(key_path) for key_path in paths): + for key_path in paths: + owned.pop(key_path, None) + return + for key_path in present: + _restore(doc, key_path, baseline) + + +def _store_path() -> Path: + return managed_files.MANAGED_BACKUP_DIR / "provenance.json" + + +@contextmanager +def _store_lock() -> Iterator[None]: + """Serialize the store's read-modify-write across concurrent ug processes with an OS file lock. + + Keep the lock file in place: unlinking it could let waiters lock different inodes. + """ + backup_dir = managed_files.MANAGED_BACKUP_DIR + backup_dir.mkdir(parents=True, exist_ok=True) + os.chmod(backup_dir, 0o700) + with (backup_dir / "provenance.lock").open("a+b") as lock_file: + acquire_exclusive_file_lock(lock_file) + try: + yield + finally: + release_file_lock(lock_file) + + +def _file_key(agent: str, path: Path) -> str: + return f"{agent}:{path}" + + +def _load_store() -> dict | None: + try: + store = json.loads(_store_path().read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError): + return None + if not isinstance(store, dict) or not isinstance(store.get("files"), dict): + return None + return store + + +def _parse_entries(entries: object) -> dict[KeyPath, Any] | None: + if not isinstance(entries, list): + return None + owned: dict[KeyPath, Any] = {} + for entry in entries: + if not isinstance(entry, dict) or "value" not in entry: + continue + path = entry.get("path") + if isinstance(path, list) and all(isinstance(part, str) for part in path): + owned[tuple(path)] = entry["value"] + return owned + + +def _value_at(doc: dict, key_path: KeyPath) -> Any: + return managed_files._path_value(doc, list(key_path)) + + +def _is_intact(doc: dict, key_path: KeyPath, value: Any) -> bool: + """Whether ``doc`` still holds exactly ``value`` at ``key_path``, comparing types strictly.""" + current = _value_at(doc, key_path) + if current is managed_files._MISSING: + return False + return managed_files.is_semantically_equal(current, value) + + +def _restore(doc: dict, key_path: KeyPath, baseline: dict | None) -> None: + value = _value_at(baseline, key_path) if baseline is not None else managed_files._MISSING + if value is managed_files._MISSING: + managed_files._delete_path_value(doc, list(key_path)) + else: + managed_files._set_path_value(doc, list(key_path), value) diff --git a/tests/conftest.py b/tests/conftest.py index 2b0c9f090..f72fba2bd 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -37,6 +37,7 @@ def _isolate_ucode_state(tmp_path, monkeypatch): import ucode.state as state_mod from ucode.agents import codex as codex_mod + config_io_mod.set_dry_run(False) state_dir = tmp_path / ".ucode" state_dir.mkdir() monkeypatch.setattr(state_mod, "STATE_PATH", state_dir / "state.json") diff --git a/tests/test_cli.py b/tests/test_cli.py index 7e5124b8b..a233cf528 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -2825,6 +2825,40 @@ def test_reverts_mcp_configs_before_clearing_state(self): assert cleared == [True] assert "Claude Code MCP config: restored" in result.output + def test_clears_provenance_store(self): + from ucode import managed_files, provenance + + provenance.save_provenance("claude", Path("/tmp/settings.json"), {("env", "A"): "1"}) + store_path = managed_files.MANAGED_BACKUP_DIR / "provenance.json" + assert store_path.exists() + + with ( + patch("ucode.cli.load_state", return_value=MINIMAL_STATE), + patch("ucode.cli.restore_file", return_value=False), + patch("ucode.cli.revert_mcp_configs", return_value={}), + patch("ucode.cli.clear_state"), + ): + result = runner.invoke(app, ["revert"]) + + assert result.exit_code == 0, result.output + assert not store_path.exists() + + def test_clears_state_even_when_clearing_provenance_fails(self): + cleared: list[bool] = [] + + with ( + patch("ucode.cli.load_state", return_value=MINIMAL_STATE), + patch("ucode.cli.restore_file", return_value=False), + patch("ucode.cli.revert_mcp_configs", return_value={}), + patch("ucode.cli.clear_state", side_effect=lambda: cleared.append(True)), + patch("ucode.cli.clear_provenance", side_effect=RuntimeError("boom")), + ): + result = runner.invoke(app, ["revert"]) + + assert result.exit_code == 1 + assert "boom" in _strip_ansi(result.output) + assert cleared == [True] + class TestDoctorCommand: def test_invokes_doctor(self): diff --git a/tests/test_provenance.py b/tests/test_provenance.py new file mode 100644 index 000000000..2bcdd0048 --- /dev/null +++ b/tests/test_provenance.py @@ -0,0 +1,356 @@ +"""Tests for the per-file provenance store.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +import ucode.config_io as config_io +from ucode import managed_files, provenance + + +@pytest.fixture +def store_path() -> Path: + return managed_files.MANAGED_BACKUP_DIR / "provenance.json" + + +class TestLoadSave: + def test_load_returns_none_without_store(self): + assert provenance.load_provenance("claude", Path("/tmp/a.json")) is None + + def test_round_trip_keeps_files_and_agents_independent(self): + claude_path = Path("/etc/claude-code/managed-settings.json") + codex_path = Path("/etc/codex/managed_config.toml") + provenance.save_provenance( + "claude", claude_path, {("env", "A"): "1", ("apiKeyHelper",): "/bin/helper"} + ) + provenance.save_provenance("codex", codex_path, {("model_provider",): "Databricks"}) + + assert provenance.load_provenance("claude", claude_path) == { + ("env", "A"): "1", + ("apiKeyHelper",): "/bin/helper", + } + assert provenance.load_provenance("codex", codex_path) == { + ("model_provider",): "Databricks" + } + assert provenance.load_provenance("codex", claude_path) is None + assert provenance.load_provenance("claude", codex_path) is None + + def test_save_replaces_prior_record_but_keeps_other_files(self): + path_a = Path("/tmp/a.json") + path_b = Path("/tmp/b.json") + provenance.save_provenance("claude", path_a, {("x",): 1}) + provenance.save_provenance("claude", path_b, {("y",): 2}) + provenance.save_provenance("claude", path_a, {}) + + assert provenance.load_provenance("claude", path_a) == {} + assert provenance.load_provenance("claude", path_b) == {("y",): 2} + + def test_explicit_null_value_round_trips(self): + path = Path("/tmp/a.json") + provenance.save_provenance("claude", path, {("env", "A"): None}) + assert provenance.load_provenance("claude", path) == {("env", "A"): None} + + @pytest.mark.parametrize( + "contents", + [b"not json{", b"\xff\xfe{", json.dumps({"files": []}).encode(), None], + ids=["corrupt", "invalid-utf8", "files-not-dict", "unreadable"], + ) + def test_invalid_store_means_no_record(self, store_path, contents): + if contents is None: + store_path.mkdir(parents=True) + else: + store_path.parent.mkdir(parents=True, exist_ok=True) + store_path.write_bytes(contents) + assert provenance.load_provenance("claude", Path("/tmp/a.json")) is None + + def test_clear_without_a_store_leaves_the_backup_dir_untouched(self, store_path): + assert not store_path.parent.exists() + provenance.clear_provenance() + assert not store_path.parent.exists() + + def test_clear_removes_the_store(self, store_path): + provenance.save_provenance("claude", Path("/tmp/a.json"), {("x",): 1}) + provenance.clear_provenance() + assert not store_path.exists() + assert provenance.load_provenance("claude", Path("/tmp/a.json")) is None + + +class TestLocking: + def test_save_holds_the_lock_across_read_and_write(self, monkeypatch): + events: list[str] = [] + real_load = provenance._load_store + real_write = managed_files._write_private_file + + def record_load(): + events.append("read") + return real_load() + + def record_write(path, text): + events.append("write") + real_write(path, text) + + monkeypatch.setattr( + provenance, "acquire_exclusive_file_lock", lambda _f: events.append("acquire") + ) + monkeypatch.setattr(provenance, "release_file_lock", lambda _f: events.append("release")) + monkeypatch.setattr(provenance, "_load_store", record_load) + monkeypatch.setattr(managed_files, "_write_private_file", record_write) + + provenance.save_provenance("claude", Path("/tmp/a.json"), {("x",): 1}) + + assert events == ["acquire", "read", "write", "release"] + + def test_clear_holds_the_lock_while_removing_the_store(self, monkeypatch, store_path): + provenance.save_provenance("claude", Path("/tmp/a.json"), {("x",): 1}) + events: list[tuple[str, bool]] = [] + monkeypatch.setattr( + provenance, + "acquire_exclusive_file_lock", + lambda _f: events.append(("acquire", store_path.exists())), + ) + monkeypatch.setattr( + provenance, + "release_file_lock", + lambda _f: events.append(("release", store_path.exists())), + ) + + provenance.clear_provenance() + + assert events == [("acquire", True), ("release", False)] + + +class TestDryRun: + @pytest.fixture(autouse=True) + def _restore_dry_run(self): + yield + config_io.set_dry_run(False) + + def test_save_touches_nothing_and_takes_no_lock(self, monkeypatch, store_path): + acquired: list[object] = [] + monkeypatch.setattr(provenance, "acquire_exclusive_file_lock", lambda f: acquired.append(f)) + config_io.set_dry_run(True) + provenance.save_provenance("claude", Path("/tmp/a.json"), {("x",): 1}) + assert not store_path.exists() + assert acquired == [] + + def test_clear_does_not_delete_or_lock(self, monkeypatch, store_path): + provenance.save_provenance("claude", Path("/tmp/a.json"), {("x",): 1}) + acquired: list[object] = [] + monkeypatch.setattr(provenance, "acquire_exclusive_file_lock", lambda f: acquired.append(f)) + config_io.set_dry_run(True) + provenance.clear_provenance() + assert store_path.exists() + assert acquired == [] + + +class TestOwnedAfterWrite: + def test_generated_value_that_landed_is_owned(self): + final = {"env": {"A": "1"}} + assert provenance.owned_after_write({}, {("env", "A"): "1"}, final) == {("env", "A"): "1"} + + def test_owned_values_do_not_alias_the_written_document(self): + deny = ["WebSearch"] + final = {"permissions": {"deny": deny}} + owned = provenance.owned_after_write({}, {("permissions", "deny"): deny}, final) + deny.append("Bash(rm:*)") + assert owned == {("permissions", "deny"): ["WebSearch"]} + + def test_generated_value_replaced_in_final_is_not_owned(self): + final = {"env": {"A": "edited"}} + assert provenance.owned_after_write({}, {("env", "A"): "1"}, final) == {} + + def test_generated_value_absent_in_final_is_not_owned(self): + assert provenance.owned_after_write({}, {("env", "A"): "1"}, {"env": {}}) == {} + + def test_value_the_file_already_held_is_not_adopted(self): + doc = {"env": {"A": "1"}} + assert provenance.owned_after_write({}, {("env", "A"): "1"}, doc, before=doc) == {} + + def test_value_this_write_changed_is_adopted(self): + before = {"env": {"A": "old"}} + final = {"env": {"A": "1"}} + owned = provenance.owned_after_write({}, {("env", "A"): "1"}, final, before=before) + assert owned == {("env", "A"): "1"} + + def test_already_owned_unchanged_value_stays_owned_with_before(self): + doc = {"env": {"A": "1"}} + owned = provenance.owned_after_write( + {("env", "A"): "1"}, {("env", "A"): "1"}, doc, before=doc + ) + assert owned == {("env", "A"): "1"} + + def test_edit_that_matches_the_next_generated_value_stays_theirs(self): + doc = {"model": "B"} + owned = provenance.owned_after_write({("model",): "A"}, {("model",): "B"}, doc, before=doc) + assert owned == {} + + def test_previously_owned_ungenerated_and_unchanged_is_kept(self): + final = {"env": {"A": "1"}} + assert provenance.owned_after_write({("env", "A"): "1"}, {}, final) == {("env", "A"): "1"} + + def test_previously_owned_ungenerated_and_changed_is_dropped(self): + final = {"env": {"A": "edited"}} + assert provenance.owned_after_write({("env", "A"): "1"}, {}, final) == {} + + def test_generated_value_wins_over_prior_owned_value(self): + final = {"env": {"A": "new"}} + assert provenance.owned_after_write( + {("env", "A"): "old"}, {("env", "A"): "new"}, final + ) == {("env", "A"): "new"} + + def test_failed_generated_write_keeps_the_prior_owned_value(self): + final = {"env": {"A": "old"}} + assert provenance.owned_after_write( + {("env", "A"): "old"}, {("env", "A"): "new"}, final + ) == {("env", "A"): "old"} + + def test_third_party_value_drops_ownership(self): + final = {"env": {"A": "other"}} + assert ( + provenance.owned_after_write({("env", "A"): "old"}, {("env", "A"): "new"}, final) == {} + ) + + def test_bool_is_not_interchangeable_with_int(self): + assert provenance.owned_after_write({}, {("env", "A"): True}, {"env": {"A": 1}}) == {} + assert provenance.owned_after_write({}, {("env", "A"): [True]}, {"env": {"A": [1]}}) == {} + + def test_explicit_null_is_a_present_value(self): + generated = {("env", "A"): None} + assert provenance.owned_after_write({}, generated, {"env": {"A": None}}) == generated + assert provenance.owned_after_write({}, generated, {"env": {}}) == {} + + +class TestValuesAt: + def test_picks_present_values(self): + doc = {"env": {"A": "1"}, "apiKeyHelper": "/bin/helper"} + owned = provenance.values_at([["env", "A"], ["apiKeyHelper"]], doc) + assert owned == {("env", "A"): "1", ("apiKeyHelper",): "/bin/helper"} + + def test_skips_absent_paths(self): + owned = provenance.values_at([["env", "A"], ["env", "B"], ["other"]], {"env": {"A": "1"}}) + assert owned == {("env", "A"): "1"} + + def test_none_doc_owns_nothing(self): + assert provenance.values_at([["env", "A"]], None) == {} + + def test_values_do_not_alias_the_document(self): + doc = {"permissions": {"deny": ["WebSearch"]}} + owned = provenance.values_at([["permissions", "deny"]], doc) + doc["permissions"]["deny"].append("Bash(rm:*)") + assert owned == {("permissions", "deny"): ["WebSearch"]} + + def test_explicit_null_is_present(self): + assert provenance.values_at([["env", "A"]], {"env": {"A": None}}) == {("env", "A"): None} + + +class TestRetire: + def test_restores_baseline_value(self): + doc = {"env": {"A": "ug"}} + provenance.retire(doc, {("env", "A"): "ug"}, [("env", "A")], {"env": {"A": "orig"}}) + assert doc == {"env": {"A": "orig"}} + + def test_deletes_without_baseline_and_prunes_empty_parents(self): + doc = {"env": {"A": "ug"}, "other": 1} + provenance.retire(doc, {("env", "A"): "ug"}, [("env", "A")], {}) + assert doc == {"other": 1} + + def test_none_baseline_deletes(self): + doc = {"env": {"A": "ug"}} + provenance.retire(doc, {("env", "A"): "ug"}, [("env", "A")], None) + assert doc == {} + + def test_keeps_edited_owned_value(self): + doc = {"env": {"A": "edited"}} + provenance.retire(doc, {("env", "A"): "ug"}, [("env", "A")], {"env": {"A": "orig"}}) + assert doc == {"env": {"A": "edited"}} + + def test_leaves_unowned_paths_untouched(self): + doc = {"env": {"A": "user"}} + provenance.retire(doc, {}, [("env", "A")], {}) + assert doc == {"env": {"A": "user"}} + + def test_deleted_owned_value_counts_as_changed(self): + doc: dict = {"env": {}} + provenance.retire(doc, {("env", "A"): "ug"}, [("env", "A")], {"env": {"A": "orig"}}) + assert doc == {"env": {}} + + def test_explicit_null_baseline_is_restored_as_null(self): + doc = {"env": {"A": "ug"}} + provenance.retire(doc, {("env", "A"): "ug"}, [("env", "A")], {"env": {"A": None}}) + assert doc == {"env": {"A": None}} + + def test_bool_owned_value_against_int_live_value_counts_as_edited(self): + doc = {"flag": 1} + provenance.retire(doc, {("flag",): True}, [("flag",)], {"flag": False}) + assert doc == {"flag": 1} + + def test_nested_bool_list_against_int_list_counts_as_edited(self): + doc = {"flags": [1]} + provenance.retire(doc, {("flags",): [True]}, [("flags",)], {}) + assert doc == {"flags": [1]} + + def test_restored_value_does_not_alias_the_baseline(self): + baseline = {"env": {"A": {"nested": [1]}}} + doc = {"env": {"A": "ug"}} + provenance.retire(doc, {("env", "A"): "ug"}, [("env", "A")], baseline) + doc["env"]["A"]["nested"].append(2) + assert baseline == {"env": {"A": {"nested": [1]}}} + + +class TestRetireGroup: + PATHS = [("env", "A"), ("env", "B")] + + def test_all_owned_members_unchanged_retires_all(self): + doc = {"env": {"A": "ug", "B": "ug"}} + owned = {("env", "A"): "ug", ("env", "B"): "ug"} + provenance.retire_group(doc, owned, self.PATHS, {"env": {"A": "orig"}}) + assert doc == {"env": {"A": "orig"}} + + def test_one_edited_member_blocks_the_whole_group(self): + doc = {"env": {"A": "ug", "B": "edited"}} + owned = {("env", "A"): "ug", ("env", "B"): "ug"} + provenance.retire_group(doc, owned, self.PATHS, {"env": {"A": "orig", "B": "orig-b"}}) + assert doc == {"env": {"A": "ug", "B": "edited"}} + + def test_deleted_owned_member_blocks_the_whole_group(self): + doc = {"env": {"A": "ug"}} + owned = {("env", "A"): "ug", ("env", "B"): "ug"} + provenance.retire_group(doc, owned, self.PATHS, {"env": {"A": "orig", "B": "orig-b"}}) + assert doc == {"env": {"A": "ug"}} + assert owned == {} + + def test_bool_member_against_int_live_value_blocks_the_group(self): + doc = {"env": {"A": 1, "B": "ug"}} + owned = {("env", "A"): True, ("env", "B"): "ug"} + provenance.retire_group(doc, owned, self.PATHS, {}) + assert doc == {"env": {"A": 1, "B": "ug"}} + + def test_present_unowned_member_blocks_the_group(self): + doc = {"env": {"A": "ug", "B": "user"}} + owned = {("env", "A"): "ug"} + provenance.retire_group(doc, owned, self.PATHS, {}) + assert doc == {"env": {"A": "ug", "B": "user"}} + assert owned == {} + + def test_unowned_member_still_at_its_baseline_does_not_block(self): + doc = {"env": {"A": "ug", "B": "1"}} + provenance.retire_group( + doc, {("env", "A"): "ug"}, self.PATHS, {"env": {"A": "orig", "B": "1"}} + ) + assert doc == {"env": {"A": "orig", "B": "1"}} + + def test_absent_unowned_member_does_not_block(self): + doc = {"env": {"A": "ug"}} + provenance.retire_group(doc, {("env", "A"): "ug"}, self.PATHS, {}) + assert doc == {} + + def test_veto_forgets_the_group_so_a_later_run_keeps_the_remnant(self): + doc = {"env": {"A": "ug", "B": "edited"}} + owned = {("env", "A"): "ug", ("env", "B"): "ug"} + provenance.retire_group(doc, owned, self.PATHS, {}) + assert owned == {} + provenance.retire_group(doc, provenance.owned_after_write(owned, {}, doc), self.PATHS, {}) + assert doc == {"env": {"A": "ug", "B": "edited"}}