#!/usr/bin/env python3 """ServiceDefinition renderer (task T-0.3) — the platform's core GitOps contract. Consumes ONE ServiceDefinition (schemas/servicedefinition.schema.json, catalog/*.yaml) and emits the concrete manifest set the renderer contract specifies (schemas/renderer-contract.md): Namespace + default-deny baseline, Deployment/Service or KubeVirt VirtualMachine + CDI DataVolume, PVCs, per-dependency CiliumNetworkPolicy allows, egress-class steering label, HTTPRoute(s), ServiceMonitor/alerts, Flux image-automation, and the auth-mode / backup wiring hints. Requirements: R-GITOPS-02 (single reviewable object wires every requirement), R-NET-01/02 (default-deny + explicit allows), R-NET-05 (egress class), R-STATE-01/02/04 (storage), R-TLS/R-DNS (ingressRoutes/exposure), R-OBS-01 (observability), R-UPD-01/02/03 (update), R-IDP-02/06 (auth), R-PLAT-02/03 (kind/source). Design (documented per the task's "if the contract is ambiguous, choose the simplest reproducible option"): a deterministic Python renderer. The contract does not mandate a mechanism (script / kustomize generator / templates); a render-then-consume script is the simplest reproducible option the repo can run in CI and that Flux consumes from a rendered output path. CI invokes it via `render.py ` (or `--all`); the offline L1 render conformance verifier diffs its deterministic output against committed goldens. Determinism: fixed document order + fixed key order + PyYAML with sort_keys=False and a wide line width, so identical input always yields byte-identical output (golden-diffable). Dependency: PyYAML (input parse + output emit). This is the ONE renderer dependency; the conformance harness stays stdlib-only and only shells out to this script + diffs text. """ from __future__ import annotations import argparse import sys from pathlib import Path try: import yaml except ImportError: # pragma: no cover - explicit, actionable failure sys.stderr.write( "render.py requires PyYAML (the renderer's single dependency). " "Install with: pip install pyyaml\n" ) raise SystemExit(2) # --- Label / naming conventions (match platform/kyverno/policies + platform/*/) -------- LBL = "homelab.keithjackson.dev" ZONE = "home.keithjackson.dev" GATEWAY_NAME = "shared" GATEWAY_NS = "gateway" KP_RELEASE = "kube-prometheus-stack" # ServiceMonitor/PrometheusRule discovery label # egress classes and auth modes fixed by docs/03 / docs/13 + the Kyverno enums. EGRESS_CLASSES = ("internet", "vpn") AUTH_MODES = ("oidc", "forward-auth", "native", "none") # ============================================================================= # Defaults (renderer-contract.md "Default posture") — materialize BEFORE mapping. # ============================================================================= def apply_defaults(sd: dict) -> dict: """Return a copy of the ServiceDefinition with the contract's default posture filled. A field absent from a ServiceDefinition renders from its default posture, not "not at all" (renderer-contract.md). """ s = dict(sd) s.setdefault("kind", "container") s.setdefault("exposure", "internal") s.setdefault("egress", "internet") s.setdefault("storage", []) s.setdefault("dependencies", []) s.setdefault("ingressRoutes", []) # auth.mode default posture: forward-auth for any exposed UI, else none (internal, no # UI). An explicit auth block always wins (R-IDP-02). auth = dict(s.get("auth") or {}) if "mode" not in auth: auth["mode"] = "forward-auth" if s["ingressRoutes"] else "none" auth.setdefault("groups", []) s["auth"] = auth update = dict(s.get("update") or {}) update.setdefault("channel", "stable") update.setdefault("autoRollbackOnFailedHealth", True) s["update"] = update obs = dict(s.get("observability") or {}) obs.setdefault("metrics", True) obs.setdefault("slis", []) s["observability"] = obs # storage entry class default is longhorn (replicated tier). s["storage"] = [{**{"class": "longhorn"}, **v} for v in s["storage"]] # dependency protocol default is tcp. s["dependencies"] = [{**{"protocol": "tcp"}, **d} for d in s["dependencies"]] return s # ============================================================================= # Field -> resource mapping (renderer-contract.md "Field -> resource mapping") # ============================================================================= def r_namespace(s: dict) -> dict: """Namespace = the service trust zone / isolation boundary (R-NET, docs/03). Carries app-namespace=true so Kyverno generates the default-deny baseline and the egress-class / auth-mode admission policies apply. gateway-access=shared lets its HTTPRoutes attach to the shared Gateway (only when it exposes routes). """ labels = {f"{LBL}/app-namespace": "true"} if s["ingressRoutes"]: labels["gateway-access"] = "shared" ann = {} backup = (s.get("backup") or {}).get("policy") if backup: ann[f"{LBL}/backup-policy"] = backup # R-STATE-03 off-cluster schedule ref (hint) meta = {"name": s["name"], "labels": labels} if ann: meta["annotations"] = ann return {"apiVersion": "v1", "kind": "Namespace", "metadata": meta} def r_default_deny(s: dict) -> dict: """The default-deny baseline: deny all ingress+egress, allow ONLY DNS egress (R-NET-01). Rendered explicitly (GitOps at namespace creation, renderer-contract §dependencies) — it matches the Kyverno-generated baseline shape (platform/kyverno/.../generated-default -deny.yaml) so the render carries the guarantee even before admission generates it. """ return { "apiVersion": "networking.k8s.io/v1", "kind": "NetworkPolicy", "metadata": { "name": "default-deny", "namespace": s["name"], "labels": {f"{LBL}/baseline": "default-deny"}, }, "spec": { "podSelector": {}, "policyTypes": ["Ingress", "Egress"], "ingress": [], "egress": [ { "to": [ { "namespaceSelector": { "matchLabels": { "kubernetes.io/metadata.name": "kube-system" } } } ], "ports": [ {"protocol": "UDP", "port": 53}, {"protocol": "TCP", "port": 53}, ], } ], }, } def _pod_labels(s: dict) -> dict: """Workload pod labels: identity + the egress-class steering hook + auth-mode. egress-class is what the shared Cilium Egress Gateway selects on to steer vpn-class pods (enforcement is OUTSIDE the workload — renderer-contract §egress; ADR-0015); auth-mode satisfies the Kyverno require-auth-mode-label policy (R-IDP-02). """ return { "app.kubernetes.io/name": s["name"], f"{LBL}/egress-class": s["egress"], f"{LBL}/auth-mode": s["auth"]["mode"], } def _service_ports(s: dict) -> list: ports = sorted({r["port"] for r in s["ingressRoutes"]}) return [{"name": f"port-{p}", "port": p, "targetPort": p} for p in ports] def r_pvcs(s: dict) -> list: """One PVC per storage entry (R-STATE-01/02/04). longhorn -> 3-replica replicated tier; bulk -> external media tier (not 3x replicated). """ out = [] for vol in s["storage"]: cls = vol.get("class", "longhorn") storage_class = "longhorn" if cls == "longhorn" else "bulk" spec = { "accessModes": ["ReadWriteOnce"], "storageClassName": storage_class, } if vol.get("size"): spec["resources"] = {"requests": {"storage": vol["size"]}} out.append( { "apiVersion": "v1", "kind": "PersistentVolumeClaim", "metadata": { "name": f"{s['name']}-{vol['name']}", "namespace": s["name"], "labels": {"app.kubernetes.io/name": s["name"]}, }, "spec": spec, } ) return out def r_deployment(s: dict) -> dict: """Deployment for a container workload (R-PLAT-02).""" pod_labels = _pod_labels(s) container = { "name": s["name"], "image": s["source"]["image"], } ports = _service_ports(s) if ports: container["ports"] = [{"containerPort": p["port"]} for p in ports] volume_mounts, volumes = [], [] for vol in s["storage"]: volume_mounts.append({"name": vol["name"], "mountPath": f"/data/{vol['name']}"}) volumes.append( { "name": vol["name"], "persistentVolumeClaim": {"claimName": f"{s['name']}-{vol['name']}"}, } ) if volume_mounts: container["volumeMounts"] = volume_mounts pod_spec = {"containers": [container]} if volumes: pod_spec["volumes"] = volumes ann = _auth_annotations(s) tmpl_meta = {"labels": pod_labels} if ann: tmpl_meta["annotations"] = ann return { "apiVersion": "apps/v1", "kind": "Deployment", "metadata": { "name": s["name"], "namespace": s["name"], "labels": {"app.kubernetes.io/name": s["name"]}, }, "spec": { "replicas": 1, "selector": {"matchLabels": {"app.kubernetes.io/name": s["name"]}}, "template": {"metadata": tmpl_meta, "spec": pod_spec}, }, } def r_service(s: dict): ports = _service_ports(s) if not ports: return None return { "apiVersion": "v1", "kind": "Service", "metadata": { "name": s["name"], "namespace": s["name"], "labels": {"app.kubernetes.io/name": s["name"]}, }, "spec": { "selector": {"app.kubernetes.io/name": s["name"]}, "ports": ports, }, } def r_dependency_policies(s: dict) -> list: """One CiliumNetworkPolicy egress allow per dependency, on top of default-deny (R-NET-02). Everything undeclared stays denied; empty dependencies => reaches only DNS. """ out = [] for dep in s["dependencies"]: proto = dep.get("protocol", "tcp").upper() out.append( { "apiVersion": "cilium.io/v2", "kind": "CiliumNetworkPolicy", "metadata": { "name": f"{s['name']}-allow-{dep['to']}-{dep['port']}", "namespace": s["name"], "labels": {"app.kubernetes.io/name": s["name"]}, }, "spec": { "endpointSelector": { "matchLabels": {"app.kubernetes.io/name": s["name"]} }, "egress": [ { "toEndpoints": [ {"matchLabels": {"app.kubernetes.io/name": dep["to"]}} ], "toPorts": [ { "ports": [ {"port": str(dep["port"]), "protocol": proto} ] } ], } ], }, } ) return out def _route_meta(s: dict, route: dict) -> dict: meta = { "name": f"{s['name']}-{route['port']}", "namespace": s["name"], "labels": {"app.kubernetes.io/name": s["name"]}, } ann = _auth_annotations(s) if ann: meta["annotations"] = ann return meta def r_httproutes(s: dict) -> list: """One HTTPRoute per ingressRoute, attaching host -> service port on the shared Gateway. TLS is the Gateway's wildcard *.home.keithjackson.dev cert (no per-service cert step); ExternalDNS manages the record from the HTTPRoute (R-TLS/R-DNS, renderer-contract). """ out = [] for route in s["ingressRoutes"]: out.append( { "apiVersion": "gateway.networking.k8s.io/v1", "kind": "HTTPRoute", "metadata": _route_meta(s, route), "spec": { "parentRefs": [ { "name": GATEWAY_NAME, "namespace": GATEWAY_NS, "sectionName": "https", } ], "hostnames": [route["host"]], "rules": [ { "matches": [ {"path": {"type": "PathPrefix", "value": "/"}} ], "backendRefs": [ {"name": s["name"], "port": route["port"]} ], } ], }, } ) return out def _auth_annotations(s: dict) -> dict: """Auth wiring hints (renderer-contract §auth): the platform (oauth2-proxy/Keycloak) consumes these to protect ingressRoutes. oidc/forward-auth carry the authorized groups (R-IDP-06: access is 'in an authorized group', not merely 'logged in').""" mode = s["auth"]["mode"] if mode in ("oidc", "forward-auth"): groups = ",".join(s["auth"].get("groups", [])) return { f"{LBL}/auth-mode": mode, f"{LBL}/auth-groups": groups, } return {} def r_servicemonitor(s: dict): if not s["observability"]["metrics"]: return None return { "apiVersion": "monitoring.coreos.com/v1", "kind": "ServiceMonitor", "metadata": { "name": s["name"], "namespace": s["name"], "labels": {"release": KP_RELEASE, "app.kubernetes.io/name": s["name"]}, }, "spec": { "selector": {"matchLabels": {"app.kubernetes.io/name": s["name"]}}, "namespaceSelector": {"matchNames": [s["name"]]}, "endpoints": [{"port": "metrics", "interval": "30s"}], }, } def r_prometheusrule(s: dict): """Each SLI becomes at least one alert rule (R-OBS-01).""" slis = s["observability"]["slis"] if not slis: return None rules = [] for sli in slis: if sli == "up": rules.append( { "alert": f"{s['name']}-down".replace("_", "-"), "expr": f'up{{job="{s["name"]}"}} == 0', "for": "5m", "labels": {"severity": "critical", "service": s["name"]}, "annotations": { "summary": f"{s['name']} is down", "description": f"{s['name']} has been unreachable for 5m (SLI: up).", }, } ) else: rules.append( { "alert": f"{s['name']}-{sli}".replace("_", "-"), "expr": f'absent({sli}{{service="{s["name"]}"}})', "for": "10m", "labels": {"severity": "warning", "service": s["name"]}, "annotations": {"summary": f"{s['name']} SLI {sli} unhealthy"}, } ) return { "apiVersion": "monitoring.coreos.com/v1", "kind": "PrometheusRule", "metadata": { "name": f"{s['name']}-slis", "namespace": s["name"], "labels": {"release": KP_RELEASE, "app.kubernetes.io/name": s["name"]}, }, "spec": {"groups": [{"name": f"{s['name']}.slis", "rules": rules}]}, } def r_image_automation(s: dict) -> list: """container kind -> Flux ImageRepository + ImagePolicy encoding channel/range (R-UPD). Flux image-automation advances the running version on its scan interval; no human/AI edits a tag (R-UPD-02). `pin` freezes automation. vm kind uses in-guest updates instead. """ if s["kind"] != "container": return [] image = s["source"]["image"].split(":", 1)[0] upd = s["update"] interval = "1h0m0s" imagerepo = { "apiVersion": "image.toolkit.fluxcd.io/v1beta2", "kind": "ImageRepository", "metadata": {"name": s["name"], "namespace": s["name"]}, "spec": {"image": image, "interval": interval}, } policy_spec: dict = {"imageRepositoryRef": {"name": s["name"]}} if upd.get("pin"): # Operator freeze escape hatch: pin to a fixed tag, stopping automation. policy_spec["filterTags"] = {"pattern": f"^{upd['pin']}$"} policy_spec["policy"] = {"alphabetical": {"order": "asc"}} elif upd.get("range"): policy_spec["policy"] = {"semver": {"range": upd["range"]}} else: policy_spec["policy"] = {"semver": {"range": ">=0.0.0"}} imagepolicy = { "apiVersion": "image.toolkit.fluxcd.io/v1beta2", "kind": "ImagePolicy", "metadata": { "name": s["name"], "namespace": s["name"], "annotations": { f"{LBL}/update-channel": upd.get("channel", "stable"), f"{LBL}/auto-rollback": str( upd.get("autoRollbackOnFailedHealth", True) ).lower(), }, }, "spec": policy_spec, } return [imagerepo, imagepolicy] # --- VM (kind: vm) path: KubeVirt VirtualMachine + CDI DataVolume ---------------------- def r_datavolume(s: dict) -> dict: """Standalone CDI DataVolume (NOT a dataVolumeTemplate) so the disk OUTLIVES the VM and survives destroy/rebuild (R-STATE-01) — matches the vm-lifecycle fixture shape.""" root = s["storage"][0] if s["storage"] else {"name": "rootdisk", "size": "10Gi"} return { "apiVersion": "cdi.kubevirt.io/v1beta1", "kind": "DataVolume", "metadata": { "name": f"{s['name']}-{root['name']}", "namespace": s["name"], }, "spec": { "source": {"registry": {"url": f"docker://{s['source']['vmTemplate']}"}}, "storage": { "accessModes": ["ReadWriteOnce"], "storageClassName": "longhorn", "resources": {"requests": {"storage": root.get("size", "10Gi")}}, }, }, } VM_CLOUD_INIT = ( "#cloud-config\n" "package_update: true\n" "packages:\n" " - dnf-automatic\n" "write_files:\n" " - path: /etc/dnf/automatic.conf\n" " content: |\n" " [commands]\n" " upgrade_type = default\n" " random_sleep = 0\n" " download_updates = yes\n" " apply_updates = yes\n" "runcmd:\n" " - [ systemctl, enable, --now, dnf-automatic.timer ]\n" ) def r_virtualmachine(s: dict) -> dict: """KubeVirt VirtualMachine; egress-class propagates to the virt-launcher pod so the shared Cilium egress steering + Kyverno policy see the class (R-PLAT-03; ADR-0015). In-guest dnf-automatic delivers auto-updates with no human step (R-UPD-01, docs/11).""" root = s["storage"][0] if s["storage"] else {"name": "rootdisk"} pod_labels = { "kubevirt.io/vm": s["name"], f"{LBL}/egress-class": s["egress"], f"{LBL}/auth-mode": s["auth"]["mode"], } return { "apiVersion": "kubevirt.io/v1", "kind": "VirtualMachine", "metadata": { "name": s["name"], "namespace": s["name"], "labels": {"app.kubernetes.io/name": s["name"]}, }, "spec": { "running": False, "template": { "metadata": {"labels": pod_labels}, "spec": { "domain": { "cpu": {"cores": 1}, "memory": {"guest": "1536Mi"}, "devices": { "disks": [ {"name": "rootdisk", "disk": {"bus": "virtio"}}, {"name": "cloudinit", "disk": {"bus": "virtio"}}, ], "interfaces": [{"name": "default", "masquerade": {}}], }, "resources": {"requests": {"memory": "1536Mi"}}, }, "networks": [{"name": "default", "pod": {}}], "volumes": [ { "name": "rootdisk", "dataVolume": {"name": f"{s['name']}-{root['name']}"}, }, { "name": "cloudinit", "cloudInitNoCloud": {"userData": VM_CLOUD_INIT}, }, ], }, }, }, } # ============================================================================= # Top-level render + emit # ============================================================================= def render(sd: dict, image_automation: bool = True) -> list: """Return an ordered list of (header_comment, resource_dict) for a ServiceDefinition. image_automation=False omits the Flux ImageRepository/ImagePolicy CRs — used when onboarding apps before the Flux image-reflector/-automation controllers are installed (their CRDs would otherwise be missing and the apps Kustomization would fail). The L1 render goldens use the default (True), so their expected output is unchanged. """ s = apply_defaults(sd) if s["kind"] not in ("container", "vm"): raise ValueError(f"unknown kind: {s['kind']!r}") if s["egress"] not in EGRESS_CLASSES: raise ValueError(f"invalid egress class: {s['egress']!r}") if s["auth"]["mode"] not in AUTH_MODES: raise ValueError(f"invalid auth.mode: {s['auth']['mode']!r}") docs: list = [] docs.append(("Namespace (trust zone; default-deny + policies apply)", r_namespace(s))) docs.append(("Default-deny baseline (R-NET-01): deny all, allow only DNS", r_default_deny(s))) if s["kind"] == "container": # Container PVCs. (For a vm, the CDI DataVolume owns/creates the disk PVC, so we # never render a plain PVC for the VM root disk — it would collide with CDI.) for pvc in r_pvcs(s): docs.append((f"PVC {pvc['metadata']['name']} (R-STATE)", pvc)) docs.append(("Deployment (R-PLAT-02)", r_deployment(s))) svc = r_service(s) if svc: docs.append(("Service", svc)) else: docs.append(("CDI DataVolume (standalone; survives rebuild, R-STATE-01)", r_datavolume(s))) docs.append(("KubeVirt VirtualMachine (R-PLAT-03)", r_virtualmachine(s))) for cnp in r_dependency_policies(s): docs.append((f"CiliumNetworkPolicy egress allow (R-NET-02): {cnp['metadata']['name']}", cnp)) for hr in r_httproutes(s): docs.append((f"HTTPRoute {hr['metadata']['name']} (R-TLS/R-DNS)", hr)) sm = r_servicemonitor(s) if sm: docs.append(("ServiceMonitor (R-OBS-01)", sm)) pr = r_prometheusrule(s) if pr: docs.append(("PrometheusRule SLI alerts (R-OBS-01)", pr)) if image_automation: for res in r_image_automation(s): docs.append((f"Flux {res['kind']} (R-UPD)", res)) return docs def emit(docs: list) -> str: """Deterministic multi-document YAML text (golden-diffable).""" parts = [] for comment, resource in docs: body = yaml.dump( resource, sort_keys=False, default_flow_style=False, width=4096, ) parts.append(f"---\n# {comment}\n{body}".rstrip() + "\n") return "".join(parts) def load_sd(path: Path) -> dict: with path.open() as fh: return yaml.safe_load(fh) def render_file(path: Path, image_automation: bool = True) -> str: return emit(render(load_sd(path), image_automation=image_automation)) # ============================================================================= # Self-test (deterministic, no cluster) — `render.py --selftest` # ============================================================================= def _selftest() -> int: failures = [] def check(cond, msg): if not cond: failures.append(msg) # 1) A minimal container SD renders the mandatory contract invariants. sd = {"name": "demo", "source": {"image": "nginx:1"}} docs = render(sd) kinds = [d[1]["kind"] for d in docs] check("Namespace" in kinds, "container: missing Namespace") check("NetworkPolicy" in kinds, "container: missing default-deny NetworkPolicy") check("Deployment" in kinds, "container: missing Deployment") ns = next(d[1] for d in docs if d[1]["kind"] == "Namespace") check( ns["metadata"]["labels"].get(f"{LBL}/app-namespace") == "true", "namespace missing app-namespace=true (Kyverno hook)", ) dep = next(d[1] for d in docs if d[1]["kind"] == "Deployment") plabels = dep["spec"]["template"]["metadata"]["labels"] check( plabels.get(f"{LBL}/egress-class") == "internet", "default egress-class must be internet (R-NET-05)", ) check( plabels.get(f"{LBL}/auth-mode") in AUTH_MODES, "pod must carry a valid auth-mode label (R-IDP-02)", ) ndp = next(d[1] for d in docs if d[1]["kind"] == "NetworkPolicy") check( ndp["spec"]["policyTypes"] == ["Ingress", "Egress"], "default-deny must cover both directions (R-NET-01)", ) check(ndp["spec"]["ingress"] == [], "default-deny must deny all ingress (R-NET-01)") # 2) egress: vpn flips ONLY the egress-class label (R-NET-07). internet_docs = render({"name": "x", "source": {"image": "a:1"}, "egress": "internet"}) vpn_docs = render({"name": "x", "source": {"image": "a:1"}, "egress": "vpn"}) i_pod = next(d[1] for d in internet_docs if d[1]["kind"] == "Deployment") v_pod = next(d[1] for d in vpn_docs if d[1]["kind"] == "Deployment") check( i_pod["spec"]["template"]["metadata"]["labels"][f"{LBL}/egress-class"] == "internet" and v_pod["spec"]["template"]["metadata"]["labels"][f"{LBL}/egress-class"] == "vpn", "egress class must render onto the pod label", ) # Everything except the egress-class label must be identical (R-NET-07). i_txt = emit(internet_docs).replace("egress-class: internet", "EGRESS") v_txt = emit(vpn_docs).replace("egress-class: vpn", "EGRESS") check(i_txt == v_txt, "flipping egress class changed more than the egress path (R-NET-07)") # 3) Determinism: same input twice -> byte-identical output. check(emit(render(sd)) == emit(render(sd)), "render is not deterministic") # 4) dependencies -> one CiliumNetworkPolicy allow each (R-NET-02). dep_docs = render( { "name": "y", "source": {"image": "a:1"}, "dependencies": [{"to": "db", "port": 5432}], } ) cnps = [d[1] for d in dep_docs if d[1]["kind"] == "CiliumNetworkPolicy"] check(len(cnps) == 1, "one dependency must render one CiliumNetworkPolicy allow") # 5) vm kind -> VirtualMachine + standalone DataVolume. vm_docs = render( { "name": "vmx", "kind": "vm", "source": {"vmTemplate": "quay.io/containerdisks/fedora:41"}, "storage": [{"name": "rootdisk", "size": "6Gi", "class": "longhorn"}], } ) vm_kinds = [d[1]["kind"] for d in vm_docs] check("VirtualMachine" in vm_kinds, "vm: missing VirtualMachine") check("DataVolume" in vm_kinds, "vm: missing CDI DataVolume") if failures: for f in failures: sys.stderr.write(f"SELFTEST FAIL: {f}\n") return 1 sys.stderr.write("render.py selftest: PASS\n") return 0 def main(argv=None) -> int: ap = argparse.ArgumentParser(description="Render a ServiceDefinition to manifests.") ap.add_argument("input", nargs="?", help="ServiceDefinition YAML file") ap.add_argument("--all", metavar="DIR", help="Render every *.yaml ServiceDefinition in DIR") ap.add_argument("-o", "--out-dir", help="With --all: write .yaml files into this dir") ap.add_argument("--selftest", action="store_true", help="Run the built-in self-test") ap.add_argument("--no-image-automation", action="store_true", help="Omit Flux ImageRepository/ImagePolicy CRs (for onboarding before the " "image-reflector/-automation controllers are installed)") args = ap.parse_args(argv) if args.selftest: return _selftest() image_automation = not args.no_image_automation if args.all: src = Path(args.all) files = sorted(p for p in src.glob("*.yaml") if p.name != "render.py") for f in files: text = render_file(f, image_automation=image_automation) if args.out_dir: outdir = Path(args.out_dir) outdir.mkdir(parents=True, exist_ok=True) name = load_sd(f)["name"] (outdir / f"{name}.yaml").write_text(text) else: sys.stdout.write(text) return 0 if not args.input: ap.error("provide a ServiceDefinition file, --all DIR, or --selftest") sys.stdout.write(render_file(Path(args.input), image_automation=image_automation)) return 0 if __name__ == "__main__": raise SystemExit(main())