systemd-transient 0.1.3

#!/usr/bin/env python3
"""Generate src/generated.rs from a systemd source tree.

    tools/gen.py path/to/systemd-v261 > src/generated.rs

Ground truth is the server side: the *_set_transient_property / *_set_property
handlers in src/core/dbus-{unit,service,execute,kill,cgroup}.c. Names come from
their streq()/STR_IN_SET() branches plus the prefix-dispatched families
(dependencies, Exec*, Limit*, IO*Max). Signatures come from the first
sd_bus_message_read*/enter_container in each branch, following helper calls.
String properties parsed through a `foo_from_string()` with a matching
`foo_table[]` become Rust enums. Composite signatures need an entry in
tools/types.toml; an unmapped one aborts generation.
"""
import glob, re, sys, tomllib
from pathlib import Path

ROOT = Path(sys.argv[1])
CORE = ROOT / "src/core"
CFG = tomllib.loads((Path(__file__).parent / "types.toml").read_text())
ALL = {p.name: p.read_text() for p in CORE.glob("dbus-*.c")}

def die(msg):
    sys.exit(f"gen.py: {msg}")

def func_body(fn, prefer=None):
    files = ([ALL[prefer]] if prefer in ALL else []) + list(ALL.values())
    for t in files:
        m = re.search(r'\n(?:static )?int ' + re.escape(fn) + r'\(', t)
        if m:
            body = t[m.start():t.index("\n}\n", m.start())]
            return body[body.index("{") + 1:]
    return None

# Helpers generated by macros: signature, and the parser for string enums.
MACRO_SIG, MACRO_PARSE = {}, {}
for t in ALL.values():
    for kind, fn, rest in re.findall(r'BUS_DEFINE_SET_TRANSIENT(\w*)\((\w+),\s*([^)]*)\)', t):
        h = "bus_set_transient_" + fn
        if kind in ("", "_IS_VALID", "_TO_STRING", "_TO_STRING_ALLOC"):
            MACRO_SIG[h] = re.match(r'\s*"(\w+)"', rest).group(1)
        else:
            MACRO_SIG[h] = "s"
            p = re.search(r'(\w+)_from_string', rest)
            if p: MACRO_PARSE[h] = p.group(1)
    for fn in re.findall(r'^BUS_DEFINE_SET_CGROUP_WEIGHT\((\w+),', t, re.M):
        MACRO_SIG["bus_cgroup_set_" + fn] = "t"
    for fn in re.findall(r'^BUS_DEFINE_SET_CGROUP_LIMIT\((\w+),', t, re.M):
        MACRO_SIG["bus_cgroup_set_" + fn] = "t"
        MACRO_SIG["bus_cgroup_set_" + fn + "_scale"] = "u"

READS = [
    (r"enter_container\(message, 'a', \"([^\"]+)\"\)", lambda g: "a" + g),
    (r"enter_container\(message, 'r', \"([^\"]+)\"\)", lambda g: "(" + g + ")"),
    (r"sd_bus_message_read_strv\(message", lambda g: "as"),
    (r"sd_bus_message_read_array\(message, '(\w)'", lambda g: "a" + g),
    (r"sd_bus_message_read\(message, \"([^\"]+)\"", lambda g: g),
    (r"sd_bus_message_read_basic\(message, '(\w)'", lambda g: g),
]
HELPER = re.compile(r'(bus_set_transient_\w+|bus_cgroup_set_\w+|bus_set_\w+|bus_read_\w+)\(')

def infer(snip, src, name, depth=0):
    """Returns (signature, enum_parser) for one handler branch."""
    t = re.search(r"enter_container\(message, 'a', \w+ \? \"([^\"]+)\" : \"([^\"]+)\"", snip)
    if t:
        return "a" + (t.group(1) if name.endswith("Ex") else t.group(2)), None
    best = None
    for pat, f in READS:
        m = re.search(pat, snip)
        if m and (best is None or m.start() < best[0]):
            best = (m.start(), f(m.group(1) if m.groups() else ""))
    parse = next((p for p in re.findall(r'(\w+)_from_string\(', snip) if p in TABLES), None)
    h = HELPER.search(snip)
    if h and (best is None or h.start() < best[0]):
        fn = h.group(1)
        if fn in MACRO_SIG:
            return MACRO_SIG[fn], MACRO_PARSE.get(fn)
        b = func_body(fn, src) if depth < 3 else None
        if b:
            sig, p2 = infer(b, src, name, depth + 1)
            if sig: return sig, parse or p2
    return (best[1] if best else None), parse

# String tables anywhere in the tree: foo_table[] = { "a", "b" } -> enum candidates.
TABLES = {}
for p in glob.glob(str(ROOT / "src/**/*.c"), recursive=True):
    t = open(p, errors="replace").read()
    for name, blk in re.findall(r'static const char\s*\*\s*const (\w+?)(?:_table)?\[[^\]]*\] = \{(.*?)\};', t, re.S):
        vals = re.findall(r'"([^"]*)"', re.sub(r'/\*.*?\*/', '', blk, flags=re.S))
        if vals and all(vals): TABLES.setdefault(name, vals)

