Coverage for /pythoncovmergedfiles/medio/medio/src/turbohtml/tools/fuzz/reduce.py: 26%

1"""Structural deletion keeps language constructs intact before character reduction.""" 

2 

3from __future__ import annotations 

4 

5import json 

6import os 

7import shutil 

8import subprocess 

9from pathlib import Path 

10from typing import TYPE_CHECKING, Final, Literal, TypedDict, cast 

11 

12from tinycss2 import parse_blocks_contents, parse_component_value_list, parse_stylesheet, serialize 

13from tinycss2.ast import AtRule, Declaration, ParseError, QualifiedRule 

14 

15from turbohtml import Element, parse, parse_fragment 

16 

17if TYPE_CHECKING: 

18 from collections.abc import Callable, Iterator 

19 from typing import TextIO 

20 

21 from tinycss2.ast import Node 

22 

23_NODE_DIR: Final = Path(__file__).resolve().parents[1] / "bench" / "node" 

24 

25 

26def minimize( 

27 text: str, reproduces: Callable[[str], bool], syntax: Literal["html", "css", "js"], budget: int = 600 

28) -> str: 

29 """Accept shorter candidates only when the original finding predicate holds.""" 

30 if budget <= 0: 

31 return text 

32 calls = 0 

33 

34 def check(candidate: str) -> bool: 

35 nonlocal calls 

36 if calls >= budget: 

37 return False 

38 calls += 1 

39 return reproduces(candidate) 

40 

41 if syntax == "js": 

42 return _js(text, check, budget) 

43 candidates = _html if syntax == "html" else _css 

44 while calls < budget: 

45 for candidate in candidates(text): 

46 if len(candidate) < len(text) and check(candidate): 

47 text = candidate 

48 break 

49 else: 

50 break 

51 return text 

52 

53 

54def _js(text: str, reproduces: Callable[[str], bool], budget: int) -> str: 

55 if (node := shutil.which("node")) is None or not (_NODE_DIR / "node_modules" / "uglify-js").is_dir(): 

56 msg = f"node and npm ci in {_NODE_DIR} are required for JS reduction" 

57 raise FileNotFoundError(msg) 

58 with subprocess.Popen( 

59 [node, str(Path(__file__).with_name("js_reduce_runner.js"))], 

60 stdin=subprocess.PIPE, 

61 stdout=subprocess.PIPE, 

62 text=True, 

63 encoding="utf-8", 

64 env={key: value for key, value in os.environ.items() if key not in {"LD_PRELOAD", "DYLD_INSERT_LIBRARIES"}}, 

65 ) as process: 

66 stdin = cast("TextIO", process.stdin) 

67 stdout = cast("TextIO", process.stdout) 

68 stdin.write(json.dumps({"text": text, "budget": budget}) + "\n") 

69 stdin.flush() 

70 while line := stdout.readline(): 

71 reply = cast("_Reply", json.loads(line)) 

72 if reply["kind"] == "done": 

73 return reply["text"] 

74 stdin.write(json.dumps(reproduces(reply["text"])) + "\n") 

75 stdin.flush() 

76 msg = f"JS reducer exited without a result: {process.returncode}" 

77 raise RuntimeError(msg) 

78 

79 

80def _html(text: str) -> Iterator[str]: 

81 for root, inner in ((parse(text), False), (parse_fragment(text), True)): 

82 nodes = list(root.descendants) 

83 for index, node in enumerate(nodes): 

84 replacement = parse_fragment(text) if inner else parse(text) 

85 list(replacement.descendants)[index].extract() 

86 yield replacement.serialize(inner=inner) 

87 if isinstance(node, Element): 

88 for name in node.attrs: 

89 replacement = parse_fragment(text) if inner else parse(text) 

90 target = cast("Element", list(replacement.descendants)[index]) 

91 del target.attrs[name] 

92 yield replacement.serialize(inner=inner) 

93 

94 

95def _css(text: str) -> Iterator[str]: 

96 yield from _css_delete(cast("list[Node]", parse_stylesheet(text))) 

97 yield from _css_delete(cast("list[Node]", parse_blocks_contents(text))) 

98 

99 

100def _css_delete(nodes: list[Node]) -> Iterator[str]: 

101 if any(isinstance(node, ParseError) for node in nodes): 

102 return 

103 for index, node in enumerate(nodes): 

104 if isinstance(node, Declaration): 

105 yield serialize(nodes[:index] + nodes[index + 1 :]) 

106 elif isinstance(node, (AtRule, QualifiedRule)) and node.content is not None: 

107 content = node.content 

108 for candidate in _css_delete(cast("list[Node]", parse_blocks_contents(content))): 

109 node.content = parse_component_value_list(candidate) 

110 yield serialize(nodes) 

111 node.content = content 

112 

113 

114class _Reply(TypedDict): 

115 kind: Literal["candidate", "done"] 

116 text: str 

117 

118 

119__all__ = ["minimize"]