From 0725a46f974bf8c845e648ba60002b9012a58a48 Mon Sep 17 00:00:00 2001 From: Ethanfel Date: Sun, 21 Jun 2026 19:50:52 +0200 Subject: [PATCH] feat: profiles registry read/write + find Co-Authored-By: Claude Opus 4.8 --- gates/profiles.py | 48 ++++++++++++++++++++++++++++++++++++++++++ tests/test_profiles.py | 24 +++++++++++++++++++++ 2 files changed, 72 insertions(+) create mode 100644 gates/profiles.py create mode 100644 tests/test_profiles.py diff --git a/gates/profiles.py b/gates/profiles.py new file mode 100644 index 0000000..6e4525e --- /dev/null +++ b/gates/profiles.py @@ -0,0 +1,48 @@ +"""Named-profile registry + dir ops for the Image Pool. Stdlib only.""" +import json +import os +import shutil +import zipfile +from pathlib import Path + +REGISTRY_NAME = "profiles.json" + + +def registry_path(base): + return Path(base) / REGISTRY_NAME + + +def empty_registry(): + return {"profiles": []} + + +def read_registry(base): + p = registry_path(base) + if not p.exists(): + return empty_registry() + try: + with open(p, "r", encoding="utf-8") as f: + reg = json.load(f) + if not isinstance(reg, dict) or "profiles" not in reg: + raise ValueError("bad registry") + return reg + except (ValueError, json.JSONDecodeError): + return empty_registry() + + +def write_registry(base, reg): + Path(base).mkdir(parents=True, exist_ok=True) + final = registry_path(base) + tmp = final.with_name(REGISTRY_NAME + ".tmp") + with open(tmp, "w", encoding="utf-8") as f: + json.dump(reg, f, indent=2) + os.replace(tmp, final) + return reg + + +def find_by_id(reg, pid): + return next((p for p in reg["profiles"] if p["id"] == pid), None) + + +def find_by_name(reg, name): + return next((p for p in reg["profiles"] if p["name"] == name), None) diff --git a/tests/test_profiles.py b/tests/test_profiles.py new file mode 100644 index 0000000..b17e899 --- /dev/null +++ b/tests/test_profiles.py @@ -0,0 +1,24 @@ +# tests/test_profiles.py +from gates import profiles as pr + +def test_empty_registry(): + assert pr.empty_registry() == {"profiles": []} + +def test_read_missing_is_empty(tmp_path): + assert pr.read_registry(str(tmp_path)) == {"profiles": []} + +def test_write_then_read(tmp_path): + reg = {"profiles": [{"id": "a", "name": "n", "created": 1}]} + pr.write_registry(str(tmp_path), reg) + assert (tmp_path / "profiles.json").exists() + assert pr.read_registry(str(tmp_path)) == reg + +def test_read_corrupt_is_empty(tmp_path): + (tmp_path / "profiles.json").write_text("{ not json") + assert pr.read_registry(str(tmp_path)) == {"profiles": []} + +def test_find_helpers(): + reg = {"profiles": [{"id": "a", "name": "x"}, {"id": "b", "name": "y"}]} + assert pr.find_by_id(reg, "b")["name"] == "y" + assert pr.find_by_name(reg, "x")["id"] == "a" + assert pr.find_by_id(reg, "z") is None