diff --git a/fnd/extract/_html.py b/fnd/extract/_html.py index bc6a0533..9f3e694a 100644 --- a/fnd/extract/_html.py +++ b/fnd/extract/_html.py @@ -11,17 +11,33 @@ from __future__ import annotations import re +from itertools import chain +from typing import TYPE_CHECKING import lxml.html +from lxml.etree import ParserError from lxml.html import HtmlElement from fnd.extract._sectioner import HeadingSectioner from fnd.extract._tables import gfm_table +if TYPE_CHECKING: + from fnd.extract._script_text import ScriptBlock + _WS = re.compile(r"\s+") _HEADINGS = {"h1": 1, "h2": 2, "h3": 3, "h4": 4, "h5": 5, "h6": 6} # Elements whose subtree carries no readable content. -_SKIP = {"script", "style", "head", "title", "meta", "link", "noscript", "svg", "template"} +_SKIP = { + "script", + "style", + "head", + "title", + "meta", + "link", + "noscript", + "svg", + "template", +} # Structural containers we descend into without emitting anything ourselves. _CONTAINERS = { "div", @@ -60,115 +76,498 @@ def _collapse(text: str) -> str: return _WS.sub(" ", text).strip() -def _inline_md(el: HtmlElement) -> str: - """Render an inline subtree as markdown (bold/italic/code/links preserved).""" +_BLOCK_TAGS = { + *_HEADINGS, + *("p", "ul", "ol", "li", "pre", "table", "blockquote", "div", "section", "article"), + *("main", "header", "footer", "aside", "figure", "details", "form", "nav"), +} +_LANG_CLASS = re.compile(r"(?:^|\s)(?:language|lang)-([\w+#-]+)") +# Obsidian callout type for each class name a page might give an aside box. +_CALLOUT_KINDS = { + "warn": "warning", + "warning": "warning", + "caution": "caution", + "danger": "danger", + "error": "danger", + "note": "note", + "info": "info", + "tip": "tip", + "hint": "tip", + "important": "important", + "success": "success", +} +# A whole class token (``warn``, ``alert-warning``), never a word inside one (``post-info``). +_CALLOUT_CLASS = re.compile( + rf"(?:^|\s)(?:(?:alert|callout|admonition|notice|box)[-_])?({'|'.join(_CALLOUT_KINDS)})(?=\s|$)", + re.I, +) +_BLOCK_ATTR = "fnd-block" # set on an element with a block-level descendant +_INLINE_DATA_CHARS = 4_000 +_BREAK = re.compile(r"", re.I) +_MD_LINK = re.compile(r"\[([^\[\]]*)\]\([^()]*\)") +_EMPHASIS_CLASS = re.compile(r"(?:^|\s)(?:key|takeaway|tldr)(?:\s|$)") +_BANNER_CLASS = re.compile( + r"(?:^|\s)(?:stage|chapter|part|unit|phase)(?:[-_](?:h|head|header|heading|title|banner))?(?=\s|$)" +) +_SENTENCE_END = ".!?:;" +# Parents whose children are one run of text split by styling (syntax colouring, small caps). +_TIGHT = { + "code", + "pre", + "kbd", + "samp", + "a", + "em", + "strong", + "b", + "i", + "u", + "p", + "td", + "th", + "label", + "small", + "sub", + "sup", + "mark", + *_HEADINGS, +} +_TAG = re.compile(r"]*>") +_ENTITY = re.compile(r"&(?:#\d+|#[xX][0-9a-fA-F]+|[A-Za-z][A-Za-z0-9]*);") +# A list item's own text stops at the first of these; they render as blocks after it. +_ITEM_BLOCKS = {"pre", "details", "table", "blockquote", "div", "figure"} + + +def _needs_space(child: HtmlElement, last: str, glued: bool, loose: bool) -> bool: + """Whether a page that only spaces ``child`` from the text before it with CSS needs a space. + + Minified markup leaves nothing between sibling widgets (``HomeAbout``). + """ + if str(child.tag).lower() == "br" or not last: + return False + starts_sentence = last in _SENTENCE_END and child.text_content().lstrip()[:1].isupper() + return (glued and loose and not last.isspace()) or starts_sentence + + +def _append_tail(parts: list[str], child: HtmlElement, bold_title: bool, last: str) -> None: + """Add ``child``'s tail; ``last`` is the last character the child showed.""" + tail = child.tail + if not tail: + return + ends_sentence = bool(last) and last in _SENTENCE_END + if tail[:1].isalnum() and (ends_sentence or bold_title): + parts.append(" ") + parts.append(tail) + + +def _is_title(child: HtmlElement, seen: bool) -> bool: + """A bold lead-in: the first words of the text, with nothing visible before it.""" + return str(child.tag).lower() in ("b", "strong") and not seen + + +def _is_prose(el: HtmlElement) -> bool: + """Whether ``el`` has words of its own around its children (not a bare run of widgets).""" + return bool((el.text or "").strip()) or any((c.tail or "").strip() for c in el) + + +def _is_loose(el: HtmlElement) -> bool: + """Whether the children of ``el`` are separate items rather than pieces of one phrase.""" + return str(el.tag).lower() not in _TIGHT and not _is_prose(el) + + +def _render(el: HtmlElement, plain: bool, tight: bool = False) -> tuple[str, str]: + """Inline subtree as (markdown, last visible character); ``plain`` drops the markup. + + ``tight``: an ancestor is one run of text, so nothing below it is a separate widget. + """ parts: list[str] = [] + last = "" if el.text: parts.append(el.text) + last = el.text[-1] + loose = not tight and _is_loose(el) + glued, seen = False, bool((el.text or "").strip()) for child in el: if not isinstance(child.tag, str): # comment / PI if child.tail: parts.append(child.tail) + last = child.tail[-1] + seen = seen or bool(child.tail.strip()) continue - tag = child.tag.lower() - if tag == "br": + if _needs_space(child, last, glued, loose): parts.append(" ") - elif tag in _SKIP: - pass - elif tag == "a": - # Preserve the URL as a markdown link so it stays searchable and - # visible in the preview; bare anchors fall back to their text. - inner = _inline_md(child) - href = child.get("href") - parts.append(f"[{inner}]({href})" if href and inner.strip() else inner) - else: - inner = _inline_md(child) - wrap = _WRAP.get(tag) - if wrap and inner.strip(): - parts.append(f"{wrap}{inner}{wrap}") - else: - parts.append(inner) + last = " " + glued = not child.tail + text, piece_last = _piece(child, plain, not loose) + parts.append(text) + last = piece_last or last + _append_tail(parts, child, _is_title(child, seen), piece_last) + seen = seen or bool(text.strip()) or bool((child.tail or "").strip()) if child.tail: - parts.append(child.tail) - return "".join(parts) + last = child.tail[-1] + return "".join(parts), last -def _list_items(el: HtmlElement, sec: HeadingSectioner, depth: int, ordered: bool) -> None: +def _piece(child: HtmlElement, plain: bool, tight: bool) -> tuple[str, str]: + """One inline child as (text, last visible character).""" + tag = str(child.tag).lower() + if tag == "br": + return " ", " " + if tag in _SKIP: + return "", "" + inner, last = _render(child, plain, tight) + href = child.get("href") if tag == "a" else None + wrap = _WRAP.get(tag) + if plain or not inner.strip(): + return inner, last + if href: # the preview keeps the link; the index text has only the words + return f"[{inner}]({href})", last + return (f"{wrap}{inner}{wrap}" if wrap else inner), last + + +def _inline_md(el: HtmlElement, plain: bool = False) -> str: + return _render(el, plain)[0] + + +def _inline_text(el: HtmlElement) -> str: + return _collapse(_inline_md(el, plain=True)) + + +class _Run: + """A paragraph of a list item built child by child: markdown and its plain-text twin.""" + + def __init__(self, item: HtmlElement, lead: str = "") -> None: + self.loose = _is_loose(item) + self.md, self.txt = [lead], [lead] + self.last, self.seen, self.glued = lead[-1:], bool(lead.strip()), False + + def text(self, text: str) -> None: + self.md.append(text) + self.txt.append(text) + self.last = text[-1:] or self.last + self.seen = self.seen or bool(text.strip()) + + def add(self, child: HtmlElement) -> None: + """Add one inline child and its tail.""" + if _needs_space(child, self.last, self.glued, self.loose): + self.text(" ") + self.glued = not child.tail + md, last = _piece(child, False, not self.loose) + txt = _piece(child, True, not self.loose)[0] + title = _is_title(child, self.seen) + self.md.append(md) + self.txt.append(txt) + self.last = last or self.last + self.seen = self.seen or bool(txt.strip()) + _append_tail(self.md, child, title, last) + _append_tail(self.txt, child, title, last) + if child.tail: + self.last = child.tail[-1] + self.seen = self.seen or bool(child.tail.strip()) + + def lines(self) -> tuple[str, str]: + """(markdown, plain text), whitespace collapsed.""" + return _collapse("".join(self.md)), _collapse("".join(self.txt)) + + def reset(self) -> None: + self.md, self.txt = [], [] + self.last, self.seen, self.glued = "", False, False + + +def _list_items( + el: HtmlElement, + sec: HeadingSectioner, + depth: int, + ordered: bool, + attach: dict[str, list[ScriptBlock]], +) -> None: + number = 0 for li in el: if not isinstance(li.tag, str) or li.tag.lower() != "li": continue - # The item's own text is its inline content minus any nested list. - nested = [c for c in li if isinstance(c.tag, str) and c.tag.lower() in ("ul", "ol")] - item_md_parts: list[str] = [] - if li.text: - item_md_parts.append(li.text) + number += 1 + # The item's own text is its inline content up to the first block child; + # later blocks and nested lists follow it. + run = _Run(li, li.text or "") + rest: list[HtmlElement] = [] # from the first block or nested list on, in page order for c in li: - if not isinstance(c.tag, str): - if c.tail: - item_md_parts.append(c.tail) - continue - if c.tag.lower() in ("ul", "ol"): - if c.tail: - item_md_parts.append(c.tail) - continue - item_md_parts.append(_inline_md(c)) - if c.tail: - item_md_parts.append(c.tail) - item_md = _collapse("".join(item_md_parts)) - if item_md: - sec.add_list_item(item_md, item_md, depth=depth, ordered=ordered) - for sub in nested: - _list_items(sub, sec, depth + 1, sub.tag.lower() == "ol") + tag = c.tag.lower() if isinstance(c.tag, str) else "" + lone_p = tag == "p" and run.seen + if rest or tag in ("ul", "ol") or tag in _ITEM_BLOCKS or lone_p: + rest.append(c) + elif not tag: + run.text(c.tail or "") + else: + run.add(c) + md, item = run.lines() + bare = not item and rest and rest[0].tag in ("div", "p") + if bare and _inline_only(rest[0]): # an item whose text sits in a leading block + head = rest.pop(0) + md = _collapse(f"{_inline_md(head)} {head.tail or ''}") + item = _collapse(f"{_inline_text(head)} {head.tail or ''}") + if item: + sec.add_list_item(md, item, depth=depth, ordered=ordered, number=number) + _item_rest(li, rest, sec, depth, attach) + + +def _item_rest( + li: HtmlElement, + rest: list[HtmlElement], + sec: HeadingSectioner, + depth: int, + attach: dict[str, list[ScriptBlock]], +) -> None: + """What follows a list item's first block: blocks, nested lists, and the text between them.""" + run = _Run(li) + for c in rest: + tag = c.tag.lower() if isinstance(c.tag, str) else "" + if tag and tag not in ("ul", "ol", "p") and tag not in _ITEM_BLOCKS: + run.add(c) + continue + _flush(run, sec) + if tag in ("ul", "ol"): + _list_items(c, sec, depth + 1, tag == "ol", attach) + else: + _walk_child(c, sec, attach) + run.text(c.tail or "") + _flush(run, sec) + + +def _flush(run: _Run, sec: HeadingSectioner) -> None: + md, text = run.lines() + if text: + sec.add_paragraph(md, text) + run.reset() def _table_md(el: HtmlElement) -> tuple[str, list[str]]: rows: list[list[str]] = [] cells: list[str] = [] for tr in el.iter("tr"): + section = tr.getparent() # a nested table's rows are rendered inside their cell instead + if section is not el and (section is None or section.getparent() is not el): + continue row: list[str] = [] for cell in tr: if isinstance(cell.tag, str) and cell.tag.lower() in ("td", "th"): - text = _collapse(cell.text_content()) - row.append(text) - cells.append(text) + row.append(_collapse(_inline_md(cell))) + cells.append(_inline_text(cell)) if row: rows.append(row) return gfm_table(rows), cells -def _walk(el: HtmlElement, sec: HeadingSectioner) -> None: - for child in el: - if not isinstance(child.tag, str): - continue - tag = child.tag.lower() - if tag in _SKIP: - continue - if tag in _HEADINGS: - sec.add_heading(_HEADINGS[tag], _collapse(child.text_content())) - elif tag == "p": - sec.add_paragraph(_collapse(_inline_md(child)), _collapse(child.text_content())) - elif tag in ("ul", "ol"): - _list_items(child, sec, 0, tag == "ol") - elif tag == "pre": - sec.add_code(child.text_content()) - elif tag == "blockquote": - sec.add_quote(_collapse(child.text_content())) - elif tag == "table": - md, cells = _table_md(child) - sec.add_table(md, cells) - elif tag in _CONTAINERS: - _walk(child, sec) +def _cell_md(cell: str) -> str: + """A table cell's markdown: inline HTML converted, line breaks kept as ``/``.""" + if not _TAG.search(cell) and not _ENTITY.search(cell): + return cell + holder = _fragment(_BREAK.sub(" / ", cell)) + return _collapse(_inline_md(holder)) if holder is not None else _collapse(_TAG.sub("", cell)) + + +def _fragment(markup: str) -> HtmlElement | None: + """Parse a script-built HTML fragment; None when the parser rejects it.""" + try: + holder = lxml.html.fragment_fromstring(markup, create_parent="div") + except (ValueError, ParserError, AssertionError): # AssertionError: "too many bodies" + return None + _mark_blocks(holder) + return holder + + +def add_blocks(sec: HeadingSectioner, blocks: list[ScriptBlock], *, level: int = 0) -> None: + """Emit script-derived blocks; paragraphs keep their inline markup as markdown. + + A data section's name opens a heading at ``level``; with no level it is a bold lead. + """ + for block in blocks: + try: + _add_block(sec, block, level) + except RecursionError: # a fragment nested too deep for the stack the page already uses + plain = _collapse(_TAG.sub("", block.text)) + if plain: + sec.add_paragraph(plain, plain) + + +def _add_block(sec: HeadingSectioner, block: ScriptBlock, level: int) -> None: + if block.kind == "h": + if level: + sec.add_heading(level, block.text) else: - # Unknown block: descend if it has element children, else treat as prose. - if len(child): - _walk(child, sec) - else: - text = _collapse(child.text_content()) - if text: - sec.add_paragraph(text, text) + sec.add_paragraph(f"**{block.text}**", block.text) + elif block.kind == "code": + sec.add_code(block.text, block.lang) + elif block.kind == "table": + rows = [[_cell_md(c) for c in row] for row in block.rows] + cells = [_MD_LINK.sub(r"\1", c).replace("`", "").replace("**", "") for r in rows for c in r] + sec.add_table(gfm_table(rows), cells) + elif block.kind == "html": + holder = _fragment(block.text) + if holder is None: + sec.add_paragraph( + _collapse(_TAG.sub("", block.text)), _collapse(_TAG.sub("", block.text)) + ) + elif _inline_only(holder): + _paragraph(sec, holder) + else: + _walk(holder, sec, {}) + else: + md, text = block.text, block.text + markup = _TAG.search(block.text) or _ENTITY.search(block.text) + if markup and (holder := _fragment(block.text)) is not None: + md, text = _collapse(_inline_md(holder)), _inline_text(holder) + if block.kind == "li": + sec.add_list_item(md, text) + elif text: + sec.add_paragraph(md, text) + + +def _heading_text(el: HtmlElement) -> str: + """Heading text with a leading number badge (``5Title``) spaced off.""" + full = _collapse(el.text_content()) + first = el[0] if len(el) and not el.text else None + badge = ( + _collapse(first.text_content()) if first is not None and isinstance(first.tag, str) else "" + ) + if badge.isdigit() and full.startswith(badge) and full[len(badge) :][:1] not in " .):": + return f"{badge} {full[len(badge) :]}".strip() + return full + + +def _code_lang(pre: HtmlElement) -> str: + """Language named by a ``language-x`` / ``lang-x`` class on a ``
`` or its ````."""
+    for el in (pre, *pre.iter("code")):
+        if m := _LANG_CLASS.search(el.get("class", "")):
+            return m.group(1).lower()
+    return ""
+
+
+def _callout(el: HtmlElement) -> tuple[str, str, str, str] | None:
+    """(kind, title, body markdown, body text) of a ``
Tbody
`` box.""" + m = _CALLOUT_CLASS.search(el.get("class", "")) + if el.tag != "div" or not m or not _inline_only(el): + return None + title = "" + if len(el) and el[0].tag in ("b", "strong") and not (el.text or "").strip(): + title = _collapse(el[0].text_content()) + md = _collapse(_inline_md(el)) + text = _inline_text(el) + if title: + md = md.removeprefix(f"**{title}**").strip() + text = text.removeprefix(title).strip() + kind = _CALLOUT_KINDS[m.group(1).lower()] + return (kind, title, md, text) if md or title else None + + +def _banner(el: HtmlElement) -> tuple[str, str] | None: + """(title, subtitle) of a stage/chapter banner: ``
TS``.""" + if el.tag != "div" or not _BANNER_CLASS.search(el.get("class", "")): + return None + kids = [c for c in el if isinstance(c.tag, str)] + if not kids or kids[0].tag not in ("b", "strong") or any(c.tag != "span" for c in kids[1:]): + return None + title = _collapse(kids[0].text_content()) + return (title, _collapse(" ".join(c.text_content() for c in kids[1:]))) if title else None + + +def _mark_blocks(root: HtmlElement) -> None: + """Flag every element that has a block-level descendant, in one pass over the tree.""" + for el in reversed(list(root.iter())): # reverse document order: children before parents + parent = el.getparent() + if parent is None or not isinstance(el.tag, str): + continue + if el.tag.lower() in _BLOCK_TAGS or el.get(_BLOCK_ATTR) is not None: + parent.set(_BLOCK_ATTR, "1") + + +def _inline_only(el: HtmlElement) -> bool: + """Whether ``el`` holds no block-level element (the tree went through ``_mark_blocks``).""" + return el.get(_BLOCK_ATTR) is None + + +def _paragraph(sec: HeadingSectioner, el: HtmlElement, *, bold: bool = False) -> None: + md, text = _collapse(_inline_md(el)), _inline_text(el) + if md and text: + sec.add_paragraph(f"**{text}**" if bold else md, text) + + +def _stray(sec: HeadingSectioner, text: str | None) -> None: + """Text sitting directly between a container's block children.""" + if text and text.strip(): + sec.add_paragraph(_collapse(text), _collapse(text)) + + +def _walk(el: HtmlElement, sec: HeadingSectioner, attach: dict[str, list[ScriptBlock]]) -> None: + _stray(sec, el.text) + for child in el: + _walk_child(child, sec, attach) + _stray(sec, child.tail) + + +def _walk_child( + child: HtmlElement, sec: HeadingSectioner, attach: dict[str, list[ScriptBlock]] +) -> None: + if not isinstance(child.tag, str): + return + tag = child.tag.lower() + if tag in _SKIP: + _attach(child, sec, attach) # a JSON island's own data is placed where it sits + return + if tag in _HEADINGS: + sec.add_heading(_HEADINGS[tag], _heading_text(child)) + elif tag == "p": + _paragraph(sec, child, bold=bool(_EMPHASIS_CLASS.search(child.get("class", "")))) + elif tag in ("ul", "ol"): + _list_items(child, sec, 0, tag == "ol", attach) + elif tag == "pre": + sec.add_code(child.text_content(), _code_lang(child)) + elif tag == "blockquote": + sec.add_quote(_inline_text(child), _collapse(_inline_md(child))) + elif tag == "table": + md, cells = _table_md(child) + sec.add_table(md, cells) + elif (callout := _callout(child)) is not None: + sec.add_callout(*callout) + elif (banner := _banner(child)) is not None: + sec.add_heading(2, banner[0]) + if banner[1]: + sec.add_paragraph(banner[1], banner[1]) + elif tag == "summary": + _paragraph(sec, child, bold=True) + elif (tag in _CONTAINERS or len(child)) and tag != "dl" and _inline_only(child): + _paragraph(sec, child) + elif tag in _CONTAINERS or len(child): + _walk(child, sec, attach) + _attach(child, sec, attach) + return + else: + text = _collapse(child.text_content()) + if text: + sec.add_paragraph(text, text) + _attach(child, sec, attach) + for island in child.iterdescendants("script"): # flattened above, so never walked + _attach(island, sec, attach) + + +def _attach(el: HtmlElement, sec: HeadingSectioner, attach: dict[str, list[ScriptBlock]]) -> None: + """Emit the script blocks named by ``el``'s ``id`` or a ``data-*`` attribute.""" + for name, value in el.attrib.items(): # an empty value (data-v-1a2b="") names nothing + if value and (name == "id" or name.startswith("data-")) and value in attach: + blocks = attach.pop(value) + size = sum(len(b.text) + sum(map(len, chain(*b.rows))) for b in blocks) + # A widget's data reads as part of its section; a whole dataset gets sections. + level = min(max(sec.depth + 1, 2), 6) if size > _INLINE_DATA_CHARS else 0 + add_blocks(sec, blocks, level=level) + +def walk_html( + root: HtmlElement, sec: HeadingSectioner, attach: dict[str, list[ScriptBlock]] | None = None +) -> None: + """Feed the reading-order content of ``root`` into ``sec``. -def walk_html(root: HtmlElement, sec: HeadingSectioner) -> None: - """Feed the reading-order content of ``root`` into ``sec``.""" + ``attach`` maps an attribute value to script blocks emitted after the element + carrying it; entries not claimed by any element are left in the dict. + """ + _mark_blocks(root) body = root.body if root.tag == "html" and root.body is not None else root - _walk(body, sec) + _walk(body, sec, {} if attach is None else attach) diff --git a/fnd/extract/_js_data.py b/fnd/extract/_js_data.py new file mode 100644 index 00000000..f1b84fa6 --- /dev/null +++ b/fnd/extract/_js_data.py @@ -0,0 +1,327 @@ +"""Tolerant reader for the data literals in inline scripts (nothing is executed). + +``parse_assignments`` finds ``var name = {...}``, ``name = {...}`` and +``X.name = [...]`` literals and returns them as Python values. Anything that is not plain data (calls, +functions, member access) becomes ``OPAQUE`` and the parse carries on, so a +page's computed cells never hide the literal data around them. Identifiers +that name an earlier ``var``/``let``/``const`` literal resolve to its value. +""" + +from __future__ import annotations + +import re +from bisect import bisect_left +from typing import Final, NamedTuple + +OPAQUE: Final = object() + +_MAX_DEPTH = 40 +_TOKEN = re.compile( + r""" + (?P\s+|//[^\n]*|/\*(?:.*?\*/|.*)) + | (?P"(?:[^"\\\n]|\\.)*(?:"|(?=\n)|\Z)|'(?:[^'\\\n]|\\.)*(?:'|(?=\n)|\Z)) + | (?P`(?:[^`\\]|\\.)*(?:`|\Z)) + | (?P-?(?:\d+(?:\.\d*)?|\.\d+)(?:[eE][+-]?\d+)?(?![\w$])) + | (?P[\w$]+) + | (?P[{}\[\](),:;]) + | (?P.) + """, + re.S | re.X, +) +_ASSIGN = re.compile( + r"(?:(? str: + def sub(m: re.Match[str]) -> str: + digits = m.group(1) or m.group(2) + return chr(int(digits, 16)) if digits else _ESCAPES.get(m.group(3), m.group(3)) + + text = _ESCAPE.sub(sub, raw) + if _SURROGATE.search( + text + ): # a pair of \uXXXX escapes is one character; a lone half is not text + text = text.encode("utf-16", "surrogatepass").decode("utf-16", "replace") + return text + + +def _tokens(src: str) -> list[_Tok]: + out: list[_Tok] = [] + for m in _TOKEN.finditer(src): + kind, text = m.lastgroup or "other", m.group() + if kind in ("str", "tpl") and not _closed(text): + kind = "other" # an unterminated literal swallowed its line; keep it out of the data + if kind != "skip": + out.append(_Tok(kind, text, m.start(), m.end())) + return out + + +def _closed(literal: str) -> bool: + inner = literal[1:-1] + return ( + len(literal) > 1 + and literal[-1] == literal[0] + and (len(inner) - len(inner.rstrip("\\"))) % 2 == 0 + ) + + +class _Expansion: + """Size of the values a script refers to by name, so a shared reference cannot multiply.""" + + def __init__(self) -> None: + self.left = _MAX_EXPANSION + self._weights: dict[int, int] = {} + + def weight(self, value: object) -> int: + if isinstance(value, str): + return 1 + len(value) // 16 + if not isinstance(value, (list, dict, Call)): + return 1 + known = self._weights.get(id(value)) + if known is None: + parts = ( + value.values() + if isinstance(value, dict) + else value.args + if isinstance(value, Call) + else value + ) + known = min(1 + sum(self.weight(p) for p in parts), _MAX_EXPANSION + 1) + self._weights[id(value)] = known + return known + + def take(self, value: object) -> bool: + """Charge one more use of ``value``; False once the script has expanded too far.""" + cost = self.weight(value) + if cost > self.left: + return False + self.left -= cost + return True + + +class _Parser: + def __init__( + self, toks: list[_Tok], env: dict[str, object], expansion: _Expansion | None = None + ) -> None: + self.toks = toks + self.env = env + self.expansion = expansion or _Expansion() + self.i = 0 + + def _peek(self) -> _Tok | None: + return self.toks[self.i] if self.i < len(self.toks) else None + + def _at(self, text: str) -> bool: + t = self._peek() + return t is not None and t.kind in ("punct", "other") and t.text == text + + def _skip_expr(self) -> object: + """Consume to the next depth-0 terminator; the expression is not data.""" + depth = 0 + while (t := self._peek()) is not None: + if t.kind == "punct" and t.text in "([{": + depth += 1 + elif t.kind == "punct" and t.text in ")]}": + if depth == 0: + break + depth -= 1 + elif depth == 0 and t.kind == "punct" and t.text in ",;": + break + self.i += 1 + return OPAQUE + + def value(self, depth: int = 0) -> object: + t = self._peek() + if t is None or depth > _MAX_DEPTH: + return self._skip_expr() + v: object + if t.kind == "str": + self.i += 1 + v = unescape(t.text[1:-1]) + elif t.kind == "tpl": + self.i += 1 + v = OPAQUE if "${" in t.text else unescape(t.text[1:-1]) + elif t.kind == "num": + self.i += 1 + v = float(t.text) if any(c in t.text for c in ".eE") else int(t.text) + elif t.kind == "punct" and t.text == "[": + v = self._array(depth + 1) + elif t.kind == "punct" and t.text == "{": + v = self._object(depth + 1) + elif t.kind == "word": + self.i += 1 + v = self._call(t.text, depth + 1) if self._at("(") else self._word(t.text) + else: + return self._skip_expr() + nxt = self._peek() + if nxt is None or (nxt.kind == "punct" and nxt.text in _END): + return v + if isinstance(v, str) and self._at("+"): + self.i += 1 + rest = self.value(depth + 1) + return v + rest if isinstance(rest, str) else OPAQUE + return self._skip_expr() + + def _word(self, text: str) -> object: + if text in ("true", "false"): + return text == "true" + if text in ("null", "undefined"): + return None + value = self.env.get(text, OPAQUE) + return value if self.expansion.take(value) else OPAQUE + + def _items(self, closer: str, depth: int) -> list[object]: + self.i += 1 + items: list[object] = [] + while (t := self._peek()) is not None and not (t.kind == "punct" and t.text == closer): + before = self.i + items.append(self.value(depth)) + if self._at(",") or self.i == before: + self.i += 1 + self.i += 1 + return items + + def _array(self, depth: int) -> list[object]: + return self._items("]", depth) + + def _call(self, name: str, depth: int) -> Call: + return Call(name, tuple(self._items(")", depth))) + + def _object(self, depth: int) -> dict[str, object]: + self.i += 1 + obj: dict[str, object] = {} + while (t := self._peek()) is not None and not (t.kind == "punct" and t.text == "}"): + before = self.i + key = unescape(t.text[1:-1]) if t.kind == "str" else t.text + self.i += 1 + if t.kind in ("word", "str", "num") and self._at(":"): + self.i += 1 + obj[key] = self.value(depth) + else: + self.i = before + 1 + self._skip_expr() + if self._at(",") or self.i == before: + self.i += 1 + self.i += 1 + return obj + + +def parse_literal(src: str) -> object: + """The value of a bare literal such as a ``') + + +def test_shell_steps_get_a_bash_fence(tmp_path: Path) -> None: + md = _flow(tmp_path, _PAGE_SCRIPT) + assert "```bash\nls /data\n```" in md + + +def test_a_view_caption_with_no_subtitle_is_not_left_dangling(tmp_path: Path) -> None: + script = """FLOWS = {}; FLOWS.demo = { steps: [ { say: "Show the records so far.", + views: [{ type: "rows", label: "women", items: [] }] } ] };""" + md = _flow(tmp_path, script) + assert "*women*" in md + assert "· *" not in md + + +def test_chip_rows_render_as_inline_code(tmp_path: Path) -> None: + script = """FLOWS = {}; FLOWS.demo = { steps: [ { say: "Two records, one per line.", + views: [{ type: "rows", label: "nums", sub: "RDD", + items: [C("1", "p0"), C("2", "p1")] }] } ] };""" + md = _flow(tmp_path, script) + assert "*nums · RDD*" in md + assert "`1` `2`" in md + + +def test_row_mappings_render_as_a_table_with_the_views_own_headings(tmp_path: Path) -> None: + script = """FLOWS = {}; FLOWS.demo = { steps: [ { say: "Each record is upper-cased.", + views: [{ type: "map", inLabel: "raw", mid: "step", outLabel: "clean", + rows: [{ a: C("ab"), b: C("AB"), why: "upper" }] }] } ] };""" + md = _flow(tmp_path, script) + assert "| raw | clean | step |" in md + assert "| `ab` | `AB` | upper |" in md + + +def _view_md(tmp_path: Path, view: str) -> str: + script = f"""FLOWS = {{}}; FLOWS.demo = {{ steps: [ {{ say: "Look at what the step did.", + views: [{view}] }} ] }};""" + return _flow(tmp_path, script) + + +def test_an_html_view_is_walked_as_a_document_not_dumped_as_code(tmp_path: Path) -> None: + html = ( + '
' + "
x[2]int(x[2]) < 25
1yes
" + ) + md = _view_md(tmp_path, f'{{ type: "html", html: "{html}" }}') + assert "| x[2] | int(x[2]) < 25 |" in md + assert "| 1 | yes |" in md + assert " None: + html = '
Row(x[0])
' + md = _view_md(tmp_path, f'{{ type: "html", html: "{html}" }}') + assert "Row( **x[0]** )" in md + assert "<" not in md.partition("**Step 1**")[2] + + +def test_a_true_flag_in_a_row_table_is_not_an_empty_cell(tmp_path: Path) -> None: + md = _view_md(tmp_path, '{ type: "groups", groups: [{ key: C("k"), sealed: true }] }') + assert "| `k` | yes |" in md + + +def test_a_takeaway_paragraph_is_emphasised(tmp_path: Path) -> None: + md = _md( + tmp_path, '

