diff --git a/CHANGELOG.md b/CHANGELOG.md index 2588ed4..4494c50 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,17 @@ Every change a user would notice, newest first ([Keep a Changelog](https://keepa follow [Semantic Versioning](https://semver.org/); before 1.0, a minor version may change what a command does. ## [Unreleased] +### Added + +- A Decisions tab beside the graph draws each call as a flowchart, top to bottom: each `if` as a question with an + arrow for yes and for no, `elif` and `match` as the next question, loops as boxes with an arrow back to their top, + `try` with the way taken when a line in it raises, and where the call returns or raises. The usual path runs down + the middle, and an answer that ends early (returns, raises, skips to the next item, leaves the loop) sits beside it. + A call into the code's own functions that decides things too is opened in place, or, deeper down, gets a flow of + its own its box links to. A web app has a flow for each route. It needs no JavaScript. +- Step by step says when a branch ends in `continue` or `break`, and an `except` that only continues is no longer + left out. + ## [0.9.0] - 2026-10-22 ### Added diff --git a/CLAUDE.md b/CLAUDE.md index 418aacc..ba58d83 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -54,7 +54,7 @@ Slash commands in `.claude/commands/`: `/check` (lint, tests, docs, package), `/ | `src/code_node/pages.py` | `PageOptions`, `make_page` / `make_mermaid`: read the folder, map, draw (what the command and the server share) | | `src/code_node/history.py` | What git says about each function shown (`read_history`: `git blame` once per file, onto `CodeFunction.history`). Read-only git, with no file system monitor or text conversion, and none at all when the folder's own config names a filter command | | `src/code_node/serve.py` | `--serve`: `LivePage` (the latest page, a version, a Condition), a polling watcher thread, `ThreadingHTTPServer` with `/` and `/events` (Server-Sent Events) | -| `src/code_node/report/` | Drawing: `brands.py` (which logo each AWS service and library service gets, read from `logos/*.svg` and, for AWS's, a small mark in `logos/small/` for 16 to 32 px; `logo_for(name, pixels)` picks, and the page's graph has both and picks by screen density. Written by `scripts/logos.py`: never edit them), `graph.py` (the SVG layout), `walk.py` (Step by step: the steps in run order beside their code), `tabs.py` (the Graph / Step by step tabs), `outline.py` (the step tree, for text), `build.py` (`page_blocks`, the page), `style.py` (CSS, `render_page`) | +| `src/code_node/report/` | Drawing: `brands.py` (which logo each AWS service and library service gets, read from `logos/*.svg` and, for AWS's, a small mark in `logos/small/` for 16 to 32 px; `logo_for(name, pixels)` picks, and the page's graph has both and picks by screen density. Written by `scripts/logos.py`: never edit them), `graph.py` (the SVG layout), `flow.py` (Decisions: each call as a flowchart of its ifs, loops and trys, top to bottom), `walk.py` (Step by step: the steps in run order beside their code), `tabs.py` (the Graph / Decisions / Step by step tabs), `outline.py` (the step tree, for text), `build.py` (`page_blocks`, the page), `style.py` (CSS, `render_page`) | | `src/code_node/kit/` | Report blocks (`_Title`, `_Cards`, `_Table`, `_Findings`, ...) and the HTML/text renderer, copied from aws-analyzer's `_kit` | | `examples/orders-service/` | A small Lambda service the screenshots and the CI smoke test run on. Never run | | `examples/books-api/` | A FastAPI app on the web platform (routes on an `APIRouter`, PostgreSQL, Redis, S3), with a mistake for each web finding, for `tests/snapshots/books-api.txt` and the Plugins page's web apps. Never run | diff --git a/README.md b/README.md index 9fd418d..20deb7a 100644 --- a/README.md +++ b/README.md @@ -37,6 +37,7 @@ cd your/repo && code-node | | | | --- | --- | | 🗺️ **Graph** | What comes in, who calls whom, what each path reaches. AWS calls coloured by kind (read, write, send, run); ↻ marks loops | +| 🔀 **Decisions** | Each call as a flowchart: every `if` a question with a yes and a no arrow, loops, `try`, where it returns or raises | | 👣 **Step by step** | Walk the code in run order, like a debugger: Next, Skip call, Finish, arrow keys | | 🔌 **Plugins** | boto3, Redis, SQL (sqlite3, psycopg, PyMySQL, asyncpg, SQLAlchemy), MongoDB, OpenAI, Anthropic | | 📄 **Deploy checks** | Reads SAM `template.yaml` / `serverless.yml`: unset variables, IAM calls the role won't allow, timeouts that don't fit | diff --git a/docs/images/decisions-dark.webp b/docs/images/decisions-dark.webp new file mode 100644 index 0000000..8b50990 Binary files /dev/null and b/docs/images/decisions-dark.webp differ diff --git a/docs/images/decisions-light.webp b/docs/images/decisions-light.webp new file mode 100644 index 0000000..970ded7 Binary files /dev/null and b/docs/images/decisions-light.webp differ diff --git a/docs/images/page-dark.webp b/docs/images/page-dark.webp index f06f646..4874ef3 100644 Binary files a/docs/images/page-dark.webp and b/docs/images/page-dark.webp differ diff --git a/docs/images/page-light.webp b/docs/images/page-light.webp index f885978..5979425 100644 Binary files a/docs/images/page-light.webp and b/docs/images/page-light.webp differ diff --git a/docs/images/steps-dark.webp b/docs/images/steps-dark.webp index 693f084..e919691 100644 Binary files a/docs/images/steps-dark.webp and b/docs/images/steps-dark.webp differ diff --git a/docs/images/steps-light.webp b/docs/images/steps-light.webp index c2a315b..4d77008 100644 Binary files a/docs/images/steps-light.webp and b/docs/images/steps-light.webp differ diff --git a/docs/index.md b/docs/index.md index c4d9ac1..7abc336 100644 --- a/docs/index.md +++ b/docs/index.md @@ -44,8 +44,9 @@ Select a highlighted line to see the path it takes, or point at a box. ![A code-node page: the graph in its tab](images/page-light.webp#only-light){ loading=lazy } diff --git a/docs/reading.md b/docs/reading.md index 2bd3921..2e6eb32 100644 --- a/docs/reading.md +++ b/docs/reading.md @@ -1,6 +1,6 @@ # Read the map -What each tab of the page shows. The page is two tabs, with the entry point and the folder at the right of them. +What each tab of the page shows. The page is three tabs, with the entry point and the folder at the right of them. { .gist } ## The tabs { #order } @@ -8,6 +8,7 @@ What each tab of the page shows. The page is two tabs, with the entry point and | Tab | Tells you | | --- | --- | | Graph | The paths: what comes in, who calls whom, and what it reaches | +| Decisions | Each question the code asks, and where each answer goes | | Step by step | Every line that matters, in the order it runs | ## The graph { #graph } @@ -80,9 +81,43 @@ and the boxes it calls are stacked in the same order, so you can read down the b every function at once. This is the one part of the page that needs JavaScript; without it the link isn't shown, and the Step by step tab has every call. +## Decisions { #decisions } + +The second tab draws each call as a flowchart, top to bottom, so you can see what decides which way the code goes. +{ .gist } + +It starts with what comes in and the entry point, then has a box for each line that does something, in the words +Step by step uses ("Takes `bucket`, `key` from the event"). A call into one of the code's own functions is a box +listing what it reaches, with the graph's logos. + +| Shape | Is | +| --- | --- | +| Box pointed at both ends | An `if`: the test, with **yes** and **no** on its two arrows. An `elif` is the next question on the no arrow, and so is each `case` of a `match` | +| Dashed box with an arrow back to its top | A loop: its lines, and the arrow back for the next item ("next order") | +| Dashed amber arrow | Out of a `try`, to what runs when a line in it raises | +| Black pill | Where the call returns | +| Red pill | Where it raises | +| Dashed purple pill | `continue` or `break`: skips to the next item, or leaves the loop | + +The answer that goes on runs down the middle, and one that ends early (returns, raises, skips to the next item, leaves +the loop) sits beside it, so the usual path reads as one line. `if not check(order): continue` is asked as +`check(order)?`, with yes going on and no skipping to the next order. + +![The Decisions tab](images/decisions-light.webp#only-light){ loading=lazy } +![The Decisions tab](images/decisions-dark.webp#only-dark){ loading=lazy } + +??? more "Calls that decide things too" + A call to one of the code's own functions that makes decisions of its own is opened in place: a box with its name, + holding its flow, whose returns go back to the caller. Past two calls deep, or when it would be very tall, it's a + box that says how many decisions are inside, and links to its flow under the first one. A web app has a flow for + each of its routes, each starting with its method and path. + +!!! tip "Worth knowing" + Point at a box, a question or an end to see the line of code it's for. + ## Step by step { #steps } -The tab beside the graph walks through the code in the order it runs, as a debugger steps through it. +The third tab walks through the code in the order it runs, as a debugger steps through it. { .gist } The list on the left has every line that does something, each said in a few plain words ("Takes `bucket` from the diff --git a/scripts/shots.py b/scripts/shots.py index 1f79e5e..cb3a59a 100644 --- a/scripts/shots.py +++ b/scripts/shots.py @@ -26,6 +26,7 @@ from code_node.pages import PageOptions, map_entry from code_node.project import entry_points, read_project from code_node.report.build import page_blocks +from code_node.report.flow import _Flow from code_node.report.graph import _Graph from code_node.report.style import render_page from code_node.report.tabs import _Tabs @@ -53,6 +54,7 @@ def _inside(blocks: list[Any]) -> list[Any]: FIGURES: dict[str, Callable[[list[Any]], list[Any]]] = { "page": lambda blocks: blocks, "graph": lambda blocks: [b for b in _inside(blocks) if isinstance(b, _Graph)], + "decisions": lambda blocks: [b for b in _inside(blocks) if isinstance(b, _Flow)], "steps": lambda blocks: [b for b in _inside(blocks) if isinstance(b, _Walk)], "plugins": lambda blocks: [b for b in _inside(blocks) if isinstance(b, _Graph)], # drawn from the cart example } diff --git a/src/code_node/analysis/model.py b/src/code_node/analysis/model.py index 0333ceb..6d16d2a 100644 --- a/src/code_node/analysis/model.py +++ b/src/code_node/analysis/model.py @@ -35,6 +35,7 @@ class CodeStep: # [('S3 PutObject Bucket', 'env ARCHIVE_BUCKET')] when: list[str] = field(default_factory=list) # a 'def' step: the tests of the ifs around the line calling it, # outermost first: ['args.command == "init"'], ['check(order)'] after `if not check(order): continue` + ends: str = "" # an 'if', 'else' or 'except' whose lines end by leaving its loop: 'continue' or 'break' def walk(self) -> Iterable[CodeStep]: """ diff --git a/src/code_node/analysis/reader.py b/src/code_node/analysis/reader.py index 69e4176..7ddda2b 100644 --- a/src/code_node/analysis/reader.py +++ b/src/code_node/analysis/reader.py @@ -51,6 +51,7 @@ _doc_line, _ends_early, _is_main_guard, + _jump, _names_in, _negated, _parse, @@ -548,10 +549,14 @@ def _if(self, stmt: ast.If, frame: _Frame) -> list[CodeStep]: frame.guards.append(_negated(stmt.test)) orelse = self._block(stmt.orelse, frame) frame.guards.pop() - if orelse: - rest = [CodeStep(frame.file, stmt.orelse[0].lineno - 1, "else:", "else", steps=orelse)] - if head.notes or head.calls or body or rest: - return [self._step(stmt, frame, f"if {test}:", "if", head, body), *rest] + ends = _jump(stmt.orelse) + if orelse or ends: + rest = [CodeStep(frame.file, stmt.orelse[0].lineno - 1, "else:", "else", steps=orelse, ends=ends)] + ends = _jump(stmt.body) + if head.notes or head.calls or body or rest or ends: + step = self._step(stmt, frame, f"if {test}:", "if", head, body) + step.ends = ends + return [step, *rest] return [] def _with(self, stmt: Any, frame: _Frame) -> list[CodeStep]: @@ -628,8 +633,10 @@ def _try(self, stmt: Any, frame: _Frame) -> list[CodeStep]: if silent: code = " ".join([header] + [_unparse(s, 40) for s in h.body]) self._spot("swallow", frame.file, h.lineno, frame.function, code) - if hbody or head.notes: + ends = _jump(h.body) + if hbody or head.notes or ends: handlers.append(self._step(h, frame, header, "except", head, hbody)) + handlers[-1].ends = ends orelse = self._block(stmt.orelse, frame) final = self._block(stmt.finalbody, frame) if not (body or handlers or orelse or final): diff --git a/src/code_node/analysis/syntax.py b/src/code_node/analysis/syntax.py index f413c2f..cbcf8a8 100644 --- a/src/code_node/analysis/syntax.py +++ b/src/code_node/analysis/syntax.py @@ -57,6 +57,14 @@ def _ends_early(body: list[Any]) -> bool: return bool(body) and isinstance(body[-1], (ast.Return, ast.Raise, ast.Continue, ast.Break)) +def _jump(body: list[Any]) -> str: + """ + How a block leaves its loop, when its last line does: 'continue', 'break', or ''. + """ + last = body[-1] if body else None + return "continue" if isinstance(last, ast.Continue) else "break" if isinstance(last, ast.Break) else "" + + def _secret_spots(node: Any) -> list[tuple[str, Any]]: """ The secret-looking literals in a piece of code, with the name each is given to: a password=, token= or key given diff --git a/src/code_node/report/build.py b/src/code_node/report/build.py index 6cc573c..afac53c 100644 --- a/src/code_node/report/build.py +++ b/src/code_node/report/build.py @@ -1,5 +1,6 @@ """ -The page code-node draws for one entry point: tabs for the graph and the walk through its steps. +The page code-node draws for one entry point: tabs for the graph, the decisions it makes and the walk through its +steps. """ from __future__ import annotations @@ -12,6 +13,7 @@ from ..analysis.findings import _listed, _short_name from ..analysis.graph import CodeGraph, code_graph, everything_graph, graph_views from ..analysis.model import CodeMap +from .flow import _Flow, flow_entries from .graph import _Graph from .tabs import _Tabs from .walk import _Walk, walk_stops @@ -62,7 +64,7 @@ def page_blocks( editor: str = "", ) -> list[Any]: """ - The whole page for one entry point, as report blocks: tabs for the graph and the walk through its steps, with the + The whole page for one entry point, as report blocks: tabs for the graph, its decisions and the walk through its steps, with the entry point and the folder beside them. sources is the code read, {file: its text}: with it, a box of the graph clicked shows its code. editor is a link from editor_link(), to open the code at a line in an editor. """ @@ -111,6 +113,20 @@ def page_blocks( ], ) ) + event = next((n for n in graph.nodes if n.kind == "event"), None) + routes = any(route for _, route in flow_entries(cmap.steps)) # a web app: a flow for each route + flow = _Flow( + cmap.steps, + f"How {'each route' if routes else f'{entry}()'} decides what to do: each question it asks, and where each " + "answer goes" + if entry + else "", + comes_in=(event.label, event.sub) if event else ("", ""), + ) + decisions = sum( + 1 for top in cmap.steps for s in top.walk() if s.kind in ("if", "try") and not s.code.startswith("case ") + ) + tabs.append(("Decisions", _plural(decisions, "decision"), [flow])) tabs.append(("Step by step", _plural(len(walk_stops(walk)), "step"), [walk])) read = f"{_plural(cmap.files, 'file')}, {cmap.lines:,} lines, nothing run" if cmap.files else "" blocks.append( diff --git a/src/code_node/report/flow.py b/src/code_node/report/flow.py new file mode 100644 index 0000000..93890f7 --- /dev/null +++ b/src/code_node/report/flow.py @@ -0,0 +1,952 @@ +""" +_Flow: the decisions each call makes, as a flowchart drawn top to bottom. What comes in, then each line that does +something worth knowing, each `if` as a question with an arrow for each answer, each loop as a box with an arrow back +to its top, each `try` with the way taken when a line in it raises, and where the function ends. + +No JavaScript: one SVG for each flow. The path that goes on runs down the middle; an answer whose lines end early +(return, raise, skip to the next item, leave the loop) sits beside it, so the usual path reads as one line. A call to +one of the code's own functions is a box with what it reaches; one that makes decisions of its own is opened in place, +up to _OPEN_DEPTH calls deep and _OPEN_TALL high. One opened nowhere gets a flow of its own under the first, and its +box links to it. +""" + +from __future__ import annotations + +import html +import re +from dataclasses import dataclass, field +from typing import Callable + +from ..kit.blocks import _Report, _draws +from ..kit.fmt import _plural +from ..kit.text import _esc + +from ..analysis.findings import _short_name +from ..analysis.model import CodeStep +from .brands import AWS_LOGOS, SERVICE_LOGOS, fills, logo_for +from .graph import _g_mono, _g_sans +from .walk import _SAID_FIRST, _note_said + +_GAP = 30 # between boxes on a path +_SIDE = 56 # between the path down the middle and an answer beside it +_PAD = 28 # around the drawing +_BOX_LEAST, _BOX_MOST = 230, 340 # a box's width +_ROW = 21 # a line a box lists: what a call reaches +_ROWS_MOST = 4 # lines a call's box lists, at most +_END_MOST = 300 # an end's words, at most this wide +_QUESTION_MOST = 320 # a question's code, at most this wide + +_OPEN_DEPTH = 2 # calls opened in place, at most this many deep +_OPEN_TALL = 1400 # ...and at most this tall; a call opened nowhere gets a flow of its own +_FLOWS_MOST = 24 # flows drawn, at most: a web app's routes, then the calls opened nowhere + +_REACHES = ("aws-", "lib-", "http") # the tags of what a line reaches outside the code + +_KIND_OF = { # a tag -> the colour its box takes, as the graph's arrows do + "aws-reads": "reads", + "lib-reads": "reads", + "aws-writes": "writes", + "lib-writes": "writes", + "aws-sends": "sends", + "lib-sends": "sends", + "http": "sends", + "aws-runs": "runs", + "lib-runs": "runs", + "event": "event", + "env": "env", + "warn": "warn", +} + + +@dataclass +class _Flow: + """ + How each call decides what to do: steps are the map's (CodeMap.steps), one 'def' step for the entry point, or for + a web app the app with a 'def' step for each route. comes_in is what the entry point is called with, as the graph + says it: ('An S3 event', "what it's called with"). + """ + + steps: list[CodeStep] + title: str = "" + comes_in: tuple[str, str] = ("", "") + empty: str = "(no calls to draw)" + + +# ---------------------------------------------------------------- what's drawn + + +@dataclass +class _Box: + """ + A piece of a flow, drawn at any place: its size, where the path runs through it (middle, from its left), and + whether the code goes on past its bottom. + """ + + width: float + height: float + middle: float + parts: list[Callable[[float, float], str]] = field(default_factory=list) + goes_on: bool = True + leaving: str = "" # said on the arrow that leaves it: 'when every one is done' + + def draw(self, x: float, y: float) -> str: + return "".join(part(x, y) for part in self.parts) + + +_NOTHING = _Box(0, 0, 0) + + +def _clip(text: str, room: float, measure: Callable[[str], float]) -> str: + if measure(text) <= room: + return text + while text and measure(text + "…") > room: + text = text[:-1] + return text + "…" + + +def _mono(size: float) -> Callable[[str], float]: + return lambda text: _g_mono(text, size) + + +def _sans(size: float, bold: bool = False) -> Callable[[str], float]: + return lambda text: _g_sans(text, size, bold) + + +def _arrow(points: list[tuple[float, float]], css: str = "", head: bool = True) -> str: + """ + A line through the points, with an arrowhead at the last one. + """ + path = "M" + " L".join(f"{x:g} {y:g}" for x, y in points) + out = f'' + if not head: + return out + (x0, y0), (x1, y1) = points[-2], points[-1] + if abs(y1 - y0) >= abs(x1 - x0): + s = 1 if y1 > y0 else -1 + tip = f"M{x1 - 5:g} {y1 - 8 * s:g} L{x1:g} {y1:g} L{x1 + 5:g} {y1 - 8 * s:g}Z" + else: + s = 1 if x1 > x0 else -1 + tip = f"M{x1 - 8 * s:g} {y1 - 5:g} L{x1:g} {y1:g} L{x1 - 8 * s:g} {y1 + 5:g}Z" + return out + f'' + + +def _word(x: float, y: float, text: str, css: str) -> str: + """ + A few words on an arrow, on a pill of their own: 'yes', 'no', 'next order'. + """ + width = _g_sans(text, 11.5, True) + 10 + return ( + f'' + f'{_esc(text)}' + ) + + +def _join(x: float, y: float) -> str: + return f'' + + +def _stack(boxes: list[_Box]) -> _Box: + """ + Boxes one under the other, joined by arrows, as far as the first one the code doesn't go on past. + """ + boxes = [b for b in boxes if b.height] + for i, b in enumerate(boxes): + if not b.goes_on: + boxes = boxes[: i + 1] + break + if not boxes: + return _NOTHING + left = max(b.middle for b in boxes) + right = max(b.width - b.middle for b in boxes) + parts: list[Callable[[float, float], str]] = [] + y = 0.0 + for i, b in enumerate(boxes): + parts.append(lambda x, top, b=b, at=y: b.draw(x + left - b.middle, top + at)) + y += b.height + if i < len(boxes) - 1: + gap = _GAP + (8 if b.leaving else 0) + parts.append( + lambda x, top, at=y, gap=gap, said=b.leaving: ( + _arrow([(x + left, top + at), (x + left, top + at + gap)]) + + (_word(x + left + 8, top + at + gap / 2, said, "fwq") if said else "") + ) + ) + y += gap + return _Box(left + right, y, left, parts, boxes[-1].goes_on, boxes[-1].leaving) + + +def _tip(step: CodeStep | None) -> str: + return f"{_esc(f'{step.file} line {step.line}: {step.code}')}" if step is not None else "" + + +def _logo(name: str | None, x: float, y: float, size: float) -> str: + found = logo_for(name, size) if name else None + if not found: + return "" + width, height, body = found + body = body.replace('id="cnl-', 'id="cnf-').replace("url(#cnl-", "url(#cnf-") # the graph's, in a hidden tab + return ( + f'' + ) + + +def _logo_name(kind: str, text: str) -> str | None: + """ + The logo of what a tag reaches: 'aws-s3' for ('aws-reads', 'S3 GetObject'), 'redis' for ('lib-reads', 'Redis: GET'). + """ + if kind.startswith("aws-"): + service = text.rpartition(" ")[0] or text + return AWS_LOGOS.get(service.lower().replace(" ", "")) + if kind.startswith("lib-"): + return SERVICE_LOGOS.get(text.partition(":")[0].lower()) + return None + + +_BADGES = { # 26 x 26 line drawings, stroked in the box's colour + "event": "M14.5 5.5 L8.5 14 H13 L11.5 20.5 L17.5 12 H13Z", + "env": "M7 9h12M7 13h12M7 17h8", + "log": "M8 8h10v10H8zM10.5 11.5h5M10.5 14.5h3", + "code": "M10 9 L6.5 13 L10 17M16 9 L19.5 13 L16 17", +} + + +def _badge(badge: str, x: float, y: float) -> str: + """ + A box's badge at x, y: 'entry' (the entry point), 'call' (one of its own functions), 'logo:aws-s3', a line drawing + from _BADGES, or 'dot'. + """ + tile = f'' + if badge == "entry": + return ( + tile + f'' + ) + if badge == "call": + return tile + f'λ' + if badge.startswith("logo:"): + found = logo_for(badge[5:], 26) + if found and fills(found): + return _logo(badge[5:], x, y, 26) + return tile + _logo(badge[5:], x + 4, y + 4, 18) + if badge in _BADGES: + return tile + f'' + return tile + f'' + + +def _card( + title: str, + *, + sub: str = "", + rows: list[tuple[str, str]] | None = None, + badge: str = "dot", + css: str = "", + mono: bool = False, + foot: str = "", + link: str = "", + step: CodeStep | None = None, +) -> _Box: + """ + A box on the path: a badge, a title with a quiet line under it, and the lines it lists, (tag, what) each: what a + call reaches, or a warning. foot is a line at its bottom, a link to link's flow when given. + """ + rows = rows or [] + measure = _mono(13) if mono else _sans(13.5, True) + width = max( + _BOX_LEAST, + min( + _BOX_MOST, + max( + measure(title) + 60, + _g_sans(sub, 12) + 60, + max((_g_mono(text, 12) + 60 for _, text in rows), default=0), + _g_sans(foot, 12) + 32, + ), + ), + ) + head = 46 if sub else 40 + height = head + (8 + _ROW * len(rows) if rows else 0) + (26 if foot else 0) + + def draw(x: float, y: float) -> str: + out = [f'{_tip(step)}{_badge(badge, x + 10, y + (head - 26) / 2)}') + name = _esc(_clip(title, width - 58, measure)) + if sub: + out.append(f'{name}') + out.append( + f'{_esc(_clip(sub, width - 58, _sans(12)))}' + ) + else: + out.append( + f'{name}' + ) + if rows: + out.append(f'') + for i, (kind, text) in enumerate(rows): + row = y + head + 8 + _ROW * i + 14 + logo = _logo(_logo_name(kind, text), x + 14, row - 11, 14) + out.append( + logo + or f'' + ) + out.append(f'{_esc(_clip(text, width - 48, _mono(12)))}') + if foot: + said = f'{_esc(foot)}' + out.append(f'{said}' if link else said) + out.append("") + return "".join(out) + + return _Box(width, height, width / 2, [draw]) + + +def _end(text: str, css: str, glyph: str, step: CodeStep | None = None) -> _Box: + """ + Where a path ends: the function returns or raises, or the loop goes on to its next item or stops. + """ + text = _clip(text, _END_MOST, _mono(12.5)) + width = _g_mono(text, 12.5) + 48 + marks = { + "stop": lambda x, y: f'', + "back": lambda x, y: ( + f'' + ), + "loop": lambda x, y: f'', + "out": lambda x, y: ( + f'' + ), + "raise": lambda x, y: ( + f'' + f'' + ), + } + + def draw(x: float, y: float) -> str: + return ( + f'{_tip(step)}' + f'{marks[glyph](x, y)}{_esc(text)}' + ) + + return _Box(width, 32, width / 2, [draw], goes_on=False) + + +_LOOP = "M12.4 9.3A4.6 4.6 0 1 1 11.3 4.7M12.1 2.3v2.8H9.3" # the graph's loop mark + + +def _question(text: str, step: CodeStep | None) -> _Box: + """ + A decision: the test, in a box pointed at both ends. + """ + text = _clip(text, _QUESTION_MOST, _mono(13)) + width, height = _g_mono(text, 13) + 56, 44 + + def draw(x: float, y: float) -> str: + m = height / 2 + shape = ( + f"M{x:g} {y + m:g} L{x + 18:g} {y:g} H{x + width - 18:g} L{x + width:g} {y + m:g} " + f"L{x + width - 18:g} {y + height:g} H{x + 18:g}Z" + ) + return ( + f'{_tip(step)}{_esc(text)}' + ) + + return _Box(width, height, width / 2, [draw]) + + +def _frame(body: _Box, title: str, css: str, *, sub: str = "", back: str = "", step: CodeStep | None = None) -> _Box: + """ + A box around a path: a loop, with an arrow from the end of its lines back to its top saying back ('next + order'), a try, an else, or a call opened in place. + """ + head = 46 if sub else 40 + room = _g_sans(title, 13, True) + 60 if css != "fcall" else _g_mono(title, 13) + 60 + left = 44 if back else 22 + width = max(body.width + left + 22, room, _BOX_LEAST + 40) + middle = left + body.middle + if body.width + left + 22 < width: # the lines in the middle of a box wider than they are + middle = max(middle, width / 2) + below = 40 if back and body.goes_on else 20 + height = head + _GAP + body.height + below + marks = { + "floop": f'', + "fcall": 'λ', + "ftry": '', + "felse": '', + } + + def draw(x: float, y: float) -> str: + out = [f'{_tip(step)}') + out.append(marks[css].replace("{x}", f"{x + 14:g}").replace("{y}", f"{y + 12:g}")) + mono = " mono" if css == "fcall" else "" + out.append(f'{_esc(title)}') + if sub: + out.append(f'{_esc(sub)}') + out.append(f'') + out.append(_arrow([(x + middle, y + head), (x + middle, y + head + _GAP)])) + out.append(body.draw(x + middle - body.middle, y + head + _GAP)) + if back and body.goes_on: + bottom = y + head + _GAP + body.height + lane = x + 18 + out.append( + _arrow( + [(x + middle, bottom), (x + middle, bottom + 18), (lane, bottom + 18), (lane, y + head + 1)], + "back", + ) + ) + out.append(_word(lane + 6, bottom + 18, back, "fwl")) + return "".join(out) + + return _Box(width, height, middle, [draw]) + + +# ---------------------------------------------------------------- the code, as boxes + + +def _plain(markup: str) -> str: + return html.unescape(re.sub(r"<[^>]+>", "", markup)) + + +def _reaches(step: CodeStep) -> list[tuple[str, str]]: + """ + What a step and every step under it reach outside the code, and the warnings about them, once each. + """ + found: list[tuple[str, str]] = [] + for s in step.walk(): + for kind, text in s.notes: + if kind.startswith(_REACHES) or kind == "warn": + if (kind, text) not in found: + found.append((kind, text)) + return found + + +def _decisions(step: CodeStep) -> int: + """ + The decisions made under a step: its ifs (a match counts once), loops and trys. + """ + return sum( + 1 for s in step.walk() if s is not step and s.kind in ("if", "loop", "try") and not s.code.startswith("case ") + ) + + +@dataclass +class _Context: + """ + Where a flow's drawing is: the loops around (their item, innermost last) and how many calls deep. anchor gives + a call opened nowhere a flow of its own, and the id of it to link to. + """ + + anchor: Callable[[CodeStep], str] + loops: list[str] = field(default_factory=list) + depth: int = 0 + + +def _call_box(call: CodeStep, context: _Context) -> _Box: + """ + A call to one of its own functions: opened in place when it makes decisions and there's room, else a box with + what it reaches. + """ + name = f"{_short_name(call.name)}()" + made = _decisions(call) + if made and not call.seen and context.depth < _OPEN_DEPTH: + inside = _Context(context.anchor, list(context.loops), context.depth + 1) + body = _stack(_boxes(call.steps, inside)) + opened = _frame(body, name, "fcall", sub=f"{call.file} line {call.line}", step=call) + if opened.height <= _OPEN_TALL: + return opened + rows = _reaches(call) + if len(rows) > _ROWS_MOST: + rows = rows[: _ROWS_MOST - 1] + [("more", f"+{len(rows) - _ROWS_MOST + 1} more")] + link = context.anchor(call) if made else "" + foot = f"◇ {_plural(made, 'decision')} inside · its flow ↓" if link else "" + sub = f"{call.file} line {call.line}" + (", again" if call.seen else "") + return _card(name, sub=sub, rows=rows, badge="call", css="k-call", mono=True, foot=foot, link=link, step=call) + + +def _taken(step: CodeStep) -> list[tuple[str, str]]: + """ + The names a line takes from what comes in or the environment, as (name, where): ('bucket', 'event'). + """ + if step.kind != "do" or step.calls or any(k.startswith(_REACHES) for k, _ in step.notes): + return [] + words = {what.split(" ", 1)[0] for _, what in step.holds} + if not step.holds or len(words) != 1 or not words <= {"event", "request", "env"}: + return [] + return [(name, what.split(" ", 1)[0]) for name, what in step.holds] + + +def _line_box(step: CodeStep, context: _Context) -> _Box: + """ + A line that does something: its calls into its own functions, else what it reaches, logs or takes. + """ + own = [call for call in step.calls if call.kind == "def"] + reach = next(((k, t) for k, t in step.notes if k in _SAID_FIRST), None) + if own: + boxes = [_call_box(call, context) for call in own] + if reach and reach[0].startswith(_REACHES): + boxes.append(_reach_box(step, *reach)) + return _stack(boxes) + if reach: + return _reach_box(step, *reach) + for kind, text in step.notes: + if kind in ("log", "warn"): + return _card(_plain(_note_said(kind, text)), badge="log", css="k-quiet", step=step) + for name, what in step.holds: + word = what.split(" ", 1)[0] + if word in ("event", "request"): + return _card(f"Takes {name} from the {word}", badge="event", css="k-event", step=step) + if word == "env": + return _card(f"Takes {name} from the environment", badge="env", css="k-env", step=step) + for kind, text in step.notes: + if kind in ("event", "env"): + return _card(_plain(_note_said(kind, text)), badge=kind, css=f"k-{kind}", step=step) + if step.steps: # a class, its body run where it's defined + return _stack(_boxes(step.steps, context)) + return _card(step.code, badge="code", css="k-quiet", mono=True, step=step) + + +def _reach_box(step: CodeStep, kind: str, text: str) -> _Box: + name = _logo_name(kind, text) + warnings = [(k, t) for k, t in step.notes if k == "warn"] + return _card( + _plain(_note_said(kind, text)), + sub=f"line {step.line}", + rows=warnings, + badge=f"logo:{name}" if name and logo_for(name, 26) else "dot", + css=f"k-{_KIND_OF.get(kind, 'quiet')}", + step=step, + ) + + +def _leaves(step: CodeStep, context: _Context) -> _Box: + """ + The end of the lines under an if, an else or an except that leave the loop. + """ + if step.ends == "continue": + item = context.loops[-1] if context.loops else "" + return _end(f"skips to the next {item}" if item else "goes round again", "fskip", "loop", step) + if step.ends == "break": + return _end("leaves the loop", "fskip", "out", step) + return _NOTHING + + +def _answer(step: CodeStep | None, context: _Context) -> _Box: + if step is None: + return _NOTHING + return _stack([*_boxes(step.steps, context), _leaves(step, context)]) + + +def _boxes(steps: list[CodeStep], context: _Context) -> list[_Box]: + """ + A function's lines, or a block's, as boxes one under the other. + """ + out: list[_Box] = [] + i = 0 + while i < len(steps): + step = steps[i] + taken = _taken(step) + if taken: # lines in a row that only take names from what comes in: one box + where = taken[0][1] + while i + 1 < len(steps) and {w for _, w in _taken(steps[i + 1])} == {where}: + i += 1 + taken += _taken(steps[i]) + names = ", ".join(dict.fromkeys(name for name, _ in taken)) + said = f"Takes {names} from the environment" if where == "env" else f"Takes {names} from the {where}" + out.append( + _card( + said, + badge="env" if where == "env" else "event", + css=f"k-{'env' if where == 'env' else 'event'}", + step=step, + ) + ) + elif step.kind == "if" and step.code.startswith("match "): + out.append(_match_box(step, context)) + elif step.kind == "if": + chain = [step] + while i + 1 < len(steps) and steps[i + 1].kind == "if" and steps[i + 1].code.startswith("elif "): + i += 1 + chain.append(steps[i]) + otherwise = None + if i + 1 < len(steps) and steps[i + 1].kind == "else": + i += 1 + otherwise = steps[i] + out.append(_decision_box([(_test(s.code), s) for s in chain], otherwise, context)) + elif step.kind == "try": + handlers = [] + while i + 1 < len(steps) and steps[i + 1].kind == "except": + i += 1 + handlers.append(steps[i]) + out.append(_try_box(step, handlers, context)) + elif step.kind == "else": + after = steps[i - 1].kind if i else "" + title = ( + "Finally, whatever happened above" + if step.code.startswith("finally") + else "If nothing in the try raised" + if after in ("try", "except") + else "If the loop ran to its end, with no break" + ) + out.append(_frame(_answer(step, context), title, "felse", step=step)) + elif step.kind == "loop": + out.append(_loop_box(step, context)) + elif step.kind == "with": + out.append( + _stack([*[_call_box(c, context) for c in step.calls if c.kind == "def"], *_boxes(step.steps, context)]) + ) + elif step.kind == "return": + out.append(_return_box(step, context)) + elif step.kind == "raise": + what = step.code.removeprefix("raise").strip() + out.append(_end(f"Raises {what}" if what else "Raises the error again", "fraise", "raise", step)) + elif step.kind in ("do", "def"): + out.append(_line_box(step, context)) + i += 1 + return out + + +def _test(code: str) -> str: + found = re.fullmatch(r"(?:el)?if (.+):", code.strip()) + return found.group(1) if found else code + + +def _return_box(step: CodeStep, context: _Context) -> _Box: + what = step.code.removeprefix("return").strip() or "None" + calls = [_call_box(c, context) for c in step.calls if c.kind == "def"] + if context.depth: # the end of a call opened in place: back to its caller + end = _end(f"returns {what}", "fback", "back", step) + else: + end = _end(f"Returns {what}", "fret", "stop", step) + return _stack([*calls, end]) + + +def _loop_box(step: CodeStep, context: _Context) -> _Box: + found = re.fullmatch(r"(?:async )?for (.+?) in (.+):", step.code) + if found: + item, items = found.groups() + title, back, leaving = f"For each {item} in {items}", f"next {item}", "when every one is done" + else: + item, title, back, leaving = ( + "", + f"While {step.code.removeprefix('while ').rstrip(':')}", + "again", + "when it stops", + ) + before = [_call_box(c, context) for c in step.calls if c.kind == "def"] # what gives it its items: once + context.loops.append(item) + body = _stack(_boxes(step.steps, context)) + context.loops.pop() + box = _frame(body, title, "floop", back=back, step=step) + box.leaving = leaving + return _stack([*before, box]) if before else box + + +def _decision_box(chain: list[tuple[str, CodeStep]], otherwise: CodeStep | None, context: _Context) -> _Box: + """ + A question and its two answers: the one that goes on down the middle, the other beside it. An elif is the next + question, on the no path; `if not x` is asked as x, so yes stays the usual answer. + """ + test, step = chain[0] + yes = _answer(step, context) + if len(chain) > 1: + no = _decision_box(chain[1:], otherwise, context) + else: + no = _answer(otherwise, context) + if test.startswith("not ") and len(chain) == 1: + test, yes, no = test[4:], no, yes + said = ("no", "yes") + else: + said = ("yes", "no") + yes_word, no_word = said + # the answer that ends early, or the only one with lines, goes beside the path; else the next question stays + # in the middle, and yes does when there's none + if not no.height or (yes.height and not yes.goes_on and no.goes_on) or (len(chain) > 1 and yes.goes_on): + down, side, down_word, side_word = no, yes, no_word, yes_word + else: + down, side, down_word, side_word = yes, no, yes_word, no_word + if not side.height: + down, side, down_word, side_word = side, down, side_word, down_word + question = _question(test + "?", step) + half = question.width / 2 + right_of_down = max(down.width - down.middle, half) if down.height else half + side_x = right_of_down + _SIDE + side.middle # the side path, from the middle + left = max(question.middle, down.middle if down.height else 0) + right = max(half, side_x + side.width - side.middle) + top = question.height + _GAP + 6 + down_on = down.goes_on or not down.height + goes_on = down_on or side.goes_on + meet = max(top + down.height, top + side.height) + (18 if side.goes_on else 0) # clear of both + height = meet + + def draw(x: float, y: float) -> str: + c = x + left + out = [question.draw(c - half, y)] + if down.height: + out.append(_arrow([(c, y + question.height), (c, y + top)])) + out.append(down.draw(c - down.middle, y + top)) + if down_on and goes_on: + out.append( + _arrow([(c, y + (top + down.height if down.height else question.height)), (c, y + height)], head=False) + ) + out.append(_word(c + 8, y + question.height + 14, down_word, f"fw{down_word}")) + sx = c + side_x + out.append(_arrow([(c + half, y + question.height / 2), (sx, y + question.height / 2), (sx, y + top)])) + out.append(_word(c + half + 8, y + question.height / 2 - 12, side_word, f"fw{side_word}")) + out.append(side.draw(sx - side.middle, y + top)) + if side.goes_on: + out.append(_arrow([(sx, y + top + side.height), (sx, y + meet), (c + 6, y + meet)])) + out.append(_join(c, y + meet)) + return "".join(out) + + return _Box(left + right, height, left, [draw], goes_on) + + +def _match_box(step: CodeStep, context: _Context) -> _Box: + """ + A match: each case a question on the no path of the one before. + """ + subject = step.code.removeprefix("match ").rstrip(":") + cases = [(f"{subject} is {case.code.removeprefix('case ').rstrip(':')}", case) for case in step.steps] + if not cases: + return _NOTHING + if cases[-1][0].endswith(" is _"): # the case for anything else + last = cases.pop()[1] + return _decision_box(cases, last, context) if cases else _answer(last, context) + return _decision_box(cases, None, context) + + +def _try_box(step: CodeStep, handlers: list[CodeStep], context: _Context) -> _Box: + """ + A try: its lines in a box, and beside it what runs when one of them raises, for the first except. + """ + box = _frame(_stack(_boxes(step.steps, context)), "Tries", "ftry", step=step) + if not handlers: + return box + handler = handlers[0] + found = re.fullmatch(r"except (.+?)(?: as \w+)?:", handler.code) + said = f"if it raises {found.group(1)}" if found else "if anything raises" + if len(handlers) > 1: + said += f" (+{len(handlers) - 1} more)" + side = _answer(handler, context) + if not side.height: + side = _end("drops the error", "fwarn", "raise", handler) + side.goes_on = True + side_x = box.width - box.middle + _SIDE + side.middle + meet = max(box.height, 70 + side.height) + (18 if side.goes_on else 0) + goes_on = box.goes_on or side.goes_on + + def draw(x: float, y: float) -> str: + c = x + box.middle + out = [box.draw(x, y)] + sx = c + side_x + out.append(_arrow([(x + box.width, y + 20), (sx, y + 20), (sx, y + 70)], "fail")) + out.append(_word(x + box.width + 8, y + 8, said, "fwf")) + out.append(side.draw(sx - side.middle, y + 70)) + if box.goes_on: + out.append(_arrow([(c, y + box.height), (c, y + meet)], head=False)) + if side.goes_on: + out.append(_arrow([(sx, y + 70 + side.height), (sx, y + meet), (c + 6, y + meet)])) + out.append(_join(c, y + meet)) + return "".join(out) + + return _Box(box.middle + side_x + side.width - side.middle, meet, box.middle, [draw], goes_on) + + +# ---------------------------------------------------------------- the flows + + +def flow_entries(steps: list[CodeStep]) -> list[tuple[CodeStep, str]]: + """ + The functions each call starts in, with the route each is for: the entry point, or a web app's routes. + """ + if len(steps) == 1 and steps[0].steps and all(s.kind == "def" for s in steps[0].steps): + return [(s, next((t for k, t in s.notes if k == "route"), "")) for s in steps[0].steps if not s.seen] + return [(s, "") for s in steps if s.kind == "def"] + + +def _flow_svg(entry: CodeStep, route: str, comes_in: tuple[str, str], context: _Context, first: bool) -> str: + boxes = [] + if route: + boxes.append(_card(route, sub="a request comes in", badge="event", css="k-event", mono=True)) + elif first and comes_in[0]: + boxes.append(_card(comes_in[0], sub=comes_in[1], badge="event", css="k-event")) + name = f"{_short_name(entry.name)}()" + css = "k-entry" if first else "k-call" + boxes.append( + _card( + name, + sub=f"{entry.file} line {entry.line}", + badge="entry" if first else "call", + css=css, + mono=True, + step=entry, + ) + ) + boxes += _boxes(entry.steps, context) + body = _stack(boxes) + if body.goes_on: + end = "Returns None" if first else "returns None" + body = _stack([body, _end(end, "fret" if first else "fback", "stop" if first else "back")]) + width, height = body.width + 2 * _PAD, body.height + 2 * _PAD + return ( + f'{body.draw(_PAD, _PAD)}' + ) + + +_KEY = ( + '
' + '' + 'a decision: yes and no on its arrows' + '' + "a loop: the arrow goes back for the next item" + '' + "where the call ends" + '' + "it raises" + 'when a line raises' + "
" +) + + +@_draws(_Flow, "html") +def _flow_html(block: _Flow, report: _Report) -> str: + made = report.shared("flows", lambda: [0]) # flows drawn so far, so each has ids of its own + group = f"cnd-flow{made[0]}" + made[0] += 1 + entries = flow_entries(block.steps) + if not entries: + return f"

{_esc(block.empty)}

" + own: dict[str, str] = {} # a call opened nowhere -> the id of its flow + later: list[CodeStep] = [] + + def anchor(call: CodeStep) -> str: + if call.name not in own: + own[call.name] = f"{group}-{len(own)}" + later.append(call) + return own[call.name] + + out = [f"

{_esc(block.title)}

" if block.title else ""] + shown = 0 + for i, (entry, route) in enumerate(entries[:_FLOWS_MOST]): + svg = _flow_svg(entry, route, block.comes_in, _Context(anchor), first=True) + head = ( + f'

{_esc(_short_name(entry.name))}(){f" · {_esc(route)}" if route else ""}

' + ) + out.append(f'
{head if len(entries) > 1 else ""}
{svg}
') + shown += 1 + done = 0 + while done < len(later) and shown < _FLOWS_MOST: # the calls opened nowhere, each with a flow of its own + call = later[done] + done += 1 + svg = _flow_svg(call, "", block.comes_in, _Context(anchor, depth=1), first=False) + out.append( + f'

Inside {_esc(_short_name(call.name))}()' + f' {_esc(call.file)} line {call.line}

{svg}
' + ) + shown += 1 + left = len(entries) - min(len(entries), _FLOWS_MOST) + len(later) - done + if left: + out.append(f'

And {_plural(left, "more flow")}, left out to keep the page light.

') + return "".join(out) + _KEY + + +# ---------------------------------------------------------------- in text + + +def _said(steps: list[CodeStep]) -> str: + """ + Where an answer goes, in a few words: its first line that does something, then how it ends. + """ + words = [] + for step in steps: + if step.kind in ("do", "def"): + own = [c for c in step.calls if c.kind == "def"] + reach = next(((k, t) for k, t in step.notes if k in _SAID_FIRST or k == "log"), None) + if own: + words.append("calls " + ", ".join(f"{_short_name(c.name)}()" for c in own)) + elif reach: + said = _plain(_note_said(*reach)) + words.append(said[:1].lower() + said[1:]) + elif step.kind == "return": + words.append(f"returns {step.code.removeprefix('return').strip() or 'None'}") + elif step.kind == "raise": + words.append(f"raises {step.code.removeprefix('raise').strip() or 'the error again'}") + elif step.kind in ("if", "loop", "try"): + words.append("decides more (below)") + return ", then ".join(words[:3]) + (" …" if len(words) > 3 else "") + + +_ENDS = {"continue": "skips to the next item", "break": "leaves the loop"} + + +def _answer_said(step: CodeStep | None) -> str: + if step is None: + return "goes on" + said = _said(step.steps) + ends = _ENDS.get(step.ends, "") + return ", then ".join(filter(None, [said, ends])) or "goes on" + + +def _text_lines(steps: list[CodeStep], pad: int, out: list[str], answered: bool = False) -> None: + """ + Each decision under steps as a line saying where each answer goes, its loops, trys and ends, indented as the code + is; the calls that make decisions of their own after the line calling them. answered: steps are an answer's, + whose end its question's line said. + """ + indent = " " * pad + i = 0 + while i < len(steps): + step = steps[i] + under: list[CodeStep] = [step] + if step.kind == "if" and not step.code.startswith("match "): + chain = [step] + while i + 1 < len(steps) and steps[i + 1].kind == "if" and steps[i + 1].code.startswith("elif "): + i += 1 + chain.append(steps[i]) + otherwise = None + if i + 1 < len(steps) and steps[i + 1].kind == "else": + i += 1 + otherwise = steps[i] + for k, asked in enumerate(chain): + test, yes = _test(asked.code), _answer_said(asked) + no = "asks the next question" if k + 1 < len(chain) else _answer_said(otherwise) + if test.startswith("not ") and len(chain) == 1: + test, yes, no = test[4:], no, yes + out.append(f"{asked.line:>5} {indent}{test}? yes: {yes} · no: {no}") + under = [*chain, *([otherwise] if otherwise else [])] + elif step.kind == "if": + out.append(f"{step.line:>5} {indent}{step.code.rstrip(':')}") + elif step.kind == "loop": + out.append(f"{step.line:>5} {indent}{step.code.rstrip(':')} (back to its top for each)") + elif step.kind == "try": + out.append(f"{step.line:>5} {indent}tries:") + elif step.kind == "except": + out.append(f"{step.line:>5} {indent}{step.code.rstrip(':')}: {_answer_said(step)}") + elif step.kind == "return" and not answered: + out.append(f"{step.line:>5} {indent}returns {step.code.removeprefix('return').strip() or 'None'}") + elif step.kind == "raise" and not answered: + out.append(f"{step.line:>5} {indent}raises {step.code.removeprefix('raise').strip() or 'the error again'}") + for each in under: + for call in each.calls: + if call.kind == "def" and _decisions(call) and not call.seen: + out.append(f"{each.line:>5} {indent} ↳ {_short_name(call.name)}() {call.file} line {call.line}") + _text_lines(call.steps, pad + 4, out) + if each.kind != "do": + _text_lines(each.steps, pad + 2, out, answered=each.kind in ("if", "else", "except")) + i += 1 + + +@_draws(_Flow, "text") +def _flow_lines(block: _Flow, report: _Report) -> list[str]: + out = ["", f"-- {block.title} --"] if block.title else [""] + entries = flow_entries(block.steps) + if not entries: + return out + [block.empty] + for entry, route in entries[:_FLOWS_MOST]: + out.append(f" {_short_name(entry.name)}() {entry.file} line {entry.line}{f' ({route})' if route else ''}") + if not _decisions(entry): + out.append(" (no decisions: one path from start to end)") + continue + _text_lines(entry.steps, 0, out) + return out diff --git a/src/code_node/report/outline.py b/src/code_node/report/outline.py index f816dd3..e288247 100644 --- a/src/code_node/report/outline.py +++ b/src/code_node/report/outline.py @@ -210,7 +210,8 @@ def walk(steps: list[CodeStep], pad: int, called: bool) -> None: ) walk(step.steps, pad + (4 if called else 2), False) continue - out.append(f"{step.line or '':>5} {' ' * pad}{step.code}{comment(step.notes, step=step)}") + ends = f"then {step.ends}" if step.ends else "" + out.append(f"{step.line or '':>5} {' ' * pad}{step.code}{comment(step.notes, ends, step=step)}") for call in step.calls: walk([call], pad + 4, True) walk(step.steps, pad + 2, False) diff --git a/src/code_node/report/style.py b/src/code_node/report/style.py index 51e882d..2d722d2 100644 --- a/src/code_node/report/style.py +++ b/src/code_node/report/style.py @@ -443,6 +443,55 @@ .cnd .walk .wfacts code{font:12.5px var(--mono)} .cnd .walk :is(.wfm,.wfv){opacity:.6} .cnd .walk .wthen{margin-top:10px;font-size:12px;opacity:.6} +.cnd .cdf{margin:6px 0 18px;--fe:rgba(127,127,127,.85);--fl:light-dark(#7c3aed,#a78bfa);--fa:light-dark(#d97706,#f59e0b);--fx:light-dark(#dc2626,#f87171);--fbg:var(--jp-layout-color0,var(--vscode-editor-background,Canvas))} +.cnd .cdf .fhd{margin:0 0 6px;font-size:13.5px}.cnd .cdf .fhd small{opacity:.6;margin-left:6px} +.cnd .cdfc{width:fit-content;max-width:100%;overflow:auto;border:1px solid rgba(127,127,127,.2);border-radius:14px;background-color:rgba(127,127,127,.035);background-attachment:local;background-size:18px 18px; + background-image:radial-gradient(circle at 1px 1px,rgba(127,127,127,.3) 1px,transparent 0);background-position:9px 9px} +.cnd .cdfc>svg{display:block;max-width:none;height:auto;font-family:-apple-system,BlinkMacSystemFont,"Segoe UI",Roboto,Helvetica,Arial,sans-serif} +.cnd .cdf svg text{fill:currentColor} +.cnd .cdf :is(.mono,.fr,.fp text,.fq text,.ff){font-family:var(--mono)} +.cnd .cdf .fe{fill:none;stroke:var(--fe);stroke-width:1.6;stroke-linejoin:round}.cnd .cdf .fh{fill:var(--fe)} +.cnd .cdf .fe.back{stroke:var(--fl);stroke-dasharray:5 4}.cnd .cdf .fh.back{fill:var(--fl)} +.cnd .cdf .fe.fail{stroke:var(--fa);stroke-dasharray:5 4}.cnd .cdf .fh.fail{fill:var(--fa)} +.cnd .cdf .fj{fill:var(--fe)} +.cnd .cdf .fw rect{fill:var(--fbg);stroke:rgba(127,127,127,.3)}.cnd .cdf .fw text{font-size:11.5px;font-weight:650;fill-opacity:.65} +.cnd .cdf :is(.fwyes,.fwno,.fwl,.fwf) text{fill-opacity:1} +.cnd .cdf .fwyes text,.cnd .fkey .fwyes{fill:light-dark(#15803d,#4ade80);color:light-dark(#15803d,#4ade80)} +.cnd .cdf .fwno text,.cnd .fkey .fwno{fill:light-dark(#b91c1c,#f87171);color:light-dark(#b91c1c,#f87171)} +.cnd .cdf .fwl text{fill:var(--fl)}.cnd .cdf .fwf text{fill:var(--fa)} +.cnd .cdf .fbg{fill:var(--fbg);stroke:rgba(127,127,127,.32);filter:drop-shadow(0 1px 1.5px rgba(0,0,0,.07)) drop-shadow(0 6px 16px rgba(15,23,42,.06))} +.cnd .cdf .fbd{fill:var(--t,rgba(127,127,127,.12));stroke:var(--c,rgba(127,127,127,.5));stroke-opacity:.22} +.cnd .cdf .fbi{fill:none;stroke:var(--c,currentColor);stroke-width:1.6;stroke-linecap:round;stroke-linejoin:round}.cnd .cdf .fbi.fill{fill:var(--c,currentColor);stroke:none} +.cnd .cdf .fbl,.cnd .cdf .fgt{font:600 15px Georgia,serif} +.cnd .cdf .ft{font-size:13.5px;font-weight:650}.cnd .cdf .ft.mono{font-size:13px} +.cnd .cdf .fs{font-size:12px;fill-opacity:.62}.cnd .cdf .fr{font-size:12px;fill-opacity:.88} +.cnd .cdf .ff{font-size:11.5px;fill:light-dark(#2563eb,#60a5fa)}.cnd .cdf a:hover .ff{text-decoration:underline} +.cnd .cdf .fl{stroke:rgba(127,127,127,.2);fill:none} +.cnd .cdf .fdot{fill:var(--c,rgba(127,127,127,.6))} +.cnd .cdf .k-event{--c:#16a34a;--t:rgba(22,163,74,.12)}.cnd .cdf .k-env{--c:#0d9488;--t:rgba(13,148,136,.13)} +.cnd .cdf .k-reads{--c:#3b82f6;--t:rgba(59,130,246,.11)}.cnd .cdf .k-writes{--c:#ea580c;--t:rgba(234,88,12,.11)} +.cnd .cdf .k-sends{--c:#db2777;--t:rgba(219,39,119,.1)}.cnd .cdf .k-runs{--c:#9333ea;--t:rgba(147,51,234,.1)} +.cnd .cdf .k-warn{--c:#f59e0b}.cnd .cdf .k-quiet{--c:#64748b} +.cnd .cdf .k-entry .fbg{stroke:currentColor;stroke-opacity:.9;stroke-width:1.5}.cnd .cdf .k-entry .fbd{fill:currentColor;stroke:none}.cnd .cdf .k-entry .fbi.fill{fill:var(--fbg)} +.cnd .cdf .fq path{fill:light-dark(#fffbeb,rgba(245,158,11,.1));stroke:var(--fa);stroke-width:1.5}.cnd .cdf .fq text{font-size:13px;font-weight:600} +.cnd .cdf .fx .fxb{fill:rgba(127,127,127,.04);stroke:rgba(127,127,127,.5);stroke-dasharray:6 4} +.cnd .cdf .fxt{font-size:13px;font-weight:650}.cnd .cdf .fgl{fill:none;stroke:currentColor;stroke-width:1.7;stroke-linecap:round;stroke-linejoin:round}.cnd .cdf .fgl.fill{fill:currentColor;stroke:none} +.cnd .cdf .floop .fxb{fill:light-dark(rgba(124,58,237,.05),rgba(167,139,250,.07));stroke:var(--fl);stroke-opacity:.5}.cnd .cdf .floop .fgl{stroke:var(--fl)} +.cnd .cdf .ftry .fxb{fill:rgba(217,119,6,.04);stroke:var(--fa);stroke-opacity:.6}.cnd .cdf .ftry .fgl{stroke:var(--fa)} +.cnd .cdf .felse .fgl{stroke-opacity:.6} +.cnd .cdf .fp rect{fill:var(--fbg);stroke:rgba(127,127,127,.45);stroke-width:1.3}.cnd .cdf .fp text{font-size:12.5px} +.cnd .cdf .fret rect{fill:currentColor;stroke:none}.cnd .cdf .fret text{fill:var(--fbg)}.cnd .cdf .fret .fgl.fill{fill:var(--fbg)} +.cnd .cdf .fraise rect{fill:light-dark(rgba(220,38,38,.07),rgba(248,113,113,.1));stroke:var(--fx)}.cnd .cdf .fraise text{fill:var(--fx)}.cnd .cdf .fraise .fgl.fill{fill:var(--fx)} +.cnd .cdf .fwarn rect{fill:rgba(245,158,11,.08);stroke:var(--fa)}.cnd .cdf .fwarn text{fill:var(--fa)}.cnd .cdf .fwarn .fgl.fill{fill:var(--fa)} +.cnd .cdf .fgx{stroke:#fff;stroke-width:2;stroke-linecap:round} +.cnd .cdf .fskip rect{fill:light-dark(rgba(124,58,237,.05),rgba(167,139,250,.08));stroke:var(--fl);stroke-dasharray:4 3}.cnd .cdf .fskip text{fill:var(--fl)}.cnd .cdf .fskip .fgl{stroke:var(--fl)} +.cnd .fmore{opacity:.7;font-size:13px} +.cnd .fkey{display:flex;flex-wrap:wrap;gap:8px 18px;margin:2px 0 8px;font-size:12.5px;align-items:center;opacity:.85} +.cnd .fkey svg{vertical-align:middle;margin-right:6px}.cnd .fkey b{font-weight:650} +.cnd .fkey .kq{fill:light-dark(#fffbeb,rgba(245,158,11,.12));stroke:light-dark(#d97706,#f59e0b);stroke-width:1.4} +.cnd .fkey .kl{fill:none;stroke:light-dark(#7c3aed,#a78bfa);stroke-dasharray:4 3} +.cnd .fkey .kr{fill:currentColor}.cnd .fkey .kx{fill:light-dark(rgba(220,38,38,.08),rgba(248,113,113,.1));stroke:light-dark(#dc2626,#f87171)} +.cnd .fkey .kf{stroke:light-dark(#d97706,#f59e0b);stroke-dasharray:4 3;stroke-width:1.6} .cnd .walk .wf{display:none;position:absolute;left:var(--lw);right:0;top:var(--ch);bottom:0;flex-direction:column;border-left:1px solid rgba(127,127,127,.2);background:var(--bg)} .cnd .walk .wfn>.wr:checked~.wf{display:flex;animation:cnd-fade .25s ease} .cnd .walk .wfh{animation:cnd-in .25s ease} diff --git a/src/code_node/report/walk.py b/src/code_node/report/walk.py index 0a092eb..ca750fa 100644 --- a/src/code_node/report/walk.py +++ b/src/code_node/report/walk.py @@ -46,6 +46,11 @@ "raise": "Raises an error: the function stops here, and so does each caller that doesn't catch it.", } +_ENDS_SAYS = { # how the lines under it end, when they leave the loop: CodeStep.ends + "continue": "Then it skips the rest of the loop, on to the next item.", + "break": "Then it leaves the loop.", +} + _AWS_DOES = { # an AWS call, by what it does, as a sentence: 'Reads from S3 (GetObject)' "reads": "Reads from {}", "writes": "Writes to {}", @@ -370,6 +375,8 @@ def _says( ) elif step.kind in _KIND_SAYS: said.append(_KIND_SAYS[step.kind]) + if step.ends: + said.append(_ENDS_SAYS[step.ends]) own = [call for call in step.calls if call.kind == "def"] if own: it = "it" if len(own) == 1 else "each" diff --git a/tests/snapshots/books-api.txt b/tests/snapshots/books-api.txt index 728da66..f5f092a 100644 --- a/tests/snapshots/books-api.txt +++ b/tests/snapshots/books-api.txt @@ -90,6 +90,19 @@ flowchart LR classDef warn stroke:#f59e0b,stroke-width:2px class f3,f4,f7,lib_sql_postgresql,http warn +-- How each route decides what to do: each question it asks, and where each answer goes -- + get_book() books/routes.py line 28 (GET /books/{book_id}) + 30 cached? yes: returns json.loads(cached) · no: goes on + 34 returns found + search() books/routes.py line 38 (GET /books) + (no decisions: one path from start to end) + add_review() books/routes.py line 43 (POST /books/{book_id}/reviews) + (no decisions: one path from start to end) + set_cover() books/routes.py line 49 (PUT /books/{book_id}/cover) + (no decisions: one path from start to end) + export() books/routes.py line 56 (GET /books/{book_id}/export) + (no decisions: one path from start to end) + -- When Python imports it, once, before the first call -- 7 from . import routes ↳ books/routes.py, run when it's imported diff --git a/tests/snapshots/cart-service.txt b/tests/snapshots/cart-service.txt index 52e7d4e..07bccad 100644 --- a/tests/snapshots/cart-service.txt +++ b/tests/snapshots/cart-service.txt @@ -66,6 +66,11 @@ flowchart LR classDef warn stroke:#f59e0b,stroke-width:2px class env,f0,f3,aws_dynamodb,aws_sns,lib_sql_postgresql,lib_llm_openai warn +-- How on_checkout() decides what to do: each question it asks, and where each answer goes -- + on_checkout() cart/app.py line 20 + 25 for item in cart['items'] (back to its top for each) + 32 returns {'statusCode': 200, 'body': json.dumps(receipt)} + -- When Python imports it, once, before the first call -- 13 from .assistant import summary ↳ cart/assistant.py, run when it's imported diff --git a/tests/snapshots/orders-service.txt b/tests/snapshots/orders-service.txt index 3a986d4..a28226f 100644 --- a/tests/snapshots/orders-service.txt +++ b/tests/snapshots/orders-service.txt @@ -76,6 +76,13 @@ flowchart LR classDef warn stroke:#f59e0b,stroke-width:2px class f7,http warn +-- How lambda_handler() decides what to do: each question it asks, and where each answer goes -- + lambda_handler() orders/handler.py line 30 + 32 for record in event['Records'] (back to its top for each) + 35 for order in read_orders(bucket, key) (back to its top for each) + 36 check(order)? yes: goes on · no: logs a warning, then skips to the next item + 42 returns {'saved': saved} + -- When Python imports it, once, before the first call -- 12 from .notify import announce ↳ orders/notify.py, run when it's imported @@ -95,7 +102,7 @@ flowchart LR 19 ↳ def read_orders(bucket, key): # orders/handler.py 21 body = s3.get_object(Bucket=bucket, Key=key)['Body'].read() # S3 GetObject (reads) · Bucket ← event Records[].s3.bucket.name · Key ← event Records[].s3.object.key 22 return [json.loads(line) for line in body.splitlines() if line] - 36 if not check(order): + 36 if not check(order): # then continue 25 ↳ def check(order): # orders/handler.py 27 return bool(order.get('id')) and bool(order.get('customer')) and (order.get('total', 0) > 0) 37 log.warning('skipping %s', order.get('id')) # logs a warning diff --git a/tests/unit/test_reader.py b/tests/unit/test_reader.py index 914455e..a82b327 100644 --- a/tests/unit/test_reader.py +++ b/tests/unit/test_reader.py @@ -241,3 +241,27 @@ def test_the_entry_point_s_file_is_found_from_its_name() -> None: assert entry_file("tool.main", names) == ("src/tool/__init__.py", "src/") # a package's __init__ assert entry_file("app.handler", names) == ("app.py", "") assert entry_file("missing.main", names) is None + + +def test_an_if_or_except_that_leaves_its_loop_says_how() -> None: + code = """ +import requests + +def handler(event, context): + for url in event["urls"]: + if not url: + continue + try: + answer = requests.get(url, timeout=5) + except requests.Timeout: + continue + if answer.ok: + break +""" + cmap = map_code({"app.py": code}, "app.handler", platform="lambda") + loop = cmap.steps[0].steps[0] + assert [(s.kind, s.code, s.ends) for s in loop.steps if s.kind in ("if", "except")] == [ + ("if", "if not url:", "continue"), # kept for its continue alone: it decides what runs + ("except", "except requests.Timeout:", "continue"), + ("if", "if answer.ok:", "break"), + ] diff --git a/tests/unit/test_report.py b/tests/unit/test_report.py index e51587e..d837d89 100644 --- a/tests/unit/test_report.py +++ b/tests/unit/test_report.py @@ -142,6 +142,8 @@ def test_services_show_their_real_logos(tmp_path: Path) -> None: page = render_page(page_blocks_for(project, "app.handler"), "x") assert page.count('
')] + + +def test_the_decisions_tab_draws_each_question_and_where_its_answers_go(tmp_path: Path) -> None: + page = _decisions_page(tmp_path, {"app.py": DECIDES}) + assert '' in page + flow = _flow_part(page) + # an elif is the next question on the no path, and else's raise sits beside it + assert "kind == 'refund'?" in flow and "kind == 'charge'?" in flow + assert "Raises ValueError(kind)" in flow and "Deletes" not in flow and "Writes to DynamoDB (DeleteItem)" in flow + # `if not x: continue` is asked as x, and its no answer goes on to the next item + assert "item.get('sku')?" in flow and "skips to the next item" in flow + assert ">next item" in flow and ">when every one is done" in flow # the arrow back, and out + assert '' in flow and '' in flow + assert "Returns {'statusCode': 200}" in flow + # charge() decides too, so it's opened in place: its try, the way taken when it raises, the loop's else + assert '' in flow and "charge()" in flow + assert "if it raises requests.Timeout" in flow and "skips to the next attempt" in flow + assert "leaves the loop" in flow and "If the loop ran to its end, with no break" in flow + assert "Sends a message with SQS" in flow + assert " None: + deep = "".join( + f"def {name}(x):\n if x:\n return {after}(x)\n return None\n\n\n" + for name, after in zip("abcd", ["b", "c", "d", "int"]) + ) + page = _decisions_page(tmp_path, {"app.py": deep + "def lambda_handler(event, context):\n return a(event)\n"}) + flow = page[page.index('
') + assert first.count('') == 2 # a() and b(), opened in place; c() is a box linking to its flow + assert 'href="#cnd-flow0-0"' in first and "2 decisions inside · its flow ↓" in first # its own, and d()'s + assert "Inside c()" in own and own.count('') == 1 # d(), opened in c()'s + + +def test_each_route_of_a_web_app_has_a_flow(tmp_path: Path) -> None: + app = """ +from fastapi import FastAPI + +app = FastAPI() + + +@app.get("/books/{book_id}") +def get_book(book_id: int): + if book_id < 0: + raise ValueError(book_id) + return {"id": book_id} + + +@app.get("/health") +def health(): + return {"ok": True} +""" + project = read_project(write(tmp_path / "code", {"main.py": app})) + from code_node.report.build import page_blocks + + cmap = map_code(project.files, "main.app", libraries=project.libraries(), platform="web") + blocks = page_blocks(cmap, project="x", max_functions=24, max_depth=5, sources=project.files) + page = render_page(blocks, "x") + assert "How each route decides what to do" in page + assert "get_book() · GET /books/{book_id}" in page and "health() · GET /health" in page + assert "book_id < 0?" in page and "Raises ValueError(book_id)" in page + + +def test_the_decisions_in_text_say_where_each_answer_goes(tmp_path: Path) -> None: + from code_node.kit.render import _render_text + + project = read_project(write(tmp_path / "code", {"app.py": DECIDES})) + lines = _render_text(page_blocks_for(project, "app.lambda_handler"), 100) + said = lines[lines.index("-- How lambda_handler() decides") : lines.index("-- When Python imports it")] + assert "kind == 'refund'? yes: writes to DynamoDB (DeleteItem) · no: asks the next question" in said + assert "kind == 'charge'? yes: calls charge() · no: raises ValueError(kind)" in said + assert "item.get('sku')? yes: goes on · no: skips to the next item" in said + assert "↳ charge() app.py line" in said and "except requests.Timeout: skips to the next item" in said