From f320d648369da213dc5d55188beb41ea287eaa79 Mon Sep 17 00:00:00 2001 From: Kevin Veen-Birkenbach Date: Wed, 2 Sep 2026 00:19:43 +0200 Subject: [PATCH] feat(cli): edit role bonds in a matrix instead of hunting through files A bond lives on the consumer's side of a service relation, so seeing how two roles couple means opening both roles' meta/services.yml and comparing, and retuning one means editing a scalar buried among dozens of unrelated keys. Add `infinito meta roles applications bond` (make bond / i8bond): it resolves every services..bond to a role-to-role edge through the service registry and serves the result as a matrix, each cell carrying both directions because the two need not agree. Brightness is the bond, black for none and white for 1, so the coupling of the whole repository reads at a glance. Clicking a cell writes the new value back to the declaring role. The write is line surgery rather than a YAML round-trip: these files carry `# nocheck:` directives and hand-tuned ordering that re-dumping would flatten, so only the bond scalar moves and a trailing comment on that line survives. The browser sends a role pair, never a path, and the server resolves the service key from its own edge map, so a pair it does not know is refused before any write. `--out FILE` exports a read-only page and `--format json` the raw edges, for callers that want to post-process rather than edit. Co-Authored-By: Claude Opus 5 (1M context) --- Makefile | 5 + aliases | 1 + cli/meta/roles/applications/bond/__init__.py | 1 + cli/meta/roles/applications/bond/__main__.py | 7 + cli/meta/roles/applications/bond/cli.py | 97 ++++++++ cli/meta/roles/applications/bond/edit.py | 103 +++++++++ cli/meta/roles/applications/bond/matrix.py | 77 +++++++ cli/meta/roles/applications/bond/render.py | 218 ++++++++++++++++++ cli/meta/roles/applications/bond/serve.py | 111 +++++++++ .../meta/roles/applications/bond/__init__.py | 0 .../meta/roles/applications/bond/test_edit.py | 76 ++++++ 11 files changed, 696 insertions(+) create mode 100644 cli/meta/roles/applications/bond/__init__.py create mode 100755 cli/meta/roles/applications/bond/__main__.py create mode 100644 cli/meta/roles/applications/bond/cli.py create mode 100644 cli/meta/roles/applications/bond/edit.py create mode 100644 cli/meta/roles/applications/bond/matrix.py create mode 100644 cli/meta/roles/applications/bond/render.py create mode 100644 cli/meta/roles/applications/bond/serve.py create mode 100644 tests/unit/python/cli/meta/roles/applications/bond/__init__.py create mode 100644 tests/unit/python/cli/meta/roles/applications/bond/test_edit.py diff --git a/Makefile b/Makefile index ac0a913716..6d6a58a751 100644 --- a/Makefile +++ b/Makefile @@ -47,6 +47,11 @@ autoformat: install-lint autoformat-restage: @bash scripts/git/autoformat_restage.sh "$(MAKE)" autoformat +.PHONY: bond +# Serve the role bond matrix, where editing a cell rewrites the role's bond. +bond: + @"$${PYTHON}" -m cli.meta.roles.applications.bond $(args) + .PHONY: bootstrap # Install dependencies and prepare the project. bootstrap: install setup diff --git a/aliases b/aliases index 29a1ce3d4c..4266c40977 100644 --- a/aliases +++ b/aliases @@ -1,5 +1,6 @@ alias i8='infinito' # the infinito CLI alias i8bo='m bootstrap' +alias i8bond='m bond' alias i8bu='m build' alias i8buce='m build-cleanup' alias i8bude='m build-dependency' diff --git a/cli/meta/roles/applications/bond/__init__.py b/cli/meta/roles/applications/bond/__init__.py new file mode 100644 index 0000000000..85993a0f11 --- /dev/null +++ b/cli/meta/roles/applications/bond/__init__.py @@ -0,0 +1 @@ +"""Render the bond matrix: which roles couple to which, and how tightly.""" diff --git a/cli/meta/roles/applications/bond/__main__.py b/cli/meta/roles/applications/bond/__main__.py new file mode 100755 index 0000000000..a253a81cfb --- /dev/null +++ b/cli/meta/roles/applications/bond/__main__.py @@ -0,0 +1,7 @@ +#!/usr/bin/env python3 +"""Serve the bond matrix of every role that declares one.""" + +from .cli import main + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/cli/meta/roles/applications/bond/cli.py b/cli/meta/roles/applications/bond/cli.py new file mode 100644 index 0000000000..b349b3d110 --- /dev/null +++ b/cli/meta/roles/applications/bond/cli.py @@ -0,0 +1,97 @@ +"""Serve the editable bond matrix, or export it as a static file.""" + +from __future__ import annotations + +import argparse +import contextlib +import shutil +import subprocess +import sys +from pathlib import Path + +from utils import PROJECT_ROOT + +from .matrix import collect_edges, participants +from .render import render, render_json +from .serve import serve + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="bond", + description=( + "Show which roles bond to which, and how tightly. Serves an editable " + "matrix on localhost; a cell edit rewrites the bond in the role." + ), + ) + parser.add_argument( + "--roles-dir", + default=str(PROJECT_ROOT / "roles"), + help="roles directory to read and write (default: the project's own)", + ) + parser.add_argument( + "-o", + "--out", + default="", + help="export a read-only file here instead of serving", + ) + parser.add_argument( + "--format", + choices=("html", "json"), + default="html", + help="export format, json implies --out (default: html)", + ) + parser.add_argument( + "--port", + type=int, + default=0, + help="port to serve on (default: any free port)", + ) + parser.add_argument( + "--no-open", + action="store_true", + help="do not open a browser", + ) + return parser + + +def _open(target: str) -> None: + """Open the matrix with the desktop's handler, if there is one.""" + opener = shutil.which("xdg-open") or shutil.which("open") + if not opener: + print(f"no xdg-open/open on PATH; the matrix is at {target}", file=sys.stderr) + return + subprocess.run([opener, target], check=False) + + +def main(argv: list[str] | None = None) -> int: + args = build_parser().parse_args(argv) + roles_dir = Path(args.roles_dir) + if not roles_dir.is_dir(): + print(f"no such roles directory: {roles_dir}", file=sys.stderr) + return 2 + + edges = collect_edges(roles_dir) + if not edges: + print(f"no services..bond found under {roles_dir}", file=sys.stderr) + return 1 + + if args.format == "json": + payload = render_json(edges) + if not args.out: + print(payload) + return 0 + elif args.out: + payload = render(edges, participants(edges)) + else: + httpd, url = serve(roles_dir, edges, args.port) + print(f"{url} ({len(edges)} bonds, editing writes to {roles_dir})") + if not args.no_open: + _open(url) + with contextlib.suppress(KeyboardInterrupt), httpd: + httpd.serve_forever() + return 0 + + Path(args.out).write_text(payload, encoding="utf-8") + print(args.out) + return 0 diff --git a/cli/meta/roles/applications/bond/edit.py b/cli/meta/roles/applications/bond/edit.py new file mode 100644 index 0000000000..808fda52a7 --- /dev/null +++ b/cli/meta/roles/applications/bond/edit.py @@ -0,0 +1,103 @@ +"""Write a single ``bond`` scalar back into a role's ``meta/services.yml``. + +The file is edited line by line rather than parsed and re-dumped: these files +carry ``# nocheck:`` directives and hand-tuned ordering that a YAML round-trip +would drop. +""" + +from __future__ import annotations + +import re +from typing import TYPE_CHECKING + +from utils.roles.mapping import ROLE_FILE_META_SERVICES + +if TYPE_CHECKING: + from pathlib import Path + +_TOP_KEY = re.compile(r"^(?P[^\s#][^:]*):\s*(?:#.*)?$") +_BOND = re.compile(r"^(?P\s*bond:\s*)(?P[^#]*?)(?P\s*#.*)?$") + + +class EditError(Exception): + """Raised when the requested bond cannot be written.""" + + +def parse_bond(raw: str) -> float | None: + """Return the bond a user typed, or None when they cleared the field.""" + text = str(raw).strip() + if text in ("", "-", "·"): + return None + try: + value = float(text) + except ValueError as exc: + raise EditError(f"not a number: {raw!r}") from exc + if not 0.0 <= value <= 1.0: + raise EditError(f"bond must be between 0 and 1, got {value:g}") + return value + + +def _block(lines: list[str], key: str) -> tuple[int, int]: + """Return the half-open line range of the top-level ``key`` mapping.""" + start = -1 + for index, line in enumerate(lines): + match = _TOP_KEY.match(line) + if match is None: + continue + if start < 0: + if match.group("key").strip() == key: + start = index + continue + return start, index + if start < 0: + raise EditError(f"no top-level entry {key!r}") + return start, len(lines) + + +def set_bond(roles_dir: Path, role: str, service_key: str, bond: float | None) -> str: + """Set, or with ``bond=None`` remove, ``.bond`` in ``role``. + + Args: + roles_dir: directory holding the role directories. + role: the consumer role whose file declares the bond. + service_key: the top-level key inside that role's services file. + bond: the new value, or None to drop the declaration entirely. + + Returns: + The value as it now stands in the file, or "" when it was removed. + """ + path = roles_dir / role / ROLE_FILE_META_SERVICES + if not path.is_file(): + raise EditError(f"no such file: {path}") + + raw = path.read_text(encoding="utf-8") # nocheck: cache-read — a cached read would serve the text from before the previous edit and silently revert it + lines = raw.splitlines(keepends=True) + start, end = _block(lines, service_key) + + target = next( + (i for i in range(start + 1, end) if _BOND.match(lines[i].rstrip("\n"))), -1 + ) + if bond is None: + if target < 0: + return "" + del lines[target] + else: + text = f"{bond:g}" + if target < 0: + indent = next( + ( + line[: len(line) - len(line.lstrip())] + for line in lines[start + 1 : end] + if line.strip() and not line.lstrip().startswith("#") + ), + " ", + ) + lines.insert(start + 1, f"{indent}bond: {text}\n") + else: + match = _BOND.match(lines[target].rstrip("\n")) + if match is None: + raise EditError(f"cannot rewrite {lines[target]!r}") + lines[target] = f"{match['lead']}{text}{match['tail'] or ''}\n" + + path.write_text("".join(lines), encoding="utf-8") + return "" if bond is None else f"{bond:g}" diff --git a/cli/meta/roles/applications/bond/matrix.py b/cli/meta/roles/applications/bond/matrix.py new file mode 100644 index 0000000000..a6954e279f --- /dev/null +++ b/cli/meta/roles/applications/bond/matrix.py @@ -0,0 +1,77 @@ +"""Read every ``services..bond`` and resolve it to a role-to-role edge. + +``bond`` states how tightly a role couples to the service it declares: ``>= 1`` +keeps the partner on the same host, ``< 1`` lets it live on its own. The value +sits on the consumer's side of the relation, so the pair is directed and the +two directions of one pair need not agree. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +from utils.cache.yaml import load_yaml_any +from utils.roles.applications.services.registry import ( + build_service_registry_from_roles_dir, +) +from utils.roles.mapping import ROLE_FILE_META_SERVICES + +if TYPE_CHECKING: + from pathlib import Path + +DEFAULT_BOND = 1.0 + + +def _as_bond(value: Any) -> float | None: + """Return the bond as a float, or None when the entry declares none.""" + if isinstance(value, bool) or value is None: + return None + try: + return float(str(value).strip()) + except (TypeError, ValueError): + return None + + +def collect_edges(roles_dir: Path) -> dict[tuple[str, str], dict[str, Any]]: + """Return ``{(consumer, provider): {bond, service_key, enabled}}``. + + A role's entry for its own entity is skipped: a role does not bond to + itself, and those entries carry unrelated topics such as users or domains. + """ + registry = build_service_registry_from_roles_dir(roles_dir) + provider_of = { + key: str((entry or {}).get("role") or "") + for key, entry in registry.items() + if isinstance(entry, dict) + } + + edges: dict[tuple[str, str], dict[str, Any]] = {} + for path in sorted(roles_dir.glob(f"*/{ROLE_FILE_META_SERVICES}")): + consumer = path.parent.parent.name + services = load_yaml_any(str(path), default_if_missing={}) or {} + if not isinstance(services, dict): + continue + for key, entry in services.items(): + if not isinstance(entry, dict): + continue + bond = _as_bond(entry.get("bond")) + if bond is None: + continue + provider = provider_of.get(str(key), "") + if not provider or provider == consumer: + continue + edges[(consumer, provider)] = { + "bond": bond, + "service_key": str(key), + "enabled": entry.get("enabled"), + } + return edges + + +def participants(edges: dict[tuple[str, str], dict[str, Any]]) -> list[str]: + """Return every role on either end of a bond, in stable order.""" + seen: set[str] = set() + for consumer, provider in edges: + seen.add(consumer) + seen.add(provider) + return sorted(seen) diff --git a/cli/meta/roles/applications/bond/render.py b/cli/meta/roles/applications/bond/render.py new file mode 100644 index 0000000000..52b4474b5a --- /dev/null +++ b/cli/meta/roles/applications/bond/render.py @@ -0,0 +1,218 @@ +"""Render the bond matrix as an HTML page, optionally editable. + +Every cell carries both directions of one pair, row-to-column above +column-to-row, because a bond is declared on the consumer's side and the two +need not agree. Brightness is the bond: black for none, white for 1. +""" + +from __future__ import annotations + +import html +import json +from typing import Any + +_STYLE = """ +:root { color-scheme: dark; --bg:#0b0c0e; --fg:#d8d8d8; --grid:#ffffff1c; } +* { box-sizing: border-box; } +body { margin:0; background:var(--bg); color:var(--fg); + font:12px/1.35 ui-sans-serif,system-ui,-apple-system,"Segoe UI",sans-serif; } +header { position:sticky; top:0; z-index:5; background:var(--bg); + border-bottom:1px solid var(--grid); padding:10px 14px; } +h1 { margin:0 0 4px; font-size:15px; font-weight:600; } +.meta { opacity:.65; } +.legend { margin-top:7px; display:flex; gap:14px; flex-wrap:wrap; align-items:center; } +.ramp { display:inline-block; width:120px; height:11px; border:1px solid var(--grid); + background:linear-gradient(to right,#000,#fff); vertical-align:-2px; margin:0 5px; } +#q { margin-top:8px; padding:4px 7px; width:260px; border:1px solid var(--grid); + border-radius:4px; background:transparent; color:inherit; font:inherit; } +#msg { margin-left:10px; font-size:11px; opacity:0; transition:opacity .15s; } +#msg.on { opacity:1; } #msg.bad { color:#f77; } +.wrap { overflow:auto; max-height:calc(100vh - 130px); } +table { border-collapse:separate; border-spacing:0; } +th, td { border-right:1px solid var(--grid); border-bottom:1px solid var(--grid); } +thead th { position:sticky; top:0; z-index:3; background:var(--bg); height:132px; + vertical-align:bottom; padding:0; } +thead th div { writing-mode:vertical-rl; transform:rotate(180deg); + padding:6px 3px; white-space:nowrap; font-weight:500; } +tbody th { position:sticky; left:0; z-index:2; background:var(--bg); text-align:left; + padding:0 8px 0 6px; white-space:nowrap; font-weight:500; } +thead th:first-child { position:sticky; left:0; z-index:4; background:var(--bg); } +td { width:30px; min-width:30px; padding:0; } +.b, .nb { display:block; height:13px; text-align:center; font-size:9px; line-height:13px; + font-variant-numeric:tabular-nums; } +.nb { background:#000; } +.b[data-c] { cursor:pointer; } +.b[data-c]:hover { outline:1px solid #4af; outline-offset:-1px; } +.diag { background:repeating-linear-gradient(45deg,#000,#000 3px,#1b1d22 3px,#1b1d22 6px); } +#ed { position:fixed; z-index:9; width:44px; height:22px; padding:0 3px; text-align:center; + border:1px solid #4af; border-radius:3px; background:#000; color:#fff; + font:11px ui-monospace,monospace; display:none; } +.hidden { display:none; } +""" + +_SCRIPT = """ +const box = document.getElementById('q'); +const rows = [...document.querySelectorAll('tbody tr')]; +const heads = [...document.querySelectorAll('thead th[data-role]')]; +box.addEventListener('input', () => { + const q = box.value.trim().toLowerCase(); + rows.forEach(tr => tr.classList.toggle('hidden', !!q && !tr.dataset.role.includes(q))); + heads.forEach((th, i) => { + const hit = !q || th.dataset.role.includes(q) || + rows.some(tr => !tr.classList.contains('hidden') && + tr.children[i + 1].dataset.filled === '1'); + th.classList.toggle('hidden', !hit); + rows.forEach(tr => tr.children[i + 1].classList.toggle('hidden', !hit)); + }); +}); + +const ed = document.getElementById('ed'); +const msg = document.getElementById('msg'); +let cur = null; + +function say(text, bad) { + msg.textContent = text; + msg.className = 'on' + (bad ? ' bad' : ''); + setTimeout(() => { msg.className = ''; }, 2600); +} + +function close() { ed.style.display = 'none'; cur = null; } + +document.querySelector('tbody').addEventListener('click', e => { + const bar = e.target.closest('.b[data-c]'); + if (!bar) { close(); return; } + cur = bar; + const r = bar.getBoundingClientRect(); + ed.style.left = Math.round(r.left + r.width / 2 - 22) + 'px'; + ed.style.top = Math.round(r.top + r.height / 2 - 11) + 'px'; + ed.style.display = 'block'; + ed.value = bar.dataset.b || ''; + ed.select(); +}); + +ed.addEventListener('keydown', e => { + if (e.key === 'Escape') { close(); return; } + if (e.key !== 'Enter') return; + const bar = cur, value = ed.value; + close(); + fetch('bond', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ consumer: bar.dataset.c, provider: bar.dataset.p, value }) + }).then(r => r.json()).then(d => { + if (!d.ok) { say(d.error, true); return; } + bar.textContent = d.text; + bar.dataset.b = d.text; + bar.style.background = d.bg; + bar.style.color = d.fg; + say(d.written); + }).catch(err => say(String(err), true)); +}); +ed.addEventListener('blur', close); +""" + + +def shade(bond: float) -> tuple[str, str]: + """Return the (background, foreground) pair for a bond between 0 and 1.""" + level = round(max(0.0, min(1.0, bond)) * 255) + return f"rgb({level},{level},{level})", "#000" if level > 140 else "#fff" + + +def _bar( + edge: dict[str, Any] | None, consumer: str, provider: str, editable: bool +) -> str: + """Return one of a cell's two direction bars.""" + if edge is None: + return '' + bond = edge["bond"] + background, foreground = shade(bond) + text = f"{bond:g}" + handle = "" + if editable: + handle = ( + f' data-c="{html.escape(consumer)}" data-p="{html.escape(provider)}"' + f' data-b="{text}"' + ) + return ( + f'' + f"{text}" + ) + + +def render( + edges: dict[tuple[str, str], dict[str, Any]], + roles: list[str], + *, + editable: bool = False, +) -> str: + """Return the full HTML page for the given edges. + + Args: + edges: the collected ``{(consumer, provider): edge}`` mapping. + roles: every participating role, in the order rows and columns take. + editable: wire the cells to the server's write endpoint. + """ + head = "".join( + f'
{html.escape(r)}
' + for r in roles + ) + body = [] + for row in roles: + cells = [] + for col in roles: + if row == col: + cells.append('') + continue + out, back = edges.get((row, col)), edges.get((col, row)) + filled = "1" if out is not None or back is not None else "0" + cells.append( + f'' + + _bar(out, row, col, editable) + + _bar(back, col, row, editable) + + "" + ) + body.append( + f'' + f"{html.escape(row)}" + "".join(cells) + "" + ) + + hint = ( + "click a bar to edit, Enter writes it to the role" + if editable + else "read only, run without --out to edit" + ) + return f""" + + +Bond matrix + +
+

