feat: add topgrade configuration management plugin

plugins/topgrade/plugin.yaml

@@ -0,0 +1,6 @@
+name: topgrade
+version: 1.0.0
+type: python
+main: src/plugin.py
+capabilities:
+  - config_provider

plugins/topgrade/src/plugin.py

@@ -0,0 +1,146 @@
+# /// script
+# requires-python = ">=3.11"
+# dependencies = [
+#     "tomlkit",
+# ]
+# ///
+
+import json
+import os
+import shutil
+import sys
+
+import tomlkit
+
+
+def log(msg):
+    sys.stderr.write(f"[topgrade-plugin] {msg}\n")
+    sys.stderr.flush()
+
+
+def get_topgrade_config_path():
+    appdata = os.environ.get("APPDATA")
+    if not appdata:
+        return None
+
+    main_path = os.path.join(appdata, "topgrade", "topgrade.toml")
+    fallback_path = os.path.join(appdata, "topgrade.toml")
+
+    if os.path.exists(main_path):
+        return main_path
+    if os.path.exists(fallback_path):
+        return fallback_path
+
+    return main_path
+
+
+def merge_dict(target, source):
+    for k, v in source.items():
+        if isinstance(v, dict):
+            if k not in target:
+                target[k] = tomlkit.table()
+            if isinstance(target[k], dict):
+                merge_dict(target[k], v)
+            else:
+                target[k] = v
+        else:
+            target[k] = v
+
+
+def check_installed(args, request_id):
+    installed = shutil.which("topgrade") is not None
+    return installed
+
+
+def apply_config(args, request_id):
+    dry_run = args.get("dryRun", False)
+    settings = args.get("settings", {})
+
+    if not isinstance(settings, dict):
+        raise ValueError("settings must be an object")
+
+    config_path = get_topgrade_config_path()
+    if not config_path:
+        raise ValueError("APPDATA environment variable not set")
+
+    try:
+        if os.path.exists(config_path):
+            with open(config_path, "r", encoding="utf-8") as f:
+                doc = tomlkit.load(f)
+        else:
+            doc = tomlkit.document()
+            os.makedirs(os.path.dirname(config_path), exist_ok=True)
+
+        orig_content = doc.as_string()
+        merge_dict(doc, settings)
+        new_content = doc.as_string()
+        changed = orig_content != new_content
+
+        if not changed or dry_run:
+            if dry_run:
+                log(f"Would update {config_path}")
+            return {"changed": changed}
+
+        import tempfile
+
+        fd, temp_path = tempfile.mkstemp(dir=os.path.dirname(config_path), text=True)
+        try:
+            with os.fdopen(fd, "w", encoding="utf-8") as f:
+                f.write(new_content)
+            os.replace(temp_path, config_path)
+        except Exception as e:
+            os.remove(temp_path)
+            raise e
+
+        log(f"Updated topgrade config: {config_path}")
+        return {"changed": True}
+
+    except Exception as e:
+        log(f"Failed to apply config: {e}")
+        raise e
+
+
+def main():
+    input_data = sys.stdin.read()
+
+    if not input_data:
+        response = {"requestId": "unknown", "error": "No input received"}
+        sys.stdout.write(json.dumps(response) + "\n")
+        sys.stdout.flush()
+        return
+
+    try:
+        request = json.loads(input_data)
+    except Exception as e:
+        response = {"requestId": "unknown", "error": f"Failed to parse JSON request: {str(e)}"}
+        sys.stdout.write(json.dumps(response) + "\n")
+        sys.stdout.flush()
+        return
+
+    request_id = request.get("requestId")
+    if request_id is None:
+        request_id = "unknown"
+
+    command = request.get("command")
+    args = request.get("args", {})
+
+    response = {"requestId": request_id}
+
+    try:
+        if command == "check_installed":
+            installed = check_installed(args, request_id)
+            response["installed"] = installed
+        elif command == "apply":
+            res = apply_config(args, request_id)
+            response.update(res)
+        else:
+            response["error"] = f"Unknown command: {command}"
+    except Exception as fatal_err:
+        response["error"] = f"Internal error: {str(fatal_err)}"
+
+    sys.stdout.write(json.dumps(response) + "\n")
+    sys.stdout.flush()
+
+
+if __name__ == "__main__":
+    main()

plugins/topgrade/test/test_topgrade.py

@@ -0,0 +1,82 @@
+import json
+import os
+import subprocess
+
+import tomlkit
+
+PLUGIN_SCRIPT = os.path.join(os.path.dirname(__file__), "..", "src", "plugin.py")
+
+
+def run_plugin(request_dict):
+    """Helper to run the plugin script with a given JSON request via uv."""
+    input_str = json.dumps(request_dict)
+    # Use standard python to run since uv might not be perfectly nested here,
+    # but the script uses inline dependencies so we must use 'uv run'
+    result = subprocess.run(["uv", "run", PLUGIN_SCRIPT], input=input_str, text=True, capture_output=True)
+    return json.loads(result.stdout) if result.stdout else None, result.stderr
+
+
+def test_check_installed():
+    req = {"requestId": "123", "command": "check_installed"}
+    resp, stderr = run_plugin(req)
+    assert resp is not None
+    assert resp["requestId"] == "123"
+    assert isinstance(resp["installed"], bool)
+
+
+def test_apply_new_file(monkeypatch, tmp_path):
+    # Set APPDATA to a temporary directory
+    monkeypatch.setenv("APPDATA", str(tmp_path))
+
+    settings = {"disable": ["pip", "npm"], "set_title": True, "git_repos": {"~/Projects/dotfiles": "main"}}
+
+    req = {"requestId": "456", "command": "apply", "args": {"settings": settings}}
+
+    resp, stderr = run_plugin(req)
+    assert resp is not None
+    assert resp["requestId"] == "456"
+    assert resp.get("changed") is True
+
+    config_file = tmp_path / "topgrade" / "topgrade.toml"
+    assert config_file.exists()
+
+    with open(config_file, "r", encoding="utf-8") as f:
+        doc = tomlkit.load(f)
+
+    assert doc["disable"] == ["pip", "npm"]
+    assert doc["set_title"] is True
+    assert doc["git_repos"]["~/Projects/dotfiles"] == "main"
+
+
+def test_apply_merge_existing(monkeypatch, tmp_path):
+    monkeypatch.setenv("APPDATA", str(tmp_path))
+
+    config_dir = tmp_path / "topgrade"
+    config_dir.mkdir(parents=True, exist_ok=True)
+    config_file = config_dir / "topgrade.toml"
+
+    initial_content = """
+disable = ["gem"]
+display_time = true
+
+[git_repos]
+"~/Projects/old" = "master"
+"""
+    with open(config_file, "w", encoding="utf-8") as f:
+        f.write(initial_content)
+
+    settings = {"disable": ["pip", "npm"], "git_repos": {"~/Projects/dotfiles": "main"}}
+
+    req = {"requestId": "789", "command": "apply", "args": {"settings": settings}}
+
+    resp, stderr = run_plugin(req)
+    assert resp is not None
+    assert resp.get("changed") is True
+
+    with open(config_file, "r", encoding="utf-8") as f:
+        doc = tomlkit.load(f)
+
+    assert doc["disable"] == ["pip", "npm"]
+    assert doc["display_time"] is True
+    assert doc["git_repos"]["~/Projects/old"] == "master"
+    assert doc["git_repos"]["~/Projects/dotfiles"] == "main"