Read only from the path you name.

Plain text here.

' + ) + assert "**Read only from the path you name.**" in md + assert "**Plain text here.**" not in md + + +def test_a_pipeline_of_plan_steps_renders_as_one_arrow_line(tmp_path: Path) -> None: + md = _view_md( + tmp_path, '{ type: "plan", steps: [{ name: "textFile" }, { name: "map split" }] }' + ) + assert "`textFile` → `map split`" in md + + +def test_a_dropped_row_says_so(tmp_path: Path) -> None: + md = _view_md( + tmp_path, + '{ type: "map", inLabel: "in", outLabel: "out", rows: [{ a: C("x"), b: null, why: "no" }] }', + ) + assert "| `x` | dropped | no |" in md + + +def test_gate_results_name_the_rule_that_matched(tmp_path: Path) -> None: + md = _view_md( + tmp_path, + '{ type: "gates", col: "x[2]", rules: ["a < 1", "else"], items: [{ v: "3", hit: 1, out: "B" }] }', + ) + assert "| x[2] | first true rule | out |" in md + assert "**col**" not in md + assert "**rules**" not in md + assert "| 3 | `else` | B |" in md + + +def test_partition_lists_render_one_line_per_partition(tmp_path: Path) -> None: + script = """FLOWS = {}; FLOWS.demo = { steps: [ { say: "The list is cut into pieces.", + views: [{ type: "parts", label: "rdd", parts: [[C("1"), C("2")], [C("3")]] }] } ] };""" + md = _flow(tmp_path, script) + assert "- **Part 0** `1` `2`" in md + assert "- **Part 1** `3`" in md + + +def test_minified_sibling_elements_keep_their_words_apart(tmp_path: Path) -> None: + html = ( + "" + "
Apple
A fruit
Brick
A block
" + "
Price$10
" + ) + body = _body(tmp_path, html) + for words in ("Home About", "Apple\nA fruit", "Brick\nA block", "Price $10"): + assert words in body + + +def test_words_split_by_styling_stay_whole(tmp_path: Path) -> None: + html = ( + '