BRANCH = re.compile(r'(?:if|else if) \((?:streq\(name, "(\w+)"\)|STR_IN_SET\(name,([^)]*)\))')
FUNCS = [("dbus-unit.c", "bus_unit_set_live_property"), ("dbus-unit.c", "bus_unit_set_transient_property"),
         ("dbus-service.c", "bus_service_set_transient_property"),
         ("dbus-execute.c", "bus_exec_context_set_transient_property"),
         ("dbus-kill.c", "bus_kill_context_set_transient_property"),
         ("dbus-cgroup.c", "bus_cgroup_set_property"), ("dbus-cgroup.c", "bus_cgroup_set_transient_property")]

props, deprecated = {}, set()
for src, fn in FUNCS:
    b = func_body(fn, src) or die(f"{fn} not found in {src}")
    ms = list(BRANCH.finditer(b))
    for i, m in enumerate(ms):
        end = lambda k: ms[i + k].start() if i + k < len(ms) else len(b)
        snip = b[m.end():end(1)]
        names = [m.group(1)] if m.group(1) else re.findall(r'"(\w+)"', m.group(2))
        if re.match(r'\s*\)?\s*\{?\s*r = sd_bus_message_skip', snip):
            deprecated |= set(names)          # accepted and ignored: not worth a variant
            continue
        if re.search(r'return bus_\w+_set_transient_property\(', snip[:300]):
            continue                           # pure dispatch; the real handler defines the key
        for n in names:
            if n in props or n in deprecated: continue
            s, k = infer(snip, src, n), 2
            while not s[0] and i + k <= len(ms):  # nested streq() split the branch
                s, k = infer(b[m.end():end(k)], src, n), k + 1
            props[n] = s

def table(path, name):
    t = (ROOT / "src" / path).read_text()
    return re.findall(r'= "(\w+)"', re.search(re.escape(name) + r'\[[^\]]*\] = \{(.*?)\};', t, re.S).group(1))

# Prefix-dispatched families.
allowed = re.search(r'if \(!IN_SET\(d,(.*?)\)\)', func_body("bus_unit_set_transient_property", "dbus-unit.c"), re.S).group(1)
dep_names = dict(re.findall(r'\[(UNIT_\w+)\]\s*=\s*"(\w+)"', (ROOT / "src/basic/unit-def.c").read_text()))
for d in re.findall(r'UNIT_\w+', allowed):
    props.setdefault(dep_names.get(d) or die(f"no name for {d}"), ("as", None))
for n in table("core/service.c", "service_exec_command_table"):
    props.setdefault(n, ("a(sasb)", None))
for n in table("core/service.c", "service_exec_ex_command_table"):
    props.setdefault(n, ("a(sasas)", None))
for r in table("basic/rlimit-util.c", "rlimit_table"):
    props.setdefault(f"Limit{r}", ("t", None))
    props.setdefault(f"Limit{r}Soft", ("t", None))
for n in table("basic/cgroup-util.c", "cgroup_io_limit_type_table"):
    props.setdefault(n, ("a(st)", None))

for n in CFG.get("skip", []): props.pop(n, None)
missing = [n for n, (s, _) in props.items() if not s]
if missing: die(f"no signature inferred for {missing}")

# ---- Rust type resolution ----
camel = lambda s: "".join(w[:1].upper() + w[1:] for w in re.split(r'[^0-9A-Za-z]+', s) if w)
SCALAR = {"b": "bool", "s": "String", "t": "u64", "u": "u32", "i": "i32", "q": "u16",
          "as": "Vec<String>", "ay": "Vec<u8>", "aay": "Vec<Vec<u8>>", "h": "Fd<'static>"}
enums = {}  # rust name -> values

def enum_for(parser):
    if parser not in TABLES: die(f"no string table {parser}_table")
    rust = CFG.get("enum_names", {}).get(parser, camel(parser))
    enums[rust] = TABLES[parser]
    return rust

for parser in CFG.get("extra_enums", []):
    enum_for(parser)

resolved = {}
for n, (sig, parser) in sorted(props.items()):
    if n in CFG.get("types", {}):
        ty = CFG["types"][n]
    elif n in CFG.get("enum_props", {}):
        ty = enum_for(CFG["enum_props"][n])
    elif sig == "s" and parser in TABLES and parser not in CFG.get("not_enums", []):
        ty = enum_for(parser)
    elif sig == "t" and n.endswith("USec"):
        ty = "Usec"
    elif sig in SCALAR:
        ty = SCALAR[sig]
    else:
        die(f"{n}: composite signature {sig} has no entry in types.toml")
    resolved[n] = (ty, sig)

def variant(v):
    c = camel(v) or die(f"empty enum value")
    if c[0].isdigit(): c = "V" + c
    return {"Self": "Self_"}.get(c, c)

out = ["// @generated by tools/gen.py from systemd " + CFG["systemd_version"] + ". Do not edit.", "",
       "use crate::props::*;", ""]
for rust, vals in sorted(enums.items()):
    seen = set()
    vs = []
    for v in vals:
        k = variant(v)
        if k in seen: die(f"{rust}: duplicate variant {k}")
        seen.add(k); vs.append(f"{k} => {v!r}".replace("'", '"'))
    out.append(f"str_enum!({rust} {{ {', '.join(vs)} }});")
out += ["", "unit_properties! {"]
for n, (ty, sig) in resolved.items():
    out.append(f"    {n}: {ty} = \"{sig}\",")
out.append("}")
print("\n".join(out))
print(f"gen.py: {len(resolved)} properties, {len(enums)} enums, {len(deprecated)} deprecated keys skipped",
      file=sys.stderr)