Bond matrix

+
{len(edges)} bonds across {len(roles)} roles · + row → column above column → row · {hint}
+
+ 01 + no bond +
+ +
+
+{head} +{"".join(body)} +
+ + + +""" + + +def render_json(edges: dict[tuple[str, str], dict[str, Any]]) -> str: + """Return the same data as JSON, for a caller that wants to post-process.""" + return json.dumps( + [ + {"consumer": consumer, "provider": provider, **edge} + for (consumer, provider), edge in sorted(edges.items()) + ], + indent=2, + ) diff --git a/cli/meta/roles/applications/bond/serve.py b/cli/meta/roles/applications/bond/serve.py new file mode 100644 index 0000000000..dd08ae9db0 --- /dev/null +++ b/cli/meta/roles/applications/bond/serve.py @@ -0,0 +1,111 @@ +"""Serve the bond matrix on localhost and write every cell edit back to disk. + +The edge map is built once at startup and kept in step with the writes this +server performs, so a role file edited by hand elsewhere needs a restart. +""" + +from __future__ import annotations + +import json +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import TYPE_CHECKING, Any + +from .edit import EditError, parse_bond, set_bond +from .matrix import participants +from .render import render, shade + +if TYPE_CHECKING: + from pathlib import Path + +MAX_BODY = 4096 + + +class _Handler(BaseHTTPRequestHandler): + server_version = "bond-matrix" + roles_dir: Path + edges: dict[tuple[str, str], dict[str, Any]] + + def log_message(self, fmt: str, *args: Any) -> None: + return + + def _send(self, code: int, body: bytes, content_type: str) -> None: + self.send_response(code) + self.send_header("Content-Type", content_type) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def _json(self, code: int, payload: dict[str, Any]) -> None: + self._send(code, json.dumps(payload).encode(), "application/json") + + def do_GET(self) -> None: + if self.path.split("?")[0] not in ("/", "/index.html"): + self._json(404, {"ok": False, "error": "not found"}) + return + page = render(self.edges, participants(self.edges), editable=True) + self._send(200, page.encode(), "text/html; charset=utf-8") + + def do_POST(self) -> None: + if self.path.split("?")[0] != "/bond": + self._json(404, {"ok": False, "error": "not found"}) + return + length = int(self.headers.get("Content-Length") or 0) + if length > MAX_BODY: + self._json(413, {"ok": False, "error": "body too large"}) + return + try: + payload = json.loads(self.rfile.read(length) or b"{}") + key = (str(payload["consumer"]), str(payload["provider"])) + raw = payload["value"] + except (ValueError, KeyError, TypeError): + self._json(400, {"ok": False, "error": "malformed request"}) + return + + edge = self.edges.get(key) + if edge is None: + self._json(404, {"ok": False, "error": f"no bond {key[0]} -> {key[1]}"}) + return + + try: + bond = parse_bond(raw) + text = set_bond(self.roles_dir, key[0], edge["service_key"], bond) + except EditError as exc: + self._json(400, {"ok": False, "error": str(exc)}) + return + + if bond is None: + del self.edges[key] + background, foreground = "#000", "#fff" + else: + edge["bond"] = bond + background, foreground = shade(bond) + written = f"{key[0]}/{edge['service_key']}.bond = {text or 'removed'}" + self._json( + 200, + { + "ok": True, + "text": text, + "bg": background, + "fg": foreground, + "written": written, + }, + ) + + +def serve( + roles_dir: Path, + edges: dict[tuple[str, str], dict[str, Any]], + port: int, +) -> tuple[ThreadingHTTPServer, str]: + """Return a server bound to loopback and the URL it listens on. + + Args: + roles_dir: directory the writes land in. + edges: the collected edge map, mutated in place as edits arrive. + port: the port to bind, or 0 to let the OS pick one. + """ + handler = type( + "_BoundHandler", (_Handler,), {"roles_dir": roles_dir, "edges": edges} + ) + httpd = ThreadingHTTPServer(("127.0.0.1", port), handler) + return httpd, f"http://127.0.0.1:{httpd.server_address[1]}/" diff --git a/tests/unit/python/cli/meta/roles/applications/bond/__init__.py b/tests/unit/python/cli/meta/roles/applications/bond/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/unit/python/cli/meta/roles/applications/bond/test_edit.py b/tests/unit/python/cli/meta/roles/applications/bond/test_edit.py new file mode 100644 index 0000000000..ce7e178e60 --- /dev/null +++ b/tests/unit/python/cli/meta/roles/applications/bond/test_edit.py @@ -0,0 +1,76 @@ +import unittest +from pathlib import Path +from tempfile import TemporaryDirectory + +from cli.meta.roles.applications.bond.edit import EditError, parse_bond, set_bond +from utils.roles.mapping import ROLE_FILE_META_SERVICES + +SOURCE = """--- +litellm: + bond: 0.5 + enabled: "{{ 'svc-ai-litellm' in group_names }}" +# nocheck: dockerfile-custom +sso: + bond: 1 # nocheck: dynamic-flag + flavor: oidc +ldap: + enabled: true +""" + + +class TestSetBond(unittest.TestCase): + def setUp(self) -> None: + self._tmp = TemporaryDirectory() + self.roles = Path(self._tmp.name) + self.path = self.roles / "web-app-demo" / ROLE_FILE_META_SERVICES + self.path.parent.mkdir(parents=True) + self.path.write_text(SOURCE, encoding="utf-8") + self.addCleanup(self._tmp.cleanup) + + def read(self) -> str: + return self.path.read_text(encoding="utf-8") # nocheck: cache-read — re-reads the file set_bond just rewrote + + def test_rewrites_the_value_in_place(self) -> None: + self.assertEqual(set_bond(self.roles, "web-app-demo", "litellm", 0.25), "0.25") + self.assertIn(" bond: 0.25\n", self.read()) + self.assertIn("# nocheck: dockerfile-custom\n", self.read()) + + def test_keeps_a_trailing_comment(self) -> None: + set_bond(self.roles, "web-app-demo", "sso", 0.5) + self.assertIn(" bond: 0.5 # nocheck: dynamic-flag\n", self.read()) + + def test_inserts_a_missing_bond(self) -> None: + set_bond(self.roles, "web-app-demo", "ldap", 1) + self.assertIn("ldap:\n bond: 1\n enabled: true\n", self.read()) + + def test_removes_on_none(self) -> None: + self.assertEqual(set_bond(self.roles, "web-app-demo", "litellm", None), "") + self.assertNotIn("bond: 0.5", self.read()) + self.assertIn("litellm:\n enabled:", self.read()) + + def test_rejects_an_unknown_key(self) -> None: + with self.assertRaises(EditError): + set_bond(self.roles, "web-app-demo", "nope", 1) + + def test_does_not_match_a_nested_key(self) -> None: + with self.assertRaises(EditError): + set_bond(self.roles, "web-app-demo", "flavor", 1) + + +class TestParseBond(unittest.TestCase): + def test_blank_clears(self) -> None: + self.assertIsNone(parse_bond("")) + self.assertIsNone(parse_bond(" - ")) + + def test_rejects_out_of_range(self) -> None: + for raw in ("2", "-0.1", "abc"): + with self.assertRaises(EditError): + parse_bond(raw) + + def test_accepts_the_edges(self) -> None: + self.assertEqual(parse_bond("0"), 0.0) + self.assertEqual(parse_bond("1"), 1.0) + + +if __name__ == "__main__": + unittest.main()