os.path

' + "

$10.99 and THE

" + "
42
" + ) + body = _body(tmp_path, html) + assert "os.path" in body + assert "$10.99" in body + assert "THE" in body + assert "42" in body + + +def test_markdown_and_index_text_space_a_sentence_boundary_alike(tmp_path: Path) -> None: + html = "

Note.Important stuff

" + assert "Note. Important stuff" in _body(tmp_path, html) + assert "*Note.* **Important** stuff" in _md(tmp_path, html) + + +def test_inline_markup_after_a_block_in_a_list_item_stays_one_paragraph(tmp_path: Path) -> None: + md = _md( + tmp_path, + '
  • Run
    make
    then check the log output
', + ) + assert "then `check` the [log](/log) output" in md + + +def test_a_list_item_made_of_blocks_keeps_its_bullet(tmp_path: Path) -> None: + md = _md(tmp_path, "
  • Ada
    Engineer
") + assert "- Ada" in md + assert "Engineer" in md + + +def test_a_standalone_button_label_is_indexed(tmp_path: Path) -> None: + html = '
An answer.
' + assert "What is it?" in _body(tmp_path, html) + + +def test_table_cells_keep_a_bare_less_than_and_url_ampersands(tmp_path: Path) -> None: + script = ( + "var DATA = { rows: [{ a: 'x

