diff --git a/README.md b/README.md index 0c14d207c..bcff36b7f 100644 --- a/README.md +++ b/README.md @@ -343,6 +343,7 @@ To remove graphify from all platforms at once: `graphify uninstall` (add `--purg | Terraform / HCL | `.tf .tfvars .hcl` (requires `uv tool install graphifyy[terraform]`) | | OCaml | `.ml .mli` (requires `uv tool install graphifyy[ocaml]`) | | Common Lisp | `.lisp .cl .lsp .asd` (requires `uv tool install graphifyy[commonlisp]`) | +| systemd units | `.service .timer .socket .target .path .mount .slice` (INI, no grammar) — a timer/socket/path `activates` its unit, a service `runs` the script named by `ExecStart=`, `Documentation=file://` becomes `documented_by`, and `After=`/`Wants=`/`Requires=`/`WantedBy=`… become unit→unit edges (only to units in the same directory, so the host's `network.target` is never fabricated) | | MCP configs | `.mcp.json` `mcp.json` `mcp_servers.json` `claude_desktop_config.json` — extracts server nodes, package refs, env var requirements | | Package manifests | `apm.yml` `pyproject.toml` `go.mod` `pom.xml` — one canonical package node per package (by name) plus `depends_on` edges, so a package referenced from many manifests is a single hub | | Docs | `.md .mdx .qmd .html .txt .rst .yaml .yml` (markdown `[text](./other.md)` links and `[[wikilinks]]` become `references` edges between docs) | diff --git a/graphify/detect.py b/graphify/detect.py index 3668eb6fc..29080d694 100644 --- a/graphify/detect.py +++ b/graphify/detect.py @@ -41,7 +41,7 @@ class FileType(str, Enum): _MTIME_COARSE_S = 2.0 _MTIME_SUBSECOND_S = 0.05 -CODE_EXTENSIONS = {'.py', '.ts', '.tsx', '.mts', '.cts', '.js', '.jsx', '.mjs', '.cjs', '.ejs', '.ets', '.go', '.rs', '.java', '.groovy', '.gradle', '.cpp', '.cc', '.cxx', '.c', '.h', '.hpp', '.cu', '.cuh', '.metal', '.rb', '.rake', '.swift', '.kt', '.kts', '.cs', '.scala', '.php', '.lua', '.luau', '.toc', '.zig', '.ps1', '.psm1', '.psd1', '.ex', '.exs', '.m', '.mm', '.ml', '.mli', '.jl', '.vue', '.svelte', '.astro', '.dart', '.v', '.sv', '.svh', '.sql', '.r', '.f', '.F', '.f90', '.F90', '.f95', '.F95', '.f03', '.F03', '.f08', '.F08', '.pas', '.pp', '.dpr', '.dpk', '.lpr', '.inc', '.dfm', '.lfm', '.lpk', '.sh', '.bash', '.json', '.tf', '.tfvars', '.hcl', '.dm', '.dme', '.dmi', '.dmm', '.dmf', '.sln', '.slnx', '.csproj', '.fsproj', '.vbproj', '.xaml', '.razor', '.cshtml', '.cls', '.trigger', '.lisp', '.cl', '.lsp', '.asd'} +CODE_EXTENSIONS = {'.py', '.ts', '.tsx', '.mts', '.cts', '.js', '.jsx', '.mjs', '.cjs', '.ejs', '.ets', '.go', '.rs', '.java', '.groovy', '.gradle', '.cpp', '.cc', '.cxx', '.c', '.h', '.hpp', '.cu', '.cuh', '.metal', '.rb', '.rake', '.swift', '.kt', '.kts', '.cs', '.scala', '.php', '.lua', '.luau', '.toc', '.zig', '.ps1', '.psm1', '.psd1', '.ex', '.exs', '.m', '.mm', '.ml', '.mli', '.jl', '.vue', '.svelte', '.astro', '.dart', '.v', '.sv', '.svh', '.sql', '.r', '.f', '.F', '.f90', '.F90', '.f95', '.F95', '.f03', '.F03', '.f08', '.F08', '.pas', '.pp', '.dpr', '.dpk', '.lpr', '.inc', '.dfm', '.lfm', '.lpk', '.sh', '.bash', '.json', '.tf', '.tfvars', '.hcl', '.dm', '.dme', '.dmi', '.dmm', '.dmf', '.sln', '.slnx', '.csproj', '.fsproj', '.vbproj', '.xaml', '.razor', '.cshtml', '.cls', '.trigger', '.lisp', '.cl', '.lsp', '.asd', '.service', '.timer', '.socket', '.target', '.path', '.mount', '.slice'} DOC_EXTENSIONS = {'.md', '.mdx', '.qmd', '.skill', '.txt', '.rst', '.html', '.yaml', '.yml'} PAPER_EXTENSIONS = {'.pdf'} IMAGE_EXTENSIONS = {'.png', '.jpg', '.jpeg', '.gif', '.webp', '.svg'} diff --git a/graphify/extract.py b/graphify/extract.py index f68757a85..7f166f024 100644 --- a/graphify/extract.py +++ b/graphify/extract.py @@ -54,6 +54,7 @@ from graphify.extractors.razor import extract_razor # noqa: F401 from graphify.extractors.rust import extract_rust # noqa: F401 from graphify.extractors.sln import extract_sln # noqa: F401 +from graphify.extractors.systemd import SYSTEMD_UNIT_EXTENSIONS, extract_systemd # noqa: F401 from graphify.extractors.sql import extract_sql # noqa: F401 from graphify.extractors.terraform import extract_terraform # noqa: F401 from graphify.extractors.verilog import extract_verilog # noqa: F401 @@ -5252,6 +5253,8 @@ def add_existing_edge(edge: dict) -> None: ".dmf": extract_dmf, ".sln": extract_sln, ".slnx": extract_slnx, + # systemd units (#2848): .service .timer .socket .target .path .mount .slice + **{ext: extract_systemd for ext in sorted(SYSTEMD_UNIT_EXTENSIONS)}, ".csproj": extract_csproj, ".fsproj": extract_csproj, ".vbproj": extract_csproj, diff --git a/graphify/extractors/resolution.py b/graphify/extractors/resolution.py index 5369637ae..b103e8d3a 100644 --- a/graphify/extractors/resolution.py +++ b/graphify/extractors/resolution.py @@ -640,6 +640,19 @@ def _blank(s: str) -> str: out.append(_blank(src[pos:])) return "".join(out), lang +# Relations whose edges may carry a transient ``target_file`` stamp naming the +# file the target id was minted from. Import-style edges (#1814/#1983), plus +# the systemd unit edges (#2848): a `.timer` and its `.service` share the +# extension-less file-node id, so without the stamp the timer's `activates` +# edge would be disambiguated against the timer's own path — a self-loop. +_TARGET_FILE_RELATIONS: frozenset[str] = frozenset({ + "imports", "imports_from", "re_exports", + "activates", "runs", "documented_by", + "after", "before", "wants", "requires", "binds_to", "part_of", "conflicts", + "wanted_by", "required_by", +}) + + def _source_key(source_file: str, root: Path) -> str: if not source_file: return "" @@ -766,7 +779,7 @@ def _disambiguate_colliding_node_ids( # every language and to re_exports. `pop` it as we consume it: this is the # hint's only reader, and its absolute path must not persist into graph.json. target_file = edge.pop("target_file", None) - if target_file and edge.get("relation") in ("imports", "imports_from", "re_exports"): + if target_file and edge.get("relation") in _TARGET_FILE_RELATIONS: target_edge_key = _source_key(str(target_file), root) else: target_edge_key = edge_source_key diff --git a/graphify/extractors/systemd.py b/graphify/extractors/systemd.py new file mode 100644 index 000000000..89e332ef4 --- /dev/null +++ b/graphify/extractors/systemd.py @@ -0,0 +1,325 @@ +"""systemd unit extractor (#2848). + +Units are INI, so no grammar is needed. A repo that keeps its units under +version control keeps its whole OS-level scheduled-job topology there — and +without this pass that topology is silently absent: the graph answers "what +runs on a schedule" from the application's in-process scheduler alone and +looks complete. + +Handled: ``.service`` ``.timer`` ``.socket`` ``.target`` ``.path`` ``.mount`` +``.slice``. Every unit becomes a file node. Edges, in the direction the +question is usually asked: + +================= ============================ ================================= +relation from -> to source key +================= ============================ ================================= +``activates`` timer/socket/path -> unit ``[Timer] Unit=`` / ``[Socket] + Service=`` / ``[Path] Unit=``, + else the same-stem ``.service`` +``runs`` service -> script ``ExecStart=`` and friends, + after stripping systemd's + prefixes and ``/usr/bin/env`` + + interpreter +``documented_by`` unit -> doc ``Documentation=file://...`` +``after`` ``before`` ``wants`` ``requires`` +``binds_to`` ``part_of`` ``conflicts`` +``wanted_by`` ``required_by`` + unit -> unit the ``[Unit]`` / ``[Install]`` + ordering and dependency keys +================= ============================ ================================= + +Unit -> unit edges are emitted only when the target unit is a file beside +this one (units of one deployment live in one directory): ``After= +network-online.target`` names the host's unit, and manufacturing a node +for every system target would put a phantom hub in every repo. A +template's instance (``backup@nightly.service``) resolves to the template +file (``backup@.service``). + +Script and doc targets are deployment paths (``ExecStart=/opt/app/bin/ +run.py``), so they rarely exist at that path inside the repo. Resolution +tries the literal path (relative to the unit's directory, then absolute), +then walks up from the unit's directory looking for the same tail +(``bin/run.py``, then ``run.py``) — the common "units/ beside bin/" layout. +A target that resolves is minted the way every other extractor mints a +file reference (``_make_id(str(resolved))``), which extract() rewires onto +the real file node; one that does not is skipped, not fabricated. +""" +from __future__ import annotations + +import re +from pathlib import Path, PurePosixPath + +from graphify.extractors.base import _make_id + +SYSTEMD_UNIT_EXTENSIONS: frozenset[str] = frozenset({ + ".service", ".timer", ".socket", ".target", ".path", ".mount", ".slice", +}) + +# [Unit] / [Install] keys that name other units, and the relation each becomes. +_UNIT_REF_KEYS: dict[str, str] = { + "after": "after", + "before": "before", + "wants": "wants", + "requires": "requires", + "requisite": "requires", + "bindsto": "binds_to", + "partof": "part_of", + "conflicts": "conflicts", + "wantedby": "wanted_by", + "requiredby": "required_by", + "upholds": "wants", +} + +# The key that names what an activating unit starts, per section. +_ACTIVATES_KEY: dict[str, str] = {"timer": "unit", "socket": "service", "path": "unit"} + +_EXEC_KEYS = ("execstart", "execstartpre", "execstartpost", "execreload", + "execstop", "execstoppost", "execcondition") + +# Interpreters whose first argument is the real script. +_INTERPRETERS: frozenset[str] = frozenset({ + "sh", "bash", "zsh", "dash", "ksh", "fish", + "python", "python2", "python3", "pypy", "pypy3", + "node", "nodejs", "deno", "bun", "ruby", "perl", "php", "lua", "Rscript", + "uv", "uvx", "npx", "pipx", "poetry", "pnpm", "npm", "yarn", +}) +_INTERPRETER_RE = re.compile(r"^(?:python|pypy)\d+(?:\.\d+)?$") + +_SECTION_RE = re.compile(r"^\s*\[([^\]]+)\]\s*$") +_SPLIT_WS_RE = re.compile(r"\s+") + + +def _read_unit(path: Path) -> list[tuple[str, str, str, int]]: + """Yield ``(section, key, value, line)`` with ``\\``-continuations joined. + + Sections and keys are lower-cased; values are stripped. Comments (``#``, + ``;``) and blank lines are dropped. ``line`` is the 1-based line the + logical entry started on. + """ + text = path.read_text(encoding="utf-8", errors="replace") + entries: list[tuple[str, str, str, int]] = [] + section = "" + pending: list[str] = [] + pending_line = 0 + for lineno, raw in enumerate(text.splitlines(), 1): + if pending: + piece = raw.rstrip() + if piece.endswith("\\"): + pending.append(piece[:-1].strip()) + continue + pending.append(piece.strip()) + logical, start = " ".join(p for p in pending if p), pending_line + pending = [] + else: + stripped = raw.strip() + if not stripped or stripped[0] in "#;": + continue + m = _SECTION_RE.match(stripped) + if m: + section = m.group(1).strip().lower() + continue + if stripped.endswith("\\"): + pending = [stripped[:-1].strip()] + pending_line = lineno + continue + logical, start = stripped, lineno + if "=" not in logical: + continue + key, value = logical.split("=", 1) + entries.append((section, key.strip().lower(), value.strip(), start)) + return entries + + +def _template_of(name: str) -> str | None: + """``backup@nightly.service`` -> ``backup@.service``; else None.""" + stem, dot, ext = name.rpartition(".") + if not dot or "@" not in stem or stem.endswith("@"): + return None + return stem.split("@", 1)[0] + "@." + ext + + +def _sibling_unit(unit_dir: Path, name: str) -> Path | None: + """The unit file ``name`` beside this unit, or its template, if present.""" + name = name.strip() + if not name or "/" in name or "\\" in name: + return None + if Path(name).suffix.lower() not in SYSTEMD_UNIT_EXTENSIONS: + return None + candidate = unit_dir / name + if candidate.is_file(): + return candidate + template = _template_of(name) + if template and (unit_dir / template).is_file(): + return unit_dir / template + return None + + +def _resolve_path_target(unit_dir: Path, raw: str) -> Path | None: + """Resolve a deployment path named by a unit to a file in the repo.""" + raw = raw.strip().strip('"').strip("'") + if not raw or raw.startswith("$"): + return None + # Deployment paths are POSIX by definition (systemd is Linux); parse them + # as such so `/opt/app/run.py` is absolute on every host graphify runs on. + p = PurePosixPath(raw) + if not p.is_absolute(): + direct = unit_dir / Path(*p.parts) + return direct if direct.is_file() else None + bases = _search_bases(unit_dir) + # An absolute path is a HOST path. It may only resolve to a file inside + # the corpus: on a Linux host `/usr/bin/mkdir` exists, and an edge to it + # would be an edge to the machine graphify happens to run on. + direct = Path(raw) + if direct.is_file() and any(_is_under(direct, b) for b in bases): + return direct + # /opt/app/bin/run.py -> look for bin/run.py, then run.py, walking up + # from the unit's directory; deepest match wins, shortest tail last. + parts = p.parts[1:] # drop the anchor + if not parts or not p.suffix: + return None + for start in range(max(0, len(parts) - 3), len(parts)): + tail = Path(*parts[start:]) + for base in bases: + candidate = base / tail + if candidate.is_file(): + return candidate + return None + + +def _search_bases(unit_dir: Path) -> list[Path]: + """Directories a deployment path may resolve under: the unit's directory + and its ancestors up to the scan root. Outside a scan (a direct call) + the walk is bounded to a few levels so it can never reach the host's + ``/`` — where `usr/bin/python3.12` would otherwise match.""" + try: + import graphify.extract as _extract + root = getattr(_extract, "_XAML_ACTIVE_EXTRACT_ROOT", None) + except Exception: # pragma: no cover + root = None + bases = [unit_dir] + for parent in unit_dir.parents: + if root is not None: + if not _is_under(parent, Path(root)): + break + elif len(bases) > 3: + break + bases.append(parent) + return bases + + +def _is_under(path: Path, base: Path) -> bool: + try: + path.resolve().relative_to(base.resolve()) + return True + except (ValueError, OSError): + return False + + +def _script_from_exec(value: str) -> str | None: + """The script a systemd ``Exec*=`` line runs, or None for a bare binary.""" + v = value.strip() + # systemd's executable prefixes: -, @, :, +, !, !! (any combination). + while v and v[0] in "-@:+!": + v = v[1:] + tokens = _SPLIT_WS_RE.split(v.strip()) + if not tokens or not tokens[0]: + return None + i = 0 + # `@/path/to/prog argv0 ...` — with `@` the second token is argv[0], skip it. + if value.strip().startswith("@") and len(tokens) > 1: + tokens = [tokens[0]] + tokens[2:] + while i < len(tokens): + tok = tokens[i] + base = Path(tok).name + if base == "env": + # skip `env`, its VAR=val assignments and -S/-i style flags + i += 1 + while i < len(tokens) and ("=" in tokens[i] or tokens[i].startswith("-")): + i += 1 + continue + if base in _INTERPRETERS or _INTERPRETER_RE.match(base): + # `python3 -m pkg.mod` is a module, not a file; `uv run script.py` + # and `npx tsx script.ts` carry the script one token further on. + i += 1 + while i < len(tokens) and tokens[i].startswith("-"): + if tokens[i] in ("-m", "--module"): + return None + i += 1 + if base in ("uv", "uvx", "npx", "pipx", "poetry", "pnpm", "npm", "yarn"): + if i < len(tokens) and tokens[i] in ("run", "exec", "tsx", "ts-node"): + i += 1 + while i < len(tokens) and tokens[i].startswith("-"): + i += 1 + continue + return tok + return None + + +def extract_systemd(path: Path) -> dict: + """Extract a systemd unit: one file node plus its activation, script, + documentation and ordering edges. See the module docstring.""" + try: + entries = _read_unit(path) + except OSError: + return {"nodes": [], "edges": [], "error": f"cannot read {path}"} + + str_path = str(path) + file_nid = _make_id(str_path) + unit_dir = path.parent + nodes: list[dict] = [{"id": file_nid, "label": path.name, "file_type": "code", + "source_file": str_path, "source_location": None}] + edges: list[dict] = [] + seen: set[tuple[str, str]] = set() + + def _edge(target: Path, relation: str, line: int) -> None: + target_nid = _make_id(str(target)) + if target_nid == file_nid or (target_nid, relation) in seen: + return + seen.add((target_nid, relation)) + edges.append({ + "source": file_nid, "target": target_nid, "relation": relation, + "confidence": "EXTRACTED", "source_file": str_path, + "source_location": f"{path.name}:{line}", "weight": 1.0, + # Transient stamp consumed by the corpus-level id disambiguation: + # `x.service` and `x.timer` share the extension-less file-node id, + # and without naming the target FILE the edge would be resolved + # against this unit's own path and turn into a self-loop. + "target_file": str(target), + }) + + unit_kind = path.suffix.lower().lstrip(".") + activates_key = _ACTIVATES_KEY.get(unit_kind) + explicit_activation = False + + for section, key, value, line in entries: + if section == unit_kind and key == activates_key: + explicit_activation = True + target = _sibling_unit(unit_dir, value) + if target is not None: + _edge(target, "activates", line) + elif section in ("unit", "install") and key in _UNIT_REF_KEYS: + for name in _SPLIT_WS_RE.split(value): + target = _sibling_unit(unit_dir, name) + if target is not None: + _edge(target, _UNIT_REF_KEYS[key], line) + elif section == "unit" and key == "documentation": + for ref in _SPLIT_WS_RE.split(value): + if ref.startswith("file://"): + target = _resolve_path_target(unit_dir, ref[len("file://"):]) + if target is not None: + _edge(target, "documented_by", line) + elif section == "service" and key in _EXEC_KEYS: + script = _script_from_exec(value) + if script: + target = _resolve_path_target(unit_dir, script) + if target is not None: + _edge(target, "runs", line) + + # A timer/socket/path with no explicit Unit= activates the same-stem + # .service by systemd's own convention. + if activates_key and not explicit_activation: + implied = _sibling_unit(unit_dir, path.stem + ".service") + if implied is not None: + _edge(implied, "activates", 1) + + return {"nodes": nodes, "edges": edges} diff --git a/tests/test_systemd_units.py b/tests/test_systemd_units.py new file mode 100644 index 000000000..d3740f0b8 --- /dev/null +++ b/tests/test_systemd_units.py @@ -0,0 +1,267 @@ +"""systemd units reach the graph (#2848). + +`.service`/`.timer` were in no extension set, so a repo that keeps its units +under version control had its whole OS-level scheduled-job topology missing — +and the graph answered "what runs on a schedule" confidently from the app's +in-process scheduler alone. Units are INI; a regex pass gives every unit a +file node and links it to what it activates, runs and documents, landing on +nodes the AST pass already creates. +""" +from __future__ import annotations + +import io +from contextlib import redirect_stdout +from pathlib import Path + +import pytest + +from graphify.detect import FileType, classify_file +from graphify.extract import _get_extractor, extract +from graphify.extractors.systemd import ( + SYSTEMD_UNIT_EXTENSIONS, + _resolve_path_target, + _script_from_exec, + _template_of, + extract_systemd, +) + + +def _rels(result, relation=None): + out = [] + for e in result["edges"]: + if relation is None or e["relation"] == relation: + out.append((e["relation"], Path(e["target_file"]).name if "target_file" in e else e["target"])) + return out + + +@pytest.fixture +def deployment(tmp_path): + """units/ beside bin/ and docs/, deployed under /opt/sl on the host.""" + units = tmp_path / "deploy" / "units" + bin_ = tmp_path / "deploy" / "bin" + docs = tmp_path / "docs" + for d in (units, bin_, docs): + d.mkdir(parents=True) + (bin_ / "daily_audit.py").write_text("def main():\n pass\n", encoding="utf-8") + (bin_ / "backup.sh").write_text("#!/bin/bash\necho hi\n", encoding="utf-8") + (docs / "audit.md").write_text("# Audit\n", encoding="utf-8") + (units / "daily-audit.service").write_text( + "[Unit]\n" + "Description=Daily audit\n" + "Documentation=file:///opt/sl/docs/audit.md https://example.com/audit\n" + "After=network-online.target backup.service\n" + "Wants=backup.service\n" + "\n" + "[Service]\n" + "Type=oneshot\n" + "ExecStart=/usr/bin/env python3 /opt/sl/bin/daily_audit.py \\\n" + " --verbose\n" + "ExecStartPre=-/usr/bin/mkdir -p /var/lib/sl\n", + encoding="utf-8", + ) + (units / "daily-audit.timer").write_text( + "[Unit]\nDescription=Run the audit daily\n\n[Timer]\nOnCalendar=daily\n\n" + "[Install]\nWantedBy=timers.target\n", + encoding="utf-8", + ) + (units / "backup@.service").write_text( + "[Service]\nExecStart=@/bin/bash backup /opt/sl/bin/backup.sh\n", encoding="utf-8", + ) + (units / "backup-nightly.timer").write_text( + "[Timer]\nUnit=backup@nightly.service\n", encoding="utf-8", + ) + (units / "watchdog.service").write_text( + "[Unit]\nRequires=daily-audit.service\nBefore=daily-audit.service\n" + "[Service]\nExecStart=/usr/bin/python3 -m sl.watchdog\n", + encoding="utf-8", + ) + return tmp_path + + +# --------------------------------------------------------------------------- +# Detection and dispatch +# --------------------------------------------------------------------------- + +@pytest.mark.parametrize("ext", sorted(SYSTEMD_UNIT_EXTENSIONS)) +def test_every_unit_type_is_classified_as_code_and_dispatched(ext): + assert classify_file(Path(f"x{ext}")) is FileType.CODE + assert _get_extractor(Path(f"x{ext}")) is extract_systemd + + +# --------------------------------------------------------------------------- +# The per-file extractor +# --------------------------------------------------------------------------- + +def test_a_service_runs_its_script_and_is_documented_by_its_doc(deployment): + r = extract_systemd(deployment / "deploy" / "units" / "daily-audit.service") + assert r["nodes"][0]["label"] == "daily-audit.service" + assert r["nodes"][0]["file_type"] == "code" + assert ("runs", "daily_audit.py") in _rels(r) + assert ("documented_by", "audit.md") in _rels(r) + + +def test_a_timer_with_no_unit_key_activates_the_same_stem_service(deployment): + r = extract_systemd(deployment / "deploy" / "units" / "daily-audit.timer") + assert _rels(r, "activates") == [("activates", "daily-audit.service")] + + +def test_an_instance_name_resolves_to_its_template(deployment): + r = extract_systemd(deployment / "deploy" / "units" / "backup-nightly.timer") + assert _rels(r, "activates") == [("activates", "backup@.service")] + + +def test_ordering_keys_become_unit_to_unit_edges(deployment): + r = extract_systemd(deployment / "deploy" / "units" / "watchdog.service") + assert ("requires", "daily-audit.service") in _rels(r) + assert ("before", "daily-audit.service") in _rels(r) + + +def test_units_of_the_host_are_never_fabricated(deployment): + """After=network-online.target and WantedBy=timers.target name units that + live on the host, not in the repo; a node for each would put a phantom + hub in every repo.""" + svc = extract_systemd(deployment / "deploy" / "units" / "daily-audit.service") + tmr = extract_systemd(deployment / "deploy" / "units" / "daily-audit.timer") + targets = {Path(e.get("target_file", "")).name for e in svc["edges"] + tmr["edges"]} + assert "network-online.target" not in targets + assert "timers.target" not in targets + assert "backup.service" not in targets # named, but no such file beside the unit + assert all(n["label"].endswith((".service", ".timer")) for n in svc["nodes"] + tmr["nodes"]) + + +def test_a_module_invocation_is_not_a_script(deployment): + r = extract_systemd(deployment / "deploy" / "units" / "watchdog.service") + assert _rels(r, "runs") == [] + + +def test_a_continuation_line_is_joined(tmp_path): + u = tmp_path / "a.service" + (tmp_path / "run.sh").write_text("", encoding="utf-8") + u.write_text("[Service]\nExecStart=/bin/sh \\\n /opt/x/run.sh \\\n --flag\n", encoding="utf-8") + assert _rels(extract_systemd(u), "runs") == [("runs", "run.sh")] + + +def test_comments_and_bare_lines_are_ignored(tmp_path): + u = tmp_path / "a.timer" + (tmp_path / "a.service").write_text("[Service]\n", encoding="utf-8") + u.write_text("# comment\n; also comment\n[Timer]\nnot a key value\nOnBootSec=5\n", encoding="utf-8") + assert _rels(extract_systemd(u), "activates") == [("activates", "a.service")] + + +def test_an_unreadable_unit_reports_an_error_not_a_crash(tmp_path): + r = extract_systemd(tmp_path / "missing.service") + assert r["nodes"] == [] and "error" in r + + +# --------------------------------------------------------------------------- +# Exec= parsing +# --------------------------------------------------------------------------- + +@pytest.mark.parametrize("value, expected", [ + ("/opt/app/bin/run.py --x", "/opt/app/bin/run.py"), + ("-/opt/app/bin/run.py", "/opt/app/bin/run.py"), + ("!!/opt/app/bin/run.py", "/opt/app/bin/run.py"), + ("/usr/bin/env bash /opt/app/run.sh", "/opt/app/run.sh"), + ("/usr/bin/env -S python3 -u /opt/app/run.py", "/opt/app/run.py"), + ("/usr/bin/env FOO=1 python3.12 /opt/app/run.py", "/opt/app/run.py"), + ("/usr/bin/python3 -u /opt/app/run.py", "/opt/app/run.py"), + ("/usr/bin/node /srv/app/worker.js", "/srv/app/worker.js"), + ("/usr/local/bin/uv run /opt/app/run.py", "/opt/app/run.py"), + ("npx tsx /srv/app/worker.ts", "/srv/app/worker.ts"), + ("@/bin/bash backup /opt/app/backup.sh", "/opt/app/backup.sh"), + ("/usr/bin/python3 -m pkg.mod", None), + ("/usr/bin/docker run image", "/usr/bin/docker"), + ("", None), +]) +def test_script_from_exec(value, expected): + assert _script_from_exec(value) == expected + + +# --------------------------------------------------------------------------- +# Deployment path resolution +# --------------------------------------------------------------------------- + +def test_a_deploy_path_resolves_by_its_tail_walking_up_from_the_unit(deployment): + unit_dir = deployment / "deploy" / "units" + assert _resolve_path_target(unit_dir, "/opt/sl/bin/daily_audit.py") == deployment / "deploy" / "bin" / "daily_audit.py" + assert _resolve_path_target(unit_dir, "/opt/sl/docs/audit.md") == deployment / "docs" / "audit.md" + assert _resolve_path_target(unit_dir, "/opt/sl/bin/nope.py") is None + assert _resolve_path_target(unit_dir, "/usr/bin/mkdir") is None # no suffix: a host binary + assert _resolve_path_target(unit_dir, "$SCRIPT") is None + + +def test_a_host_file_outside_the_scan_root_never_resolves(deployment, monkeypatch): + """On a Linux host `/usr/bin/python3.12` exists; an edge to it would point + at the machine graphify runs on, not the repo. Absolute paths resolve + only inside the scan root, and the tail walk stops there too.""" + import graphify.extract as extractmod + outside = deployment.parent / "outside_tool.py" + outside.write_text("", encoding="utf-8") + monkeypatch.setattr(extractmod, "_XAML_ACTIVE_EXTRACT_ROOT", deployment.resolve(), raising=False) + unit_dir = deployment / "deploy" / "units" + assert _resolve_path_target(unit_dir, outside.as_posix()) is None + assert _resolve_path_target(unit_dir, "/nowhere/" + outside.name) is None + # and a file inside the root still does + assert _resolve_path_target(unit_dir, "/opt/sl/bin/backup.sh") == deployment / "deploy" / "bin" / "backup.sh" + + +def test_a_posix_deploy_path_is_absolute_on_every_host(deployment): + """Windows would call `/opt/x` relative and never try the tail walk.""" + unit_dir = deployment / "deploy" / "units" + assert _resolve_path_target(unit_dir, "/opt/sl/bin/backup.sh") is not None + + +def test_a_relative_path_resolves_beside_the_unit(tmp_path): + (tmp_path / "run.sh").write_text("", encoding="utf-8") + assert _resolve_path_target(tmp_path, "run.sh") == tmp_path / "run.sh" + assert _resolve_path_target(tmp_path, "./run.sh") == tmp_path / "run.sh" + + +@pytest.mark.parametrize("name, expected", [ + ("backup@nightly.service", "backup@.service"), + ("getty@tty1.service", "getty@.service"), + ("backup@.service", None), + ("backup.service", None), +]) +def test_template_of(name, expected): + assert _template_of(name) == expected + + +# --------------------------------------------------------------------------- +# Corpus level: the edges land on the real nodes +# --------------------------------------------------------------------------- + +def _corpus(root): + files = [p for p in root.rglob("*") if p.is_file() and "graphify-out" not in p.parts] + with redirect_stdout(io.StringIO()): + r = extract(files, cache_root=root, root=root) + labels = {n["id"]: n["label"] for n in r["nodes"]} + return r, labels + + +def test_edges_land_on_the_script_and_doc_file_nodes(deployment): + r, labels = _corpus(deployment) + by_label = {(labels[e["source"]], e["relation"], labels.get(e["target"])) for e in r["edges"]} + assert ("daily-audit.service", "runs", "daily_audit.py") in by_label + assert ("daily-audit.service", "documented_by", "audit.md") in by_label + assert ("backup@.service", "runs", "backup.sh") in by_label + + +def test_a_timer_and_its_service_share_a_stem_and_still_link_correctly(deployment): + """`x.timer` and `x.service` collide on the extension-less file-node id; + the collision remap must resolve the edge by its target file, not turn + it into a self-loop on the timer.""" + r, labels = _corpus(deployment) + activates = [(labels[e["source"]], labels.get(e["target"])) for e in r["edges"] if e["relation"] == "activates"] + assert ("daily-audit.timer", "daily-audit.service") in activates + assert ("backup-nightly.timer", "backup@.service") in activates + assert all(s != t for s, t in activates) + + +def test_no_edge_dangles_and_the_stamp_does_not_leak(deployment): + r, labels = _corpus(deployment) + unit_edges = [e for e in r["edges"] if e["relation"] in + ("activates", "runs", "documented_by", "requires", "before", "wants", "after")] + assert unit_edges + assert all(e["target"] in labels for e in unit_edges) + assert not any("target_file" in e for e in r["edges"])