feat: profiles registry read/write + find
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -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)
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user