Doc

", encoding="utf-8" + ) + md = "\n".join(c.body_md for c in extract(p)) + assert "x None: + script = 'var NOTES = { hint: "The run printed \\u001b[32mpassed\\u001b[0m for every test case." };' + p = tmp_path / "page.html" + p.write_text( + f"

Doc

Ordinary page text stays searchable.

" + f"", + encoding="utf-8", + ) + chunks = list(extract(p)) + text = "\n".join(c.body for c in chunks) + md = "\n".join(c.body_md for c in chunks) + assert "Ordinary page text stays searchable." in text + assert "**passed**" in md + assert "" not in md + assert "\x1b" not in text + md + + +def test_json_ld_metadata_is_not_indexed_as_page_content(tmp_path: Path) -> None: + ld = '{"@type": "NewsArticle", "headline": "A headline about harbour works", "author": {"@type": "Person"}}' + p = tmp_path / "page.html" + p.write_text( + f'

Doc

Body text of the article.

', + encoding="utf-8", + ) + text = "\n".join(c.body for c in extract(p)) + assert "NewsArticle" not in text + assert "Script text" not in text + + +def test_list_text_after_a_block_child_is_kept(tmp_path: Path) -> None: + md = _md( + tmp_path, + "
  • Run this
    make all
    then check the output
  • " + "
  • Lead
    • child
    after nested
", + ) + assert "then check the output" in md + assert "after nested" in md + + +def test_a_button_inside_a_sentence_stays_in_the_text(tmp_path: Path) -> None: + assert "Click Save to save." in _body(tmp_path, "

Click to save.

") + + +def test_a_lead_paragraph_with_inline_bold_is_not_re_wrapped(tmp_path: Path) -> None: + assert "****" not in _md(tmp_path, '

Some bold words in a lead

') + + +def test_ordinary_class_names_do_not_become_callouts_or_headings(tmp_path: Path) -> None: + md = _md( + tmp_path, + '' + '
Something went wrong here.
' + '
7of 9
', + ) + assert "[!" not in md + assert "## 7" not in md + + +def test_a_dotted_number_split_by_a_tag_is_not_spaced(tmp_path: Path) -> None: + md = _md(tmp_path, "

Version 1.2 and example.com ok

") + assert "Version 1.2" in md + assert "example.**com**" in md + + +def test_glued_widgets_of_any_kind_keep_their_words_apart(tmp_path: Path) -> None: + html = ( + "
AlphaBeta
" + "
foobar
" + "
StatusPending
" + "
" + ) + body = _body(tmp_path, html) + for words in ("Alpha Beta", "foo bar", "Status Pending", "Comment Hello"): + assert words in body + + +def test_a_list_item_with_thousands_of_children_is_linear(tmp_path: Path) -> None: + links = "".join(f'link{i}' for i in range(8000)) + started = time.monotonic() + body = _body(tmp_path, f"
  • {links}

{'x' * 20000}

") + assert time.monotonic() - started < 3 + assert "link7999" in body + + +def test_escaped_emoji_in_script_data_is_valid_text(tmp_path: Path) -> None: + script = 'var NOTES = { intro: "\\ud83d\\ude80 Launch day is here for everyone today" };' + p = tmp_path / "page.html" + p.write_text( + f"

Doc

", encoding="utf-8" + ) + chunks = list(extract(p)) + for c in chunks: # the index encodes every field as UTF-8 + (c.body + c.body_md + c.heading_path).encode("utf-8") + assert "\U0001f680 Launch day" in "\n".join(c.body for c in chunks) + + +def test_a_shared_list_in_every_record_cannot_blow_the_budget(tmp_path: Path) -> None: + big = ", ".join(f'"item number {i}"' for i in range(1000)) + rows = ", ".join("{ a: 'row', b: 'value', c: BIG }" for _ in range(1600)) + p = tmp_path / "page.html" + p.write_text( + f"

Doc

", + encoding="utf-8", + ) + started = time.monotonic() + chunks = list(extract(p)) + assert time.monotonic() - started < 6 + assert sum(len(c.body_md) for c in chunks) < 6_000_000 + + +def test_a_shared_table_shown_in_every_step_cannot_blow_the_budget(tmp_path: Path) -> None: + rows = ", ".join(["['alpha', 'beta']"] * 2000) + step = "{ say: 'See it.', views: [{ type: 'table', cols: ['a', 'b'], rows: B }] }" + script = f"var B = [{rows}]; var F = {{}}; F.d = {{ steps: [{', '.join([step] * 1000)}] }};" + p = tmp_path / "page.html" + p.write_text( + f'

Doc

', + encoding="utf-8", + ) + started = time.monotonic() + chunks = list(extract(p)) + assert time.monotonic() - started < 6 + assert sum(len(c.body_md) for c in chunks) < 6_000_000 + + +def test_one_big_flat_list_in_a_record_is_linear(tmp_path: Path) -> None: + words = ", ".join(f"'w{i}'" for i in range(40000)) + script = f"var BIG = [{words}]; var DATA = [{{ a: 'x', b: BIG }}, {{ a: 'y', b: 'z' }}];" + p = tmp_path / "page.html" + p.write_text( + f"

Doc

", encoding="utf-8" + ) + started = time.monotonic() + list(extract(p)) + assert time.monotonic() - started < 3 + + +def test_prose_with_semicolons_and_entities_is_kept(tmp_path: Path) -> None: + script = ( + "var DATA = { notes: ['Prices rose; sales fell over the whole summer period', " + "'Fish & chips are sold on the pier every day'] };" + ) + p = tmp_path / "page.html" + p.write_text( + f"

Doc

", encoding="utf-8" + ) + text = "\n".join(c.body for c in extract(p)) + assert "Prices rose; sales fell" in text + assert "Fish & chips" in text + + +def test_words_split_inside_a_wrapper_span_stay_whole(tmp_path: Path) -> None: + html = ( + "

Read THE manual.

" + "
42
" + ) + body = _body(tmp_path, html) + assert "Read THE manual." in body + assert "42" in body + + +def test_deeply_nested_inline_markup_is_linear(tmp_path: Path) -> None: + html = "

" + "a " * 250 + "Z " * 100000 + "tail. " * 250 + "

" + started = time.monotonic() + _body(tmp_path, html) + assert time.monotonic() - started < 3 diff --git a/tests/test_web_script_bounds.py b/tests/test_web_script_bounds.py new file mode 100644 index 00000000..8337944f --- /dev/null +++ b/tests/test_web_script_bounds.py @@ -0,0 +1,292 @@ +"""A hostile or huge page script costs time and output in proportion to its size, never more.""" + +from __future__ import annotations + +import time +from pathlib import Path + +import pytest + +from fnd.extract import extract +from fnd.extract._js_data import parse_assignments + + +@pytest.fixture(autouse=True) +def _static_only(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("FND_HTML_EVAL", "0") + + +def _extract(tmp_path: Path, script: str) -> list[str]: + p = tmp_path / "page.html" + p.write_text( + f"

Doc

Ordinary page text stays searchable.

" + f"", + encoding="utf-8", + ) + return [c.body_md for c in extract(p)] + + +def test_shared_references_do_not_fan_out_exponentially(tmp_path: Path) -> None: + lines = ['var A0 = ["four plain words here"];'] + lines += [f"var A{i} = [A{i - 1}, A{i - 1}];" for i in range(1, 40)] + started = time.monotonic() + chunks = _extract(tmp_path, "\n".join(lines)) + assert time.monotonic() - started < 5 + assert "Ordinary page text stays searchable." in "\n".join(chunks) + + +def test_a_long_reference_chain_keeps_the_pages_own_text(tmp_path: Path) -> None: + lines = ['var A0 = ["a chain of four plain words"];'] + lines += [f"var A{i} = [A{i - 1}];" for i in range(1, 1500)] + assert "Ordinary page text stays searchable." in "\n".join(_extract(tmp_path, "\n".join(lines))) + + +def test_a_shared_big_string_cannot_multiply_into_a_huge_output(tmp_path: Path) -> None: + big = "word " * 100_000 + refs = ", ".join(["S"] * 3000) + chunks = _extract(tmp_path, f'var S = "{big}"; var A = [{refs}];') + assert sum(len(c) for c in chunks) < 20_000_000 + + +def test_an_unterminated_comment_is_read_in_linear_time() -> None: + src = "/* open " * 40_000 + started = time.monotonic() + parse_assignments(src) + assert time.monotonic() - started < 1.5 + + +def test_a_wide_shared_reference_graph_is_checked_in_linear_time(tmp_path: Path) -> None: + lines = ['var A0 = ["four plain words here"];'] + lines += [f"var A{i} = [{', '.join([f'A{i - 1}'] * 8)}];" for i in range(1, 10)] + started = time.monotonic() + chunks = _extract(tmp_path, "\n".join(lines)) + assert time.monotonic() - started < 5 + assert "Ordinary page text stays searchable." in "\n".join(chunks) + + +@pytest.mark.parametrize("quote", ['"', "'", "`"]) +def test_unterminated_string_literals_are_read_in_linear_time(quote: str) -> None: + started = time.monotonic() + parse_assignments(quote + ("\\" + quote) * 20_000) + assert time.monotonic() - started < 1.5 + + +def test_random_script_text_never_raises_inside_the_renderer() -> None: + """``script_text`` swallows errors by design, so call the real work to see them.""" + import random + + from fnd.extract._html import parse + from fnd.extract._script_text import _script_blocks # pyright: ignore[reportPrivateUsage] + + pieces = [ + "var A = {", + "}", + "[", + "]", + '"x y z w"', + "'", + "`", + "\\", + "/*", + "*/", + "//", + ",", + ":", + ";", + ] + pieces += [ + "\\ud83d", + "\\ude80", + "\\u001b[1m", + "FLOWS.a = ", + "say: ", + "steps: [", + "C(1)", + "null", + "", + ] + rng = random.Random(7) + for _ in range(400): + src = " ".join(rng.choice(pieces) for _ in range(rng.randint(1, 60))) + root = parse(f"".encode()) + for blocks in _script_blocks(root).values(): + for b in blocks: + (b.text + "".join(map("".join, b.rows))).encode("utf-8") + + +def _timed(tmp_path: Path, script: str, seconds: float = 5) -> list[str]: + started = time.monotonic() + chunks = _extract(tmp_path, script) + assert time.monotonic() - started < seconds + assert sum(map(len, chunks)) < 6_000_000 + return chunks + + +def test_a_long_run_of_word_characters_does_not_slow_assignment_search(tmp_path: Path) -> None: + _timed(tmp_path, 'var d = { bdata: "' + "A" * 60_000 + '" };', 3) + + +def test_a_shared_big_string_in_steps_lines_and_chips_is_bounded(tmp_path: Path) -> None: + big = "x-" * 200_000 + refs = ", ".join(["S"] * 1200) + script = ( + f'var S = "{big}"; var F = {{}}; F.d = {{ code: [S], steps: [' + f"{{ say: 'See the listing run.', line: [{', '.join(['0'] * 1200)}]," + f" views: [{{ type: 'term', lines: [{', '.join(['{ o: S }'] * 1200)}] }}," + f" {{ type: 'rows', items: [{refs}] }}] }}] }};" + ) + _timed(tmp_path, script) + + +def test_a_table_header_wider_than_its_rows_cannot_pad_without_limit(tmp_path: Path) -> None: + cols = ", ".join(['"c"'] * 10_000) + rows = ", ".join(["E"] * 5_000) + script = ( + f"var E = []; var F = {{}}; F.d = {{ steps: [{{ say: 'A wide table.'," + f" views: [{{ type: 'table', cols: [{cols}], rows: [{rows}] }}] }}] }};" + ) + _timed(tmp_path, script) + + +def test_many_keys_sharing_one_nested_list_in_a_record_are_bounded(tmp_path: Path) -> None: + lines = ['var A0 = ["w"];', f"var A1 = [{', '.join(['A0'] * 1000)}];"] + lines.append(f"var A2 = [{', '.join(['A1'] * 90)}];") + keys = ", ".join(f"k{i}: A2" for i in range(1000)) + lines.append(f'var T = [{{ {keys} }}, {{ k0: "second row" }}];') + _timed(tmp_path, "\n".join(lines)) + + +def test_a_nested_list_of_strings_in_a_record_is_checked_in_linear_time(tmp_path: Path) -> None: + words = ", ".join(f'"w{i}"' for i in range(60_000)) + _timed(tmp_path, f"var DATA = [{{ a: 'x', v: [[{words}]], d: {{ n: 1 }} }}, {{ a: 'y' }}];") + + +def test_a_shared_string_visited_again_and_again_is_bounded(tmp_path: Path) -> None: + refs = ", ".join(["S"] * 800) + _timed(tmp_path, f'var S = "{"x-" * 100_000}"; var A = [{refs}];') + + +def test_a_heading_that_starts_with_a_comment_keeps_the_page(tmp_path: Path) -> None: + p = tmp_path / "page.html" + p.write_text( + "

Installation guide

Body text.

", + encoding="utf-8", + ) + assert "Installation guide" in "\n".join(c.body for c in extract(p)) + + +_DOCS = [ + ("application/json", '{"x": "Demo"}'), + ( + "text/javascript", + "var F = {}; F.d = { steps: [{ say: 'A code view.', views: " + "[{ type: 'code', body: '' }] }] };", + ), + ("text/javascript", "var R = [{ a: '', b: 'one' }, { a: '', b: 'two' }];"), +] + + +@pytest.mark.parametrize(("kind", "script"), _DOCS) +def test_a_whole_document_inside_script_data_cannot_fail_the_page( + tmp_path: Path, kind: str, script: str +) -> None: + p = tmp_path / "page.html" + p.write_text( + f"

Doc

Ordinary page text stays searchable.

" + f'', + encoding="utf-8", + ) + assert "Ordinary page text stays searchable." in "\n".join(c.body for c in extract(p)) + + +def test_thousands_of_out_views_in_one_step_are_linear(tmp_path: Path) -> None: + views = ",".join(f"{{type:'out',cmd:'c{i}'}}" for i in range(65_000)) + _timed(tmp_path, f"var F = {{ steps: [{{ say: 'Many outputs.', views: [{views}] }}] }};", 8) + + +def test_a_shared_list_of_line_numbers_in_every_step_is_bounded() -> None: + """The renderer is fed directly: the parser's own cap would otherwise end this first.""" + from fnd.extract._script_text import _Renderer # pyright: ignore[reportPrivateUsage] + + step = {"say": "<", "line": [5] * 80_000} # out of range: no text reaches the character budget + renderer = _Renderer({}) + started = time.monotonic() + renderer.render({"code": ["x"], "steps": [step] * 8_000}) + assert time.monotonic() - started < 3 + + +def test_a_covered_nested_field_scan_is_charged_against_the_budget(tmp_path: Path) -> None: + z = ", ".join(["{ q: 'zq' }"] * 200) + fields = ", ".join(f"n{i}: Z" for i in range(800)) + script = f"var Z = [{z}]; var R = [{{ a: '{'w' * 1_000_000} zq', b: 'x', {fields} }}, {{ a: 'second' }}];" + _timed(tmp_path, script, 8) + + +def test_pathological_cell_text_does_not_stall_the_cell_rules(tmp_path: Path) -> None: + link_runs = "[" * 100_000 + breaks = "x" + " " * 100_000 + "
" + script = f"var R = [{{ a: '{link_runs}', b: 'one' }}, {{ a: '{breaks}', b: 'two' }}];" + _timed(tmp_path, script, 4) + + +def test_nested_tables_are_rendered_once_not_once_per_level(tmp_path: Path) -> None: + depth, filler = 62, "a" * 40_000 + html = "
" * depth + filler + "
" * depth + p = tmp_path / "page.html" + p.write_text(f"{html}", encoding="utf-8") + started = time.monotonic() + assert "aaaa" in "\n".join(c.body for c in extract(p)) + assert time.monotonic() - started < 4 + + +def test_marking_blocks_keeps_deeply_nested_inline_markup_linear(tmp_path: Path) -> None: + html = "
" + "" * 250 + "
" * 70_000 + "

end

" + "
" * 250 + "
" + p = tmp_path / "page.html" + p.write_text(f"{html}", encoding="utf-8") + started = time.monotonic() + assert "end" in "\n".join(c.body for c in extract(p)) + assert time.monotonic() - started < 5 + + +def test_a_fragment_nested_as_deep_as_the_page_still_indexes_the_page(tmp_path: Path) -> None: + flow = '"body": "' + "" * 250 + "inner fragment words here" + "" * 250 + '"' + script = ( + 'FLOWS.fs = {"steps":[{"say":"This step narrates the flow here.",' + f'"views":[{{"type":"flow",{flow}}}]}}]}};' + ) + page = "
" * 250 + "
x
" + "
" * 250 + p = tmp_path / "page.html" + p.write_text( + f"

Ordinary page text stays searchable.

{page}" + f"", + encoding="utf-8", + ) + assert "Ordinary page text stays searchable." in "\n".join(c.body for c in extract(p)) + + +def test_a_large_honest_table_is_kept_whole(tmp_path: Path) -> None: + rows = ",".join(f"{{a:'R{i}',b:'x',c:'y'}}" for i in range(80_000)) + p = tmp_path / "page.html" + p.write_text( + f"

Doc

", + encoding="utf-8", + ) + started = time.monotonic() + chunks = list(extract(p)) + assert time.monotonic() - started < 15 + assert "R79999" in "\n".join(c.body for c in chunks) + + +def test_a_forged_result_line_is_ignored() -> None: + from fnd.extract import _js_eval + + assert _js_eval._read("@@m@@[1, 2, 3]", "@@m@@") == {} # pyright: ignore[reportPrivateUsage] + + +def test_a_large_honest_data_page_is_not_quadratic(tmp_path: Path) -> None: + notes = ", ".join( + f'"Sentence number {i} says something distinct about harbour works."' for i in range(8000) + ) + started = time.monotonic() + _extract(tmp_path, f"var DATA = {{ notes: [{notes}] }};") + assert time.monotonic() - started < 6 diff --git a/tests/test_web_script_eval.py b/tests/test_web_script_eval.py new file mode 100644 index 00000000..7948bfb4 --- /dev/null +++ b/tests/test_web_script_eval.py @@ -0,0 +1,284 @@ +"""Computed page data: evaluated in the OS JavaScript shell when there is one, confined.""" + +from __future__ import annotations + +import subprocess +import time +from pathlib import Path +from typing import Any + +import pytest + +from fnd.extract import _js_eval, extract + +pytestmark = pytest.mark.skipif(_js_eval.find_jsc() is None, reason="no system JavaScript shell") + +_HELPERS = """ +(function () { + var C = function (t, k) { return { t: t, k: k }; }; + var WORDS = ["alpha", "beta", "gamma"]; + var FLOWS = {}; + FLOWS.demo = { steps: [ + { say: "Each word becomes one chip in the row.", + views: [{ type: "rows", label: "words", + items: WORDS.map(function (w) { return C(w, "k"); }) }] }, + { say: "Counting the chips gives the length of the row.", + views: [{ type: "out", cmd: "len(words)", out: "n = " + WORDS.length }] } + ] }; + document.getElementById("never").addEventListener("click", function () {}); +})(); +""" + + +def _md(tmp_path: Path, script: str) -> str: + p = tmp_path / "page.html" + p.write_text( + f'

Doc

Demo

' + f"", + encoding="utf-8", + ) + return "\n\n".join(c.body_md for c in extract(p)) + + +def test_computed_chips_are_recovered(tmp_path: Path) -> None: + md = _md(tmp_path, _HELPERS) + assert "`alpha` `beta` `gamma`" in md + + +def test_computed_output_strings_are_recovered(tmp_path: Path) -> None: + assert "n = 3" in _md(tmp_path, _HELPERS) + + +def test_a_value_shared_between_flows_is_recovered_in_each(tmp_path: Path) -> None: + script = """(function () { + var RED = ["red", "green"].map(function (t) { return { t: t }; }); + var FLOWS = {}; + FLOWS.demo = { steps: [{ say: "First flow shows the shared row.", + views: [{ type: "rows", label: "a", items: RED }] }] }; + FLOWS.other = { steps: [{ say: "Second flow shows the same row.", + views: [{ type: "rows", label: "b", items: RED }] }] }; + })();""" + assert _md(tmp_path, script).count("`red` `green`") == 2 + + +def test_evaluation_can_be_switched_off(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("FND_HTML_EVAL", "0") + md = _md(tmp_path, _HELPERS) + assert "`alpha`" not in md + assert "Each word becomes one chip" in md + + +def test_a_page_cannot_read_local_files(tmp_path: Path) -> None: + secret = tmp_path / "secret.txt" + secret.write_text("hunter2-token", encoding="utf-8") + script = ( + "var FLOWS = {}; FLOWS.demo = { steps: [{ say: 'Reading a file should not work here.'," + f" views: [{{ type: 'out', cmd: 'x', out: readFile('{secret}') }}] }}] }};" + ) + assert "hunter2-token" not in _md(tmp_path, script) + + +def _confined(tmp_path: Path, js: str, timeout: float = 5) -> bytes: + """Run raw JS (no prelude) through the same sandboxed command the evaluator uses.""" + jsc = _js_eval.find_jsc() + assert jsc is not None + root = tmp_path.resolve() + script = root / "raw.js" + script.write_text(js, encoding="utf-8") + return _js_eval._run(_js_eval._command(jsc, script, timeout), str(root), timeout) + + +def test_the_os_sandbox_blocks_writes_even_with_the_shell_function_present(tmp_path: Path) -> None: + target = tmp_path / "pwned.txt" + out = _confined(tmp_path, f"writeFile('{target.resolve()}', 'x'); print('ran');") + assert not target.exists() + assert b"ran" not in out + + +def test_the_os_sandbox_blocks_reading_user_files_and_dynamic_import(tmp_path: Path) -> None: + secret = tmp_path / "secret.txt" + secret.write_text("hunter2-token", encoding="utf-8") + js = ( + f"try {{ print(readFile('{secret.resolve()}')); }} catch (e) {{ print('read blocked'); }}\n" + f"import('{secret.resolve()}').then(function () {{ print('IMPORTED'); }}, function () {{}});\n" + ) + out = _confined(tmp_path, js) + assert b"hunter2-token" not in out + assert b"IMPORTED" not in out + assert b"read blocked" in out + + +def test_the_unsandboxed_shell_could_write_so_the_sandbox_is_what_stops_it(tmp_path: Path) -> None: + target = tmp_path / "control.txt" + script = tmp_path / "control.js" + script.write_text(f"writeFile('{target}', 'x');", encoding="utf-8") + jsc = _js_eval.find_jsc() + assert jsc is not None + _js_eval._run([str(jsc), str(script)], str(tmp_path), 5) + assert target.exists() + + +def test_output_is_capped_while_streaming_not_after_buffering( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(_js_eval, "_MAX_OUTPUT", 1024 * 1024) + out = _confined(tmp_path, "var s = 'x'.repeat(1 << 20); while (true) { print(s); }") + assert len(out) <= 1024 * 1024 + 65536 + 1024 * 1024 + + +def test_a_script_that_allocates_without_limit_is_killed( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(_js_eval, "_MAX_RSS_KB", 256 * 1024) + started = time.monotonic() + _confined(tmp_path, "var a = []; while (true) { a.push(new Array(1 << 20).fill(1)); }", 30) + assert time.monotonic() - started < 10 + + +def test_a_blocked_script_cannot_outlive_a_killed_parent(tmp_path: Path) -> None: + jsc = _js_eval.find_jsc() + assert jsc is not None + script = tmp_path.resolve() / "block.js" + script.write_text("Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0);") + proc = subprocess.Popen( + _js_eval._command(jsc, script, 1), + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + start_new_session=True, + ) + time.sleep(0.5) + proc.kill() # the shell only: what a crashed parent leaves behind + proc.wait() + assert _js_eval._group_rss_kb(proc.pid) > 0 # the script is still running, orphaned + deadline = time.monotonic() + 8 + while time.monotonic() < deadline and _js_eval._group_rss_kb(proc.pid) > 0: + time.sleep(0.25) + assert _js_eval._group_rss_kb(proc.pid) == 0 + + +def test_a_line_separator_in_evaluated_data_does_not_lose_the_evaluation(tmp_path: Path) -> None: + script = ( + "var FLOWS = {}; FLOWS.demo = { steps: [{ say: 'Narration for the step.', views: [" + "{ type: 'out', cmd: 'x', out: [1].map(function () { return 'first\\u2028second'; })[0] }] }] };" + ) + assert "first" in _md(tmp_path, script) + + +def test_a_lone_surrogate_in_evaluated_data_is_valid_text(tmp_path: Path) -> None: + script = ( + "var FLOWS = {}; FLOWS.demo = { steps: [{ say: 'Narration for the step.', views: [" + "{ type: 'out', cmd: 'x', out: [1].map(function () { return '\\ud83d lone after eval'; })[0] }] }] };" + ) + p = tmp_path / "page.html" + p.write_text( + f'

Doc

', + encoding="utf-8", + ) + chunks = list(extract(p)) + for c in chunks: + (c.body + c.body_md + c.heading_path).encode("utf-8") + assert "lone after eval" in "\n".join(c.body for c in chunks) + + +def test_evaluation_still_runs_when_the_prose_lives_in_another_literal(tmp_path: Path) -> None: + script = ( + "var NOTES = ['Each retry waits twice as long as the last one.'];" + "var FLOWS = {}; FLOWS.demo = { steps: NOTES.map(function (n) { return { say: n }; }) };" + ) + assert "Each retry waits twice" in _md(tmp_path, script) + + +def test_nothing_is_left_running_after_an_evaluation( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + pids: list[int] = [] + real = subprocess.Popen + + def spy(*args: Any, **kwargs: Any) -> Any: + proc = real(*args, **kwargs) + pids.append(proc.pid) + return proc + + monkeypatch.setattr(_js_eval.subprocess, "Popen", spy) + _confined(tmp_path, "print('done');") + assert pids + assert _js_eval._group_rss_kb(pids[0]) == 0 + + +def test_a_runaway_script_times_out_and_falls_back_to_static_text( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(_js_eval, "TIMEOUT_SECONDS", 1) + started = time.monotonic() + script = ( + "var FLOWS = {}; FLOWS.demo = { steps: [{ say: 'Static narration still indexes.'," + " views: [{ type: 'rows', label: 'x', items: [1].map(function () { while (true) {} }) }] }] };" + ) + assert "Static narration still indexes." in _md(tmp_path, script) + assert time.monotonic() - started < 6 + + +def _page_md(tmp_path: Path, body: str) -> str: + p = tmp_path / "page.html" + p.write_text(f"

Doc

{body}", encoding="utf-8") + return "\n\n".join(c.body_md for c in extract(p)) + + +_COMPUTED = ( + "var NOTES = { intro: 'Every retry waits twice as long as the last.', " + "points: ['alpha', 'beta'].map(function (w) { return w + ' is one backoff step here'; }) };" +) + + +def test_a_later_render_helper_does_not_lose_the_pages_data(tmp_path: Path) -> None: + helper = "function draw() { var h = ['

', NOTES.intro, '

']; return h.join(''); }" + md = _page_md(tmp_path, f"") + assert "alpha is one backoff step here" in md + + +def test_data_written_to_window_is_kept(tmp_path: Path) -> None: + script = _COMPUTED.replace("var NOTES =", "window.NOTES =") + assert "alpha is one backoff step here" in _page_md(tmp_path, f"") + + +def test_a_page_that_reads_its_own_data_island_evaluates(tmp_path: Path) -> None: + island = '' + script = ( + "var RAW = JSON.parse(document.getElementById('data').textContent);" + "var NOTES = { intro: 'Every retry waits twice as long as the last.', " + "points: RAW.words.map(function (w) { return w + ' is one backoff step here'; }) };" + ) + assert "beta is one backoff step here" in _page_md( + tmp_path, f"{island}" + ) + + +def test_a_throw_after_the_data_is_built_keeps_the_data(tmp_path: Path) -> None: + later = "var LATER = { note: ['never reached by the script'] };" + md = _page_md(tmp_path, f"") + assert "alpha is one backoff step here" in md + + +def test_many_runaway_scripts_share_one_budget_per_file( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(_js_eval, "TIMEOUT_SECONDS", 1) + runs: list[float] = [] + real = _js_eval._run + + def counted(cmd: list[str], cwd: str, timeout: float) -> bytes: + runs.append(timeout) + return real(cmd, cwd, timeout) + + monkeypatch.setattr(_js_eval, "_run", counted) + loop = ( + "var FLOWS = {}; FLOWS.demo = { steps: [{ say: 'Static narration still indexes.', " + "views: [{ type: 'rows', items: [1].map(function () { while (true) {} }) }] }] };" + ) + scripts = "".join(f"" for _ in range(6)) + p = tmp_path / "page.html" + p.write_text(f"

Doc

{scripts}", encoding="utf-8") + started = time.monotonic() + list(extract(p)) + assert runs, "the evaluator never started, so the budget was not exercised" + assert time.monotonic() - started < 4 diff --git a/tests/test_web_script_structure.py b/tests/test_web_script_structure.py new file mode 100644 index 00000000..b7e7f46b --- /dev/null +++ b/tests/test_web_script_structure.py @@ -0,0 +1,302 @@ +"""Structure recovered from inline script data: steps, fenced output, tables.""" + +from __future__ import annotations + +from pathlib import Path + +from fnd.extract import extract +from fnd.extract.base import Chunk + +_PAGE = """

Guide

+

2Retries

+""" + +_SCRIPT = """ +var KIND = { T: "Transformation: planned only", A: "Action: runs now" }; +var COLS = ["host", "status"]; +var R1 = ["alpha", "200"]; +var R2 = ["beta", "503"]; +var FLOWS = {}; +FLOWS.retry = { + code: ["client = Client(url)", "client.fetch()"], + steps: [ + { line: 0, kind: "T", say: "Build the client. Nothing is sent yet, so no error appears.", + views: [{ type: "rows", items: [1, 2].map(function (n) { return C(String(n)); }) }] }, + { line: 1, kind: "A", say: "Fetching sends the request and the first reply comes back.", + views: [{ type: "table", cols: COLS, rows: [R1, R2], note: "Beta is failing." }, + { type: "out", cmd: "client.fetch()", out: "200 OK\\nretries: 0" }] } + ] +}; +""" + + +def _chunks(tmp_path: Path, script: str = _SCRIPT) -> list[Chunk]: + p = tmp_path / "guide.html" + p.write_text(_PAGE.format(script=script), encoding="utf-8") + return list(extract(p)) + + +def _md(chunks: list[Chunk]) -> str: + return "\n\n".join(c.body_md for c in chunks) + + +def test_steps_are_numbered_and_labelled_with_the_pages_own_kinds(tmp_path: Path) -> None: + md = _md(_chunks(tmp_path)) + assert "**Step 1 · Transformation**" in md + assert "**Step 2 · Action**" in md + assert md.index("Step 1") < md.index("Step 2") + + +def test_a_step_shows_its_code_line_then_its_output_in_the_next_fence(tmp_path: Path) -> None: + md = _md(_chunks(tmp_path)) + assert "```\nclient.fetch()\n```\n\n```\n200 OK\nretries: 0\n```" in md + + +def test_the_program_listing_is_a_fenced_block(tmp_path: Path) -> None: + assert "```\nclient = Client(url)\nclient.fetch()\n```" in _md(_chunks(tmp_path)) + + +def test_table_views_resolve_named_constants(tmp_path: Path) -> None: + md = _md(_chunks(tmp_path)) + assert "| host | status |" in md + assert "| beta | 503 |" in md + assert "Beta is failing." in md + + +def test_unparseable_expressions_do_not_hide_later_steps(tmp_path: Path) -> None: + assert "Fetching sends the request" in _md(_chunks(tmp_path)) + + +def test_a_list_of_records_is_not_numbered_as_steps(tmp_path: Path) -> None: + script = """FLOWS = {}; FLOWS.retry = { legend: [ + { id: 1, description: "Verified in a primary source such as a regulator filing." }, + { id: 2, description: "Secondary source only, so check before quoting it." }] };""" + md = _md(_chunks(tmp_path, script)) + assert "Verified in a primary source" in md + assert "Secondary source only" in md + assert "Step 1" not in md + + +_HUB = """var DATA = { keyStats: [ + { label: "Visitors", value: "12,480", src: { label: "Survey summary", url: "https://example.com/r" } }, + { label: "Sales", value: "3,100", src: { label: "Survey summary", url: null } }], + notes: ["Figures are underlying unless stated otherwise.", "Kiosk sales are excluded from the harbour view."] };""" + + +def test_a_list_of_records_becomes_a_table_with_linked_sources(tmp_path: Path) -> None: + md = _md(_chunks(tmp_path, _HUB)) + assert "| Label | Value | Src |" in md + assert "| Visitors | 12,480 | [Survey summary](https://example.com/r) |" in md + assert "| Sales | 3,100 | Survey summary |" in md + + +def test_markup_inside_record_cells_is_converted_not_shown_raw(tmp_path: Path) -> None: + script = """var STEPS = [ + { n: 1, work: "d1: 1 / 8 = 0.125
d2: 0 / 13 = 0", why: "TF divides the count." }, + { n: 2, work: "Use features as the column.", why: "One vector per row." }];""" + md = _md(_chunks(tmp_path, script)) + for raw in ("", "
", ""): + assert raw not in md + assert "d1: 1 / 8 = **0.125** / d2: 0 / 13 = **0**" in md + assert "Use `features` as the column." in md + body = "\n".join(c.body for c in _chunks(tmp_path, script)) + assert "0.125" in body + assert "**" not in body + + +def test_entities_in_record_cells_are_decoded(tmp_path: Path) -> None: + script = """var CALLS = [ + { n: 1, url: "?page=0&sort=name", why: "The first request loads page zero." }, + { n: 2, url: "?page=1&sort=name", why: "The second request loads page one." }];""" + md = _md(_chunks(tmp_path, script)) + assert "?page=0&sort=name" in md + assert "&" not in md + + +def test_records_with_sparse_or_nested_fields_lose_nothing(tmp_path: Path) -> None: + script = """var DATA = { facts: [ + { label: "Piers", value: "70 berths across the harbour today.", note: "Counted in August." }, + { label: "Moorings", value: "9,000 licensed moorings in total.", + detail: { caveat: "Excludes day-use buoys.", year: "2026" } }] };""" + body = "\n".join(c.body for c in _chunks(tmp_path, script)) + for fragment in ("Counted in August.", "Excludes day-use buoys.", "70 berths across"): + assert fragment in body + + +_CITED = """var DATA = { claims: [ + { text: "Visitors grew eight per cent over the summer.", + srcs: [{ label: "Field notes, p.14", url: "https://example.com/notes.pdf" }, + { label: "Harbour log, p.9", url: "https://example.com/log.pdf" }] }, + { text: "Ferry numbers fell after the timetable change.", srcs: [] }, + { text: "Queues stayed short across both piers.", + srcs: [{ label: "Survey deck, slide 6", url: null }] }] };""" + + +def test_a_records_citations_are_links_in_its_row(tmp_path: Path) -> None: + chunks = _chunks(tmp_path, _CITED) + md = _md(chunks) + assert "[Field notes, p.14](https://example.com/notes.pdf); [Harbour log, p.9]" in md + assert "| Queues stayed short across both piers. | Survey deck, slide 6 |" in md + assert "| Label | Url |" not in md + assert not any(c.heading_path.endswith("Srcs") for c in chunks) + + +def test_a_standalone_citation_list_is_bullet_links_without_its_own_section( + tmp_path: Path, +) -> None: + script = """var DATA = { intro: "Everything here comes from the published survey pack.", + links: [{ label: "Field notes", url: "https://example.com/notes.pdf" }, + { label: "Harbour log", url: "https://example.com/log.pdf" }] };""" + chunks = _chunks(tmp_path, script) + md = _md(chunks) + assert "- [Field notes](https://example.com/notes.pdf)" in md + assert "- [Harbour log](https://example.com/log.pdf)" in md + assert not any(c.heading_path.endswith("Links") for c in chunks) + + +def test_only_the_label_of_a_citation_is_indexed(tmp_path: Path) -> None: + body = "\n".join(c.body for c in _chunks(tmp_path, _CITED)) + assert "Field notes, p.14" in body + assert "https://example.com" not in body + + +def test_unattached_data_sections_become_headed_chunks(tmp_path: Path) -> None: + chunks = _chunks(tmp_path, _HUB) + paths = [c.heading_path for c in chunks] + assert "Guide > Script text > Key stats" in paths + assert "Guide > Script text > Notes" in paths + stats = next(c for c in chunks if c.heading_path.endswith("Key stats")) + assert "| Label | Value | Src |" in stats.body_md + + +def test_data_attached_to_a_page_node_leads_in_bold_without_new_sections(tmp_path: Path) -> None: + script = "FLOWS = {}; FLOWS.retry = " + _HUB.split("DATA = ", 1)[1] + chunks = _chunks(tmp_path, script) + assert [c.heading_path for c in chunks] == ["Guide", "Guide > 2 Retries"] + md = chunks[-1].body_md + assert md.index("**Key stats**") < md.index("| Label |") + + +def test_a_list_of_sentences_becomes_bullets(tmp_path: Path) -> None: + md = _md(_chunks(tmp_path, _HUB)) + assert "- Figures are underlying unless stated otherwise." in md + assert "- Kiosk sales are excluded from the harbour view." in md + + +def test_a_short_code_field_is_an_identifier_not_a_program(tmp_path: Path) -> None: + script = """FLOWS = {}; FLOWS.retry = { goals: [ + { code: "DD", text: "Determine data value in every decision the team makes." }] };""" + md = _md(_chunks(tmp_path, script)) + assert "```" not in md + assert "`DD`" in md + + +def test_a_terminal_view_becomes_one_fenced_transcript(tmp_path: Path) -> None: + script = """FLOWS = {}; FLOWS.retry = { steps: [ + { say: "The prompt shows where you are.", + views: [{ type: "term", lines: [{ p: "$", c: "ls /data" }, { o: "a.csv\\nb.csv" }] }] }] };""" + assert "$ ls /data\na.csv\nb.csv" in _md(_chunks(tmp_path, script)) + + +def test_a_terminal_step_does_not_repeat_the_command_its_transcript_shows(tmp_path: Path) -> None: + script = """FLOWS = {}; FLOWS.retry = { code: ["ls /data"], steps: [ + { line: 0, say: "List the folder before you read anything.", + views: [{ type: "term", lines: [{ p: "$", c: "ls /data" }, { o: "a.csv" }] }] }] };""" + assert _md(_chunks(tmp_path, script)).count("ls /data") == 2 # listing + transcript + + +def test_view_captions_join_into_one_italic_line(tmp_path: Path) -> None: + script = """FLOWS = {}; FLOWS.retry = { steps: [ + { say: "Cut the list into three partitions.", + views: [{ type: "parts", label: "rdd", sub: "RDD in 3 partitions", parts: [] }] }] };""" + assert "*rdd · RDD in 3 partitions*" in _md(_chunks(tmp_path, script)) + + +def test_script_data_only_attaches_to_id_and_data_attributes(tmp_path: Path) -> None: + p = tmp_path / "role.html" + p.write_text( + '

Top

' + '

Later

' + '", + encoding="utf-8", + ) + by_heading = {c.heading_path: c.body for c in extract(p)} + assert "Grouping keeps" in by_heading["Top > Later"] + assert "Grouping keeps" not in by_heading["Top"] + + +def test_broken_script_text_never_raises(tmp_path: Path) -> None: + script = "FLOWS.retry = { steps: [ { say: 'unterminated ], x: ( [ { }" + assert _chunks(tmp_path, script) + + +def test_heading_number_badge_is_separated_from_the_title(tmp_path: Path) -> None: + assert [c.heading_path for c in _chunks(tmp_path)][-1] == "Guide > 2 Retries" + + +_TASKS = """

Guide

    +
  1. Q1Add one to each element.Uses: map

    +
    Show answer
    +
    rdd2 = rdd.map(lambda x: x + 1)
    +rdd2.collect()
    +

    Output

    [2, 3]
    +

    map keeps the record count.

  2. +
  3. Q2Keep the evens.

  4. +
""" + + +def _tasks_md(tmp_path: Path) -> str: + p = tmp_path / "tasks.html" + p.write_text(_TASKS, encoding="utf-8") + return "\n\n".join(c.body_md for c in extract(p)) + + +def test_a_list_item_keeps_its_answer_blocks_as_blocks(tmp_path: Path) -> None: + md = _tasks_md(tmp_path) + assert "```\nrdd2 = rdd.map(lambda x: x + 1)\nrdd2.collect()\n```" in md + assert "```\n[2, 3]\n```" in md + assert "map keeps the record count." in md + assert "**Show answer**" in md # a summary can be the question itself, so it is kept + + +def test_ordered_items_keep_their_numbers_across_block_content(tmp_path: Path) -> None: + md = _tasks_md(tmp_path) + assert "1. **Q1**" in md + assert "2. **Q2**" in md + + +def test_list_items_after_a_block_item_still_render(tmp_path: Path) -> None: + assert "Keep the evens." in _tasks_md(tmp_path) + + +def test_labels_and_trailing_spans_are_spaced_from_their_neighbours(tmp_path: Path) -> None: + md = _tasks_md(tmp_path) + assert "**Q1** Add one to each element. Uses: map" in md + + +def test_a_stage_banner_opens_its_own_section_instead_of_trailing_the_previous( + tmp_path: Path, +) -> None: + p = tmp_path / "stages.html" + p.write_text( + "

Guide

One

First body.

" + '
PairsCounting per key
' + "

Two

Second body.

" + '
NotePlain callout.
', + encoding="utf-8", + ) + by_heading = {c.heading_path: c.body for c in extract(p)} + assert "Counting per key" in by_heading["Guide > Pairs"] + assert "Pairs" not in by_heading["Guide > One"] + assert "Plain callout." in by_heading["Guide > Two"] + + +def test_adjacent_spans_do_not_fuse_words(tmp_path: Path) -> None: + p = tmp_path / "legend.html" + p.write_text( + "

L

  • Action" + "Runs the plan.
", + encoding="utf-8", + ) + assert "Action Runs the plan." in "\n".join(c.body for c in extract(p)) diff --git a/tests/test_web_script_text.py b/tests/test_web_script_text.py new file mode 100644 index 00000000..1011cc05 --- /dev/null +++ b/tests/test_web_script_text.py @@ -0,0 +1,250 @@ +"""HTML pages that build their body from inline script data.""" + +from __future__ import annotations + +from pathlib import Path + +from fnd.extract import extract +from fnd.extract.base import Chunk + +ANCHOR = "blue penguin sandwich" + +_PAGE = """Retry guide +

Retry guide

+

Backoff

+

Jitter

+""" + + +def _chunks(tmp_path: Path, script: str) -> list[Chunk]: + p = tmp_path / "guide.html" + p.write_text(_PAGE.format(script=script), encoding="utf-8") + return list(extract(p)) + + +def _body_under(chunks: list[Chunk], heading: str) -> str: + return "\n".join(c.body for c in chunks if c.heading_path.endswith(heading)) + + +_FLOWS = f""" +var FLOWS = {{}}; +FLOWS.backoff = {{ + code: ['client.retry(attempts=5)'], + steps: [ + {{ line: 0, say: "Each retry waits twice as long as the previous one: {ANCHOR}.", + views: [{{ type: "out", cmd: "client.retry()", out: "waited 4 seconds" }}] }} + ] +}}; +FLOWS.jitter = {{ + steps: [{{ say: 'Random jitter spreads the retries so clients do not stampede together.' }}] +}}; +""" + + +def test_script_narration_lands_under_the_section_that_owns_it(tmp_path: Path) -> None: + chunks = _chunks(tmp_path, _FLOWS) + assert ANCHOR in _body_under(chunks, "Backoff") + assert "stampede" in _body_under(chunks, "Jitter") + assert "stampede" not in _body_under(chunks, "Backoff") + + +def test_inline_markup_in_script_strings_becomes_markdown(tmp_path: Path) -> None: + md = "\n".join(c.body_md for c in _chunks(tmp_path, _FLOWS)) + assert "`previous`" in md + + +def test_short_labels_and_template_fragments_are_not_indexed(tmp_path: Path) -> None: + script = ( + _FLOWS + + """ +function row(c) { return '
'; } +document.querySelector("section .flow").classList.add("is-ready now"); +var tag = "kind"; +""" + ) + body = "\n".join(c.body for c in _chunks(tmp_path, script)) + assert "chip a k" not in body + assert "is-ready" not in body + assert "kind" not in body + + +def test_titles_inside_render_functions_are_kept_but_data_is_not_repeated(tmp_path: Path) -> None: + script = """var VIEWS = { + hint: "Remember to cap the maximum delay at one minute.", + render: function () { return section('x', 'Timeline of key dates'); } + };""" + body = _body_under(_chunks(tmp_path, script), "Script text") + assert "Timeline of key dates" in body + assert body.count("cap the maximum delay") == 1 + + +def test_call_shaped_strings_outside_data_objects_are_indexed_on_a_data_page( + tmp_path: Path, +) -> None: + script = ( + "var DATA = { note: 'Retries wait longer each time they run.' };" + "show('client.retry_after()'); show('jitter');" + ) + body = _body_under(_chunks(tmp_path, script), "Script text") + assert "client.retry_after()" in body + assert "jitter" not in body + + +def test_scripting_calls_on_an_ordinary_page_are_not_indexed(tmp_path: Path) -> None: + script = "setTimeout(function () { window.close(); }, 1); show('window.close()');" + assert _body_under(_chunks(tmp_path, script), "Script text") == "" + + +def test_escaped_quotes_are_unescaped(tmp_path: Path) -> None: + script = ( + 'FLOWS = {}; FLOWS.jitter = { say: "Call option(\\"header\\", True) before reading." };' + ) + assert 'option("header", True)' in _body_under(_chunks(tmp_path, script), "Jitter") + + +def test_repeated_strings_are_indexed_once(tmp_path: Path) -> None: + script = ( + "FLOWS = {}; FLOWS.backoff = {" + ' a: "Double the delay after every failed attempt.",' + ' b: "Double the delay after every failed attempt." };' + ) + body = _body_under(_chunks(tmp_path, script), "Backoff") + assert body.count("Double the delay") == 1 + + +def test_script_text_without_a_matching_node_is_kept_under_its_own_heading( + tmp_path: Path, +) -> None: + script = 'var NOTES = { hint: "Remember to cap the maximum delay at one minute." };' + chunks = _chunks(tmp_path, script) + assert "cap the maximum delay" in _body_under(chunks, "Script text") + + +def test_external_scripts_and_pages_without_script_data_are_unchanged(tmp_path: Path) -> None: + p = tmp_path / "plain.html" + p.write_text( + '

Doc

Plain prose here.

', + encoding="utf-8", + ) + chunks = list(extract(p)) + assert [c.heading_path for c in chunks] == ["Doc"] + + +def _md(chunks: list[Chunk]) -> str: + return "\n\n".join(c.body_md for c in chunks) + + +def test_decimal_negative_and_exponent_numbers_are_data(tmp_path: Path) -> None: + script = "var R = [{ name: 'Profit', delta: -3.5, ratio: 0.125, big: 1e5 }, { name: 'Loss', delta: 2 }];" + md = _md(_chunks(tmp_path, script)) + assert "| Profit | -3.5 | 0.125 | 100000.0 |" in md + + +def test_decimals_in_a_json_island_are_data(tmp_path: Path) -> None: + p = tmp_path / "page.html" + p.write_text( + '

Doc

", + encoding="utf-8", + ) + md = "\n\n".join(c.body_md for c in extract(p)) + assert "| Harbour | -33.86 |" in md + + +def test_a_commented_out_assignment_does_not_hide_the_data_after_it(tmp_path: Path) -> None: + script = ( + "// cfg.old = [retired]\n" + "var DATA = { items: ['Retries wait longer every time.', 'Jitter spreads the retries.'] };" + ) + assert "- Retries wait longer every time." in _md(_chunks(tmp_path, script)) # as data + + +def test_an_assignment_inside_a_string_is_not_data(tmp_path: Path) -> None: + from fnd.extract._js_data import parse_assignments + + found = parse_assignments("var s = 'a.b = [1]'; var DATA = { note: 'x' };") + assert [(a.base, a.name) for a in found] == [("", "DATA")] + + +def test_a_bare_global_assignment_is_data(tmp_path: Path) -> None: + script = "DATA = { items: ['First point about the retries here.', 'Second point about jitter here.'] };" + md = _md(_chunks(tmp_path, script)) + assert "- First point about the retries here." in md + + +def test_an_empty_data_attribute_does_not_claim_the_loose_prose(tmp_path: Path) -> None: + p = tmp_path / "page.html" + p.write_text( + "

Doc

Unrelated

Widget.
" + "", + encoding="utf-8", + ) + under = {c.heading_path: c.body for c in extract(p)} + assert "A loose sentence" not in under["Doc > Unrelated"] + assert "A loose sentence" in under["Doc > Script text"] + + +def test_a_json_island_is_placed_in_the_section_where_it_sits(tmp_path: Path) -> None: + island = '{"items": ["Opinion holder is who expresses it.", "Target is what it is about."]}' + p = tmp_path / "page.html" + p.write_text( + "

Doc

" + f'

Parts

' + "

Later

Unrelated section text.

", + encoding="utf-8", + ) + under = {c.heading_path: c.body for c in extract(p)} + assert "Opinion holder is who expresses it." in under["Doc > Parts"] + assert "Doc > Script text" not in under + + +def test_a_large_island_gets_sections_below_the_one_it_sits_in(tmp_path: Path) -> None: + notes = ", ".join( + f'"Harbour note number {i} about the berths and the tides."' for i in range(80) + ) + marks = notes.replace("Harbour note", "Pier marker") + island = f'{{"limits": [{notes}], "markers": [{marks}]}}' + p = tmp_path / "page.html" + p.write_text( + f'

Doc

Data

", + encoding="utf-8", + ) + paths = [c.heading_path for c in extract(p)] + assert "Doc > Data > Limits" in paths + assert "Doc > Data > Markers" in paths + + +def test_literals_inside_functions_that_do_not_run_at_load_are_marked_local() -> None: + from fnd.extract._js_data import parse_assignments + + src = """ + var TOP = [1]; + if (ready) { var BLOCK = [2]; } + (function () { var IIFE = [3]; })(); + (() => { var ARROW_IIFE = [4]; })(); + !function () { var BANG = [5]; }(); + function draw() { var DRAW = [6]; } + var cb = function () { var ASSIGNED = [7]; }; + items.forEach(function (x) { var CALLBACK = [8]; }); + items.map((x) => { var ARROW_CB = [9]; }); + class View { render() { var METHOD = [10]; } } + (function () { function inner() { var NESTED = [11]; } })(); + """ + local = {a.name: a.local for a in parse_assignments(src) if a.name.isupper()} + assert local == { + **dict.fromkeys(("TOP", "BLOCK", "IIFE", "ARROW_IIFE", "BANG"), False), + **dict.fromkeys(("DRAW", "ASSIGNED", "CALLBACK", "ARROW_CB", "METHOD", "NESTED"), True), + } + + +def test_a_lead_in_dropped_by_the_budget_does_not_take_earlier_content() -> None: + from fnd.extract._script_text import _Renderer # pyright: ignore[reportPrivateUsage] + + first = "A first paragraph that fits inside the budget." + renderer = _Renderer({}, chars=len(first)) # no room left for the "B" lead-in + renderer.render({"a": first, "b": ["Another sentence that will not fit at all."]}) + assert [b.text for b in renderer.blocks] == [first]