From fff7dffe7ea92d977e5ec293d075698c950a8dd5 Mon Sep 17 00:00:00 2001 From: David Liu Date: Wed, 7 Oct 2026 14:31:55 +0000 Subject: [PATCH] Record the settings key paths ug writes, per destination file Cleanup in the Claude and Codex writers removes settings by key name, so it also deletes values another tool or the developer wrote. This adds the provenance record later changes use to prove authorship: for each agent and destination file, the key paths ug wrote and the exact values it wrote there. - provenance.py stores the record under the managed-backups directory, guarded by a file lock so concurrent ug runs don't drop each other's records. - owned_after_write keeps only values that landed, or that ug wrote earlier and nobody changed; retire and retire_group restore a baseline value or delete only owned, unedited paths. Comparisons are type-strict. - values_at reads the values a document holds at given key paths. - ug revert clears the store after clearing state. Co-authored-by: Isaac --- AGENTS.md | 8 + src/ucode/cli.py | 2 + src/ucode/provenance.py | 202 ++++++++++++++++++++++ tests/conftest.py | 1 + tests/test_cli.py | 34 ++++ tests/test_provenance.py | 356 +++++++++++++++++++++++++++++++++++++++ 6 files changed, 603 insertions(+) create mode 100644 src/ucode/provenance.py create mode 100644 tests/test_provenance.py 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"}}