Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
93 changes: 83 additions & 10 deletions apps/python-app/libs/cssc_graph/cssc_graph/queries.py
Original file line number Diff line number Diff line change
Expand Up @@ -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]
Expand All @@ -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", ""), "")
Comment on lines +613 to +619


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)

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Keep classes off non-provenance Cytoscape nodes

When a regular path() or referrers() subgraph contains a deleted occurrence, _node_class(n) returns deleted even though the node has no nodeType, so this unconditionally adds a top-level "classes": "deleted" field. That changes the previously stable Cytoscape output for non-provenance queries and may unexpectedly activate a consumer's .deleted styling; only derive these classes for typed provenance nodes.

Useful? React with 👍 / 👎.

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)
48 changes: 48 additions & 0 deletions apps/python-app/libs/cssc_graph/tests/test_queries.py
Original file line number Diff line number Diff line change
Expand Up @@ -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


Loading