Coverage for /pythoncovmergedfiles/medio/medio/src/turbohtml/tools/fuzz/dom_program.py: 25%

1"""Byte programs preserve live typed registers without lifecycle opcodes.""" 

2 

3from __future__ import annotations 

4 

5import json 

6from dataclasses import dataclass 

7from typing import TYPE_CHECKING, Final, cast 

8 

9from turbohtml import Element 

10 

11if TYPE_CHECKING: 

12 import random 

13 from collections.abc import Callable 

14 

15__all__: Final = [ 

16 "UnsupportedDomProgramError", 

17 "dom_program_check", 

18 "dom_program_controls", 

19 "dom_program_generate", 

20 "dom_program_seeds", 

21 "run_dom_program", 

22] 

23 

24_MAX_BYTES: Final = 1024 

25_MAX_STEPS: Final = 64 

26_MAX_NODES: Final = 16 

27_TAGS: Final = ("p", "span", "section", "div") 

28 

29 

30class UnsupportedDomProgramError(ValueError): 

31 """Reject unsupported encodings before allocating register state.""" 

32 

33 

34def dom_program_check(text: str, run: Callable[[bytes], tuple[str, ...]] | None = None) -> str | None: 

35 """Keep the expected state independent of DOM mutation methods.""" 

36 if len(text) > _MAX_BYTES * 2: 

37 msg: Final = "DOM program exceeds the input limit" 

38 raise UnsupportedDomProgramError(msg) 

39 try: 

40 program: Final = bytes.fromhex(text) 

41 except ValueError as error: 

42 msg: Final = "DOM program must contain hexadecimal bytes" 

43 raise UnsupportedDomProgramError(msg) from error 

44 return ( 

45 None if (run or run_dom_program)(program) == _expected(program) else "DOM program differs from its value model" 

46 ) 

47 

48 

49def dom_program_generate(rng: random.Random) -> str: 

50 """Seed occupied registers before random mutation opcodes.""" 

51 prefix: Final = bytes((0, 0, 0, 0, 0, 1, 0, 0, 0, 2, 0, 0)) 

52 return (prefix + bytes(rng.randrange(256) for _ in range(rng.randrange(1, 49) * 4))).hex() 

53 

54 

55def dom_program_seeds() -> list[str]: 

56 """Ensure every supported opcode executes with live operands.""" 

57 prefix: Final = bytes((0, 0, 0, 0, 0, 1, 0, 0, 0, 2, 0, 0)) 

58 return [ 

59 (prefix + bytes((opcode, operand, operand + 1, operand + 2))).hex() 

60 for opcode in range(10) 

61 for operand in range(10) 

62 ] 

63 

64 

65def dom_program_controls() -> dict[str, bool]: 

66 """Require a wrong-result callback to fail the registered checker.""" 

67 return {"missing operation": dom_program_check(dom_program_seeds()[1], lambda _program: ()) is not None} 

68 

69 

70def run_dom_program(program: bytes) -> tuple[str, ...]: 

71 """Bound state growth while preserving public live-node operands.""" 

72 root: Final = Element("main") 

73 nodes: Final[list[Element]] = [] 

74 strings: Final = ["", "alpha", "beta", "gamma"] 

75 integers: Final = [0, 1, 2, 3] 

76 trace: Final[list[str]] = [_observe(root, nodes)] 

77 for opcode, first, second, third in _instructions(program): 

78 if opcode == 8: 

79 strings[first % 4] = chr(97 + second % 26) * (third % 9) 

80 elif opcode == 9: 

81 integers[first % 4] = second 

82 elif opcode == 0: 

83 if len(nodes) < _MAX_NODES: 

84 node: Final = Element(_TAGS[first % len(_TAGS)]) 

85 root.append(node) 

86 nodes.append(node) 

87 elif nodes: 

88 _apply(root, nodes[first % len(nodes)], opcode, strings[second % 4], integers[third % 4]) 

89 trace.append(_observe(root, nodes)) 

90 return tuple(trace) 

91 

92 

93def _instructions(program: bytes) -> list[tuple[int, int, int, int]]: 

