Coverage for /pythoncovmergedfiles/medio/medio/src/turbohtml/tools/fuzz/parser_targets.py: 92%

1"""Incremental public entry points must preserve the one-shot result.""" 

2 

3from __future__ import annotations 

4 

5import json 

6from typing import TYPE_CHECKING, Final, NamedTuple 

7 

8from turbohtml import ( 

9 Document, 

10 Element, 

11 Formatter, 

12 Html, 

13 IncrementalParser, 

14 Indent, 

15 Minify, 

16 Node, 

17 Token, 

18 Tokenizer, 

19 TokenType, 

20 parse, 

21 parse_fragment, 

22 tokenize, 

23) 

24 

25from .atheris_header import BYTEWISE, Header, feed_chunks 

26from .atheris_invariants import assert_invariant 

27from .atheris_registry import Target 

28from .round_trip_oracles import html_check, span_check, xpath_entry_check 

29 

30if TYPE_CHECKING: 

31 from collections.abc import Callable 

32 

33 from turbohtml import SourceLocation 

34 

35__all__ = [ 

36 "document_observation", 

37 "fragment_observation", 

38 "incremental_observation", 

39 "parser_targets", 

40 "token_observation", 

41] 

42 

43 

44# One node-set, one scalar and one attribute expression reach each XPath result shape on the fuzzed tree. 

45_XPATH_ENTRIES: Final = ("//*", "count(//*)", "//*/@*") 

46 

47 

48def parser_targets() -> tuple[Target, ...]: 

49 """Keep document and fragment grammar contexts distinct.""" 

50 return ( 

51 Target( 

52 "html-document", 

53 _document, 

54 ("turbohtml.parse", "turbohtml.Document", "turbohtml.Node"), 

55 (UnicodeDecodeError,), 

56 ), 

57 Target( 

58 "html-fragment", 

59 _fragment, 

60 ( 

61 "turbohtml.parse_fragment", 

62 "turbohtml.Element", 

63 "turbohtml.Html", 

64 "turbohtml.Formatter", 

65 "turbohtml.Indent", 

66 "turbohtml.Minify", 

67 ), 

68 (UnicodeDecodeError,), 

69 ), 

70 Target( 

71 "html-incremental", 

72 _incremental, 

73 ("turbohtml.IncrementalParser",), 

74 (UnicodeDecodeError,), 

75 _incremental_chunks, 

76 ), 

77 Target( 

78 "html-tokenizer", 

79 _tokens, 

80 ("turbohtml.tokenize", "turbohtml.Tokenizer", "turbohtml.Token", "turbohtml.TokenType"), 

81 (UnicodeDecodeError,), 

82 _token_chunks, 

83 ), 

84 ) 

85 

86 

87def _document(data: bytes) -> None: 

88 document_observation(data) 

89 

90 

91def _fragment(data: bytes) -> None: 

92 fragment_observation(data) 

93 

94 

95def _incremental(data: bytes) -> None: 

96 incremental_observation(data) 

97 

98 

99def _incremental_chunks(data: bytes, header: Header) -> None: 

100 incremental_observation(data, header) 

101 

102 

103def _tokens(data: bytes) -> None: 

104 token_observation(data) 

105 

106 

107def _token_chunks(data: bytes, header: Header) -> None: 

108 token_observation(data, header) 

109 

110 

111def document_observation( 

112 data: bytes, 

113 html: Callable[[str], str | None] = html_check, 

114 spans: Callable[[str], str | None] = span_check, 

115 entries: Callable[[str], str | None] = xpath_entry_check, 

116) -> str: 

117 """Exercise returned document records through serialization consumers, span and XPath entry-point oracles.""" 

118 source: Final = data.decode("utf-8") 

119 document: Final = parse(source) 

120 _require("Document result type", condition=isinstance(document, (Document, Node))) 

121 assert_invariant(html, source) 

122 assert_invariant(spans, source) 

123 for expression in _XPATH_ENTRIES: 

124 assert_invariant(entries, json.dumps({"html": source, "xpath": expression})) 

125 return _serialization(document) 

126 

127 

128def fragment_observation(data: bytes) -> str: 

129 """Grammar fragments use a div context rather than document insertion modes.""" 

130 fragment: Final = parse_fragment(data.decode("utf-8"), "div") 

131 _require("Fragment result type", condition=isinstance(fragment, Element)) 

132 return _serialization(fragment) 

133 

134 

