diff --git a/plugins/topgrade/plugin.yaml b/plugins/topgrade/plugin.yaml new file mode 100644 index 00000000..1a6fc988 --- /dev/null +++ b/plugins/topgrade/plugin.yaml @@ -0,0 +1,6 @@ +name: topgrade +version: 1.0.0 +type: python +main: src/plugin.py +capabilities: + - config_provider diff --git a/plugins/topgrade/src/plugin.py b/plugins/topgrade/src/plugin.py new file mode 100644 index 00000000..ed168789 --- /dev/null +++ b/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() diff --git a/plugins/topgrade/test/test_topgrade.py b/plugins/topgrade/test/test_topgrade.py new file mode 100644 index 00000000..619f4e78 --- /dev/null +++ b/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"