94 bounded: Final = program[:_MAX_BYTES] 

95 return [ 

96 (bounded[index] % 10, bounded[index + 1], bounded[index + 2], bounded[index + 3]) 

97 for index in range(0, min(len(bounded) - len(bounded) % 4, _MAX_STEPS * 4), 4) 

98 ] 

99 

100 

101def _apply(root: Element, node: Element, opcode: int, text: str, integer: int) -> None: 

102 if opcode == 1: 

103 node.tag = _TAGS[integer % len(_TAGS)] 

104 elif opcode == 2: 

105 node.text = text 

106 elif opcode == 3: 

107 node.attrs["data-x"] = text 

108 elif opcode == 4: 

109 if "data-x" in node.attrs: 

110 del node.attrs["data-x"] 

111 elif opcode == 5: 

112 root.append(node) 

113 elif opcode == 6: 

114 root.insert(integer % (len(root.children) + 1), node) 

115 

116 

117def _observe(root: Element, nodes: list[Element]) -> str: 

118 children: Final = cast("list[Element]", list(root.children)) 

119 return json.dumps(( 

120 root.html, 

121 [ 

122 ( 

123 node.tag, 

124 node.text, 

125 node.attrs.get("data-x"), 

126 len(node.attrs), 

127 node.parent == root, 

128 node.previous_sibling == (children[index - 1] if index else None), 

129 node.next_sibling == (children[index + 1] if index + 1 < len(children) else None), 

130 nodes.index(node), 

131 ) 

132 for index, node in enumerate(children) 

133 ], 

134 )) 

135 

136 

137def _expected(program: bytes) -> tuple[str, ...]: 

138 nodes: Final[list[_Item]] = [] 

139 order: Final[list[_Item]] = [] 

140 strings: Final = ["", "alpha", "beta", "gamma"] 

141 integers: Final = [0, 1, 2, 3] 

142 trace: Final = [_model_snapshot(order, nodes)] 

143 for opcode, first, second, third in _instructions(program): 

144 if opcode == 8: 

145 strings[first % 4] = chr(97 + second % 26) * (third % 9) 

146 elif opcode == 9: 

147 integers[first % 4] = second 

148 elif opcode == 0: 

149 if len(nodes) < _MAX_NODES: 

150 item: Final = _Item(_TAGS[first % len(_TAGS)]) 

151 nodes.append(item) 

152 order.append(item) 

153 elif nodes: 

154 _model_apply(order, nodes[first % len(nodes)], opcode, strings[second % 4], integers[third % 4]) 

155 trace.append(_model_snapshot(order, nodes)) 

156 return tuple(trace) 

157 

158 

159def _model_apply(order: list[_Item], item: _Item, opcode: int, text: str, integer: int) -> None: 

160 if opcode == 1: 

161 item.tag = _TAGS[integer % len(_TAGS)] 

162 elif opcode == 2: 

163 item.text = text 

164 elif opcode == 3: 

165 item.attribute = text 

166 elif opcode == 4: 

167 item.attribute = None 

168 elif opcode == 5: 

169 order.remove(item) 

170 order.append(item) 

171 elif opcode == 6: 

172 position: Final = integer % (len(order) + 1) 

173 old: Final = order.index(item) 

174 order.remove(item) 

175 order.insert(position - int(old < position), item) 

176 

177 

178def _model_snapshot(order: list[_Item], nodes: list[_Item]) -> str: 

179 markup: Final = "".join( 

180 f"<{item.tag}" 

181 + (f' data-x="{item.attribute}"' if item.attribute is not None else "") 

182 + f">{item.text}</{item.tag}>" 

183 for item in order 

184 ) 

185 return json.dumps(( 

186 f"<main>{markup}</main>", 

187 [ 

188 (item.tag, item.text, item.attribute, int(item.attribute is not None), True, True, True, nodes.index(item)) 

189 for item in order 

190 ], 

191 )) 

192 

193 

194@dataclass(eq=False) 

195class _Item: 

196 tag: str 

197 text: str = "" 

198 attribute: str | None = None