135def incremental_observation(data: bytes, header: Header = BYTEWISE) -> tuple[str, _Positions]: 

136 """ 

137 Empty feeds and splits inside UTF-8 sequences must preserve the one-shot parse and its positions. 

138 

139 The opts bits flip the defaults: 1 feeds decoded text, 2 drops positions, 4 records source locations. 

140 """ 

141 source: Final = data.decode("utf-8") 

142 positions: Final = not header.opts & 2 

143 locations: Final = bool(header.opts & 4) 

144 parser: Final = IncrementalParser(positions=positions, source_locations=locations) 

145 if header.opts & 1: 

146 feed_chunks(parser.feed, source, header.max_chunk) 

147 else: 

148 feed_chunks(parser.feed, data, header.max_chunk) 

149 document: Final = parser.close() 

150 expected: Final = parse(data, positions=positions, source_locations=locations) 

151 observed: Final = _serialization(document) 

152 _require("Incremental document differs", condition=observed == expected.serialize()) 

153 placed: Final = _positions(document) 

154 _require("Incremental positions differ", condition=placed == _positions(expected)) 

155 return observed, placed 

156 

157 

158def token_observation(data: bytes, header: Header = BYTEWISE) -> tuple[_TokenSnapshot, ...]: 

159 """ 

160 Streaming and reset must preserve token fields and source positions. 

161 

162 The opts bits flip the defaults: 1 keeps character references apart, 2 captures source, 4 drops attributes. 

163 """ 

164 source: Final = data.decode("utf-8") 

165 options: Final = { 

166 "resolve_references": not header.opts & 1, 

167 "capture_source": bool(header.opts & 2), 

168 "capture_attributes": not header.opts & 4, 

169 } 

170 tokenizer: Final = Tokenizer(**options) 

171 streamed: Final[list[Token]] = [] 

172 feed_chunks(lambda chunk: streamed.extend(tokenizer.feed(chunk)), source, header.max_chunk) 

173 streamed.extend(tokenizer.close()) 

174 observed: Final = tuple(_token_snapshot(token) for token in streamed) 

175 _require( 

176 "Incremental tokens differ", 

177 condition=observed == tuple(_token_snapshot(token) for token in tokenize(source, **options)), 

178 ) 

179 tokenizer.reset() 

180 _require( 

181 "Reset tokens differ", 

182 condition=tuple(_token_snapshot(token) for token in tokenizer.feed(source)) 

183 + tuple(_token_snapshot(token) for token in tokenizer.close()) 

184 == observed, 

185 ) 

186 return observed 

187 

188 

189def _positions(document: Document) -> _Positions: 

190 return tuple( 

191 (element.source_line, element.source_col, element.source_location) for element in document.iter_elements() 

192 ) 

193 

194 

195def _serialization(node: Node) -> str: 

196 for formatter in Formatter: 

197 for layout in (None, Indent(2)): 

198 options: Final = Html(formatter=formatter, layout=layout) 

199 rendered: Final = node.serialize(options) 

200 _require("Iterator serialization differs", condition="".join(node.serialize_iter(options)) == rendered) 

201 _require( 

202 "Encoded serialization differs", condition=node.encode("utf-8", options) == rendered.encode("utf-8") 

203 ) 

204 minified_options: Final = Html(layout=Minify()) 

205 _require( 

206 "Minified encoding differs", 

207 condition=node.encode("utf-8", minified_options) == node.serialize(minified_options).encode("utf-8"), 

208 ) 

209 return node.serialize() 

210 

211 

212def _token_snapshot(token: Token) -> _TokenSnapshot: 

213 _require("Token result type", condition=isinstance(token, Token)) 

214 _require("Token kind type", condition=isinstance(token.type, TokenType)) 

215 return _TokenSnapshot( 

216 token.type.name, 

217 token.tag, 

218 token.data, 

219 tuple(token.attrs) if token.attrs is not None else None, 

220 token.self_closing, 

221 token.line, 

222 token.col, 

223 ) 

224 

225 

226def _require(message: str, *, condition: bool) -> None: 

227 if not condition: 

228 raise AssertionError(message) 

229 

230 

231class _TokenSnapshot(NamedTuple): 

232 kind: str 

233 tag: str | None 

234 data: str | None 

235 attrs: tuple[tuple[str, str], ...] | None 

236 self_closing: bool 

237 line: int 

238 col: int 

239 

240 

241_Positions = tuple[tuple[int | None, int | None, "SourceLocation | None"], ...]