From 9d289eae382a619d11a8e6fdb660138df85ca309 Mon Sep 17 00:00:00 2001 From: Toddy Mladenov Date: Fri, 28 Aug 2026 21:07:55 -0700 Subject: [PATCH] graph: provenance-aware mermaid/cytoscape rendering (#212) --- .../libs/cssc_graph/cssc_graph/queries.py | 93 +++++++++++++++++-- .../libs/cssc_graph/tests/test_queries.py | 48 ++++++++++ 2 files changed, 131 insertions(+), 10 deletions(-) diff --git a/apps/python-app/libs/cssc_graph/cssc_graph/queries.py b/apps/python-app/libs/cssc_graph/cssc_graph/queries.py index 743c8b7..8c93084 100644 --- a/apps/python-app/libs/cssc_graph/cssc_graph/queries.py +++ b/apps/python-app/libs/cssc_graph/cssc_graph/queries.py @@ -554,9 +554,40 @@ def add_edge(kind: str, frm: str, to: str, date: Any, **extra: Any) -> None: # -- export ------------------------------------------------------------------- +# Typed provenance edge kinds (see :func:`provenance`); their labels carry a date. +_PROVENANCE_KINDS = ("imported", "promoted", "built", "attests") + +# Mermaid class per provenance node state; deleted nodes override to `deleted`. +_MERMAID_STATE_CLASS = { + "upstream": "tagroot", + "present-quarantine": "quarantine", + "present-golden": "golden", + "present": "present", + "deleted": "deleted", +} + +_MERMAID_CLASSDEFS = ( + " classDef tagroot fill:#f5f5f5,stroke:#999999,stroke-dasharray:5 5;", + " classDef quarantine fill:#fff3cd,stroke:#b8860b;", + " classDef golden fill:#d4edda,stroke:#28a745;", + " classDef present fill:#e2e3ff,stroke:#6f42c1;", + " classDef referrer fill:#cce5ff,stroke:#007bff;", + " classDef deleted fill:#eeeeee,stroke:#999999,stroke-dasharray:3 3,color:#888888;", +) + def _node_label(node: dict[str, Any]) -> str: """A label that disambiguates occurrences of the same repository by digest.""" + node_type = node.get("nodeType") + if node_type == "tag-root": + ref = node.get("ref") or node["key"] + tag = node.get("tag") + return f"{ref}:{tag}" if tag else str(ref) + if node_type == "referrer": + label = node.get("artifactType") or "referrer" + if node.get("deletedAt"): + label += " (deleted)" + return label ref = node.get("ref") or node["key"] digest = node.get("digest") or "" short = digest[7:19] if digest.startswith("sha256:") else digest[:12] @@ -566,34 +597,76 @@ def _node_label(node: dict[str, Any]) -> str: return label -def to_cytoscape(subgraph: dict[str, Any]) -> dict[str, Any]: - elements = [{"data": {"id": n["key"], "label": _node_label(n), **n}} for n in subgraph["nodes"]] - for i, e in enumerate(subgraph["edges"]): - label = _edge_label(e) - elements.append({"data": {"id": f"e{i}", "source": e["from"], "target": e["to"], "label": label, **e}}) - return {"elements": elements} - - def _edge_label(edge: dict[str, Any]) -> str: + kind = edge.get("type") + if kind in _PROVENANCE_KINDS: + # Timeline edges are labelled with their kind and the (date-only) date. + date = str(edge.get("date") or "")[:10] + label = f"{kind} {date}".strip() + platform = edge.get("platform") + return f"{label} ({platform})" if platform else label label = edge.get("artifactType") or edge["type"] platform = edge.get("platform") return f"{label} ({platform})" if platform else label +def _node_class(node: dict[str, Any]) -> str: + """The provenance style class for a node (empty for non-provenance nodes).""" + if node.get("deletedAt") or node.get("state") == "deleted": + return "deleted" + if node.get("nodeType") == "referrer": + return "referrer" + return _MERMAID_STATE_CLASS.get(node.get("state", ""), "") + + +def to_cytoscape(subgraph: dict[str, Any]) -> dict[str, Any]: + elements: list[dict[str, Any]] = [] + for n in subgraph["nodes"]: + element: dict[str, Any] = {"data": {"id": n["key"], "label": _node_label(n), **n}} + classes = " ".join(c for c in (n.get("nodeType"), _node_class(n)) if c) + if classes: + element["classes"] = classes + elements.append(element) + for i, e in enumerate(subgraph["edges"]): + element = { + "data": {"id": f"e{i}", "source": e["from"], "target": e["to"], "label": _edge_label(e), **e} + } + if e.get("type") in _PROVENANCE_KINDS: + element["classes"] = e["type"] + elements.append(element) + return {"elements": elements} + + def _mermaid_label(text: str) -> str: # Mermaid labels are quoted; neutralize quotes/newlines to avoid broken output. return str(text).replace('"', "#quot;").replace("\n", " ").replace("\r", " ") +def _mermaid_node_line(nid: str, node: dict[str, Any]) -> str: + label = _mermaid_label(_node_label(node)) + node_type = node.get("nodeType") + if node_type == "tag-root": # dashed rounded via shape + classDef + return f' {nid}("{label}")' + if node_type == "referrer": + return f' {nid}{{{{"{label}"}}}}' + return f' {nid}["{label}"]' + + def to_mermaid(subgraph: dict[str, Any]) -> str: lines = ["flowchart LR"] ids = {n["key"]: f"n{i}" for i, n in enumerate(subgraph["nodes"])} + styled = any(n.get("nodeType") for n in subgraph["nodes"]) for n in subgraph["nodes"]: - label = _mermaid_label(_node_label(n)) - lines.append(f' {ids[n["key"]]}["{label}"]') + lines.append(_mermaid_node_line(ids[n["key"]], n)) for e in subgraph["edges"]: frm, to = ids.get(e["from"]), ids.get(e["to"]) if frm and to: rel = _mermaid_label(_edge_label(e)) lines.append(f' {frm} -->|{rel}| {to}') + if styled: + for n in subgraph["nodes"]: + cls = _node_class(n) + if cls: + lines.append(f' class {ids[n["key"]]} {cls};') + lines.extend(_MERMAID_CLASSDEFS) return "\n".join(lines) diff --git a/apps/python-app/libs/cssc_graph/tests/test_queries.py b/apps/python-app/libs/cssc_graph/tests/test_queries.py index 12d69d1..9618d44 100644 --- a/apps/python-app/libs/cssc_graph/tests/test_queries.py +++ b/apps/python-app/libs/cssc_graph/tests/test_queries.py @@ -240,3 +240,51 @@ def test_provenance_unknown_family_is_empty(store): sub = queries.provenance(store, "does-not-exist") assert sub == {"nodes": [], "edges": []} + +def test_provenance_mermaid_styles_nodes_and_dates(store): + m = queries.to_mermaid(queries.provenance(store, "python")) + assert m.startswith("flowchart") + # style classes are emitted for a provenance subgraph + assert "classDef tagroot" in m + assert "classDef golden" in m + # tag-root uses a rounded shape; referrers use a hexagon shape + assert '("' in m + assert "{{" in m + # timeline edges carry kind + date-only + assert "imported 2026-08-09" in m + + +def test_provenance_cytoscape_tags_classes(store): + cy = queries.to_cytoscape(queries.provenance(store, "python")) + node_classes = {el.get("classes") for el in cy["elements"] if "source" not in el["data"]} + assert any(c and "tag-root" in c for c in node_classes) + assert any(c and "golden" in c for c in node_classes) + edge_classes = {el.get("classes") for el in cy["elements"] if "source" in el["data"]} + assert "imported" in edge_classes + + +def test_render_marks_deleted_node(): + sub = { + "nodes": [ + { + "key": "r/x@sha256:1111", + "ref": "r/x", + "digest": "sha256:" + "1" * 64, + "nodeType": "image", + "state": "deleted", + "deletedAt": "2026-01-01T00:00:00Z", + } + ], + "edges": [], + } + m = queries.to_mermaid(sub) + assert "class n0 deleted;" in m + assert "classDef deleted" in m + + +def test_non_provenance_subgraph_renders_without_classdefs(store): + m = queries.to_mermaid(queries.path(store, digest=DIGEST2)) + assert m.startswith("flowchart") + assert "classDef" not in m + +