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

1"""Literal observer traces detect missed edits and reentrant delivery mistakes.""" 

2 

3from __future__ import annotations 

4 

5import json 

6import re 

7from typing import TYPE_CHECKING, Final, TypeAlias, cast 

8 

9from turbohtml import Element, Node 

10from turbohtml.mutations import MutationObserver, MutationRecord 

11 

12if TYPE_CHECKING: 

13 import random 

14 from collections.abc import Callable 

15 

16 

17def observer_sequence_check(case: str, run: Callable[[str, str], str] | None = None) -> str | None: 

18 """Retain detached wrappers and pin callback batches against an independent model.""" 

19 program, separator, name = case.partition(":") 

20 if not separator or program not in _PROGRAMS or re.fullmatch(r"(?:[ab]|div|g[a-z0-9]{0,15})", name) is None: 

21 raise UnsupportedObserverCaseError(case) 

22 return None if (run or _run)(program, name) == _expected(program, name) else "observer sequence differs from model" 

23 

24 

25def _run(program: str, name: str) -> str: 

26 root: Final = Element("div") 

27 first: Final = Element(name) 

28 second: Final = Element("b") 

29 registers: Final = (root, first, second) 

30 batches: Final[list[list[_Record]]] = [] 

31 callbacks: Final[list[list[_Record]]] = [] 

32 

33 def callback(records: list[MutationRecord], observer: MutationObserver) -> None: 

34 callbacks.append(_records(records, registers)) 

35 if program == "error": 

36 message: Final = "observer callback failed" 

37 raise _ObserverCallbackError(message) 

38 if len(callbacks) == 1: 

39 root.append(second) 

40 observer.deliver() 

41 

42 observer: Final = MutationObserver(callback if program in {"nested", "error"} else None) 

43 observer.observe(root, child_list=True) 

44 root.append(first) 

45 alive = first.tag == name and first.parent == root 

46 error: str | None = None 

47 if program.startswith("reuse"): 

48 batches.append(_records(observer.take_records(), registers)) 

49 for _ in range(int(program[-1])): 

50 first.extract() 

51 alive = alive and first.tag == name and first.parent is None 

52 batches.append(_records(observer.take_records(), registers)) 

53 root.append(first) 

54 alive = alive and first.parent == root 

55 batches.append(_records(observer.take_records(), registers)) 

56 elif program == "replace": 

57 batches.append(_records(observer.take_records(), registers)) 

58 first.replace_with(second) 

59 alive = alive and first.tag == name and first.parent is None and second.parent == root 

60 batches.append(_records(observer.take_records(), registers)) 

61 else: 

62 try: 

63 batches.append(_records(observer.deliver(), registers)) 

64 except _ObserverCallbackError as caught: 

65 error = str(caught) 

66 return json.dumps((root.html, alive, batches, callbacks, _records(observer.take_records(), registers), error)) 

67 

68 

69def _records(records: list[MutationRecord], registers: tuple[Element, Element, Element]) -> list[_Record]: 

70 return [ 

71 ( 

72 record.type, 

73 _register(record.target, registers), 

74 tuple(_register(node, registers) for node in record.added_nodes), 

75 tuple(_register(node, registers) for node in record.removed_nodes), 

76 _register(record.previous_sibling, registers), 

77 _register(record.next_sibling, registers), 

78 record.attribute_name, 

79 record.old_value, 

80 ) 

81 for record in records 

82 ] 

83 

84 

85def _register(node: Node | None, registers: tuple[Element, Element, Element]) -> int | None: 

86 return None if node is None else registers.index(cast("Element", node)) 

87 

88 

89def _expected(program: str, name: str) -> str: 

90 added: Final[_Record] = ("childList", 0, (1,), (), None, None, None, None) 

91 if program.startswith("reuse"): 

92 removed: Final[_Record] = ("childList", 0, (), (1,), None, None, None, None) 

93 return json.dumps(( 

94 f"<div><{name}></{name}></div>", 

95 True, 

96 [[added], *([[removed], [added]] * int(program[-1]))], 

97 [], 

98 [], 

99 None, 

100 )) 

101 if program == "replace": 

102 inserted: Final[_Record] = ("childList", 0, (2,), (), None, 1, None, None) 

103 removed_old: Final[_Record] = ("childList", 0, (), (1,), 2, None, None, None) 

104 return json.dumps(("<div><b></b></div>", True, [[added], [inserted, removed_old]], [], [], None)) 

105 if program == "nested": 

106 nested: Final[_Record] = ("childList", 0, (2,), (), 1, None, None, None) 

107 return json.dumps((f"<div><{name}></{name}><b></b></div>", True, [[added]], [[added], [nested]], [], None)) 

108 return json.dumps((f"<div><{name}></{name}></div>", True, [], [[added]], [], "observer callback failed")) 

109 

110 

111def observer_sequence_generate(rng: random.Random) -> str: 

112 """Bound retained references and edits before invoking callbacks.""" 

113 return f"{rng.choice(_PROGRAMS)}:g{rng.randrange(10000)}" 

114 

115 

116def observer_sequence_seeds() -> list[str]: 

117 """Exercise each bounded program before random generation.""" 

118 return [f"{program}:g{index}" for program in _PROGRAMS for index in range(30)] 

119 

120 

121def observer_sequence_controls() -> dict[str, bool]: 

122 """Require exact edits and callback outcomes before accepting random programs.""" 

123 added: Final[_Record] = ("childList", 0, (1,), (), None, None, None, None) 

124 return { 

125 "missing edit": observer_sequence_check( 

126 "reuse1:a", lambda _program, _name: json.dumps(("<div></div>", True, [], [], [], None)) 

127 ) 

128 is not None, 

129 "missing callback": observer_sequence_check( 

130 "nested:a", lambda _program, _name: json.dumps(("<div><a></a></div>", True, [[added]], [], [], None)) 

131 ) 

132 is not None, 

133 "swallowed callback error": observer_sequence_check( 

134 "error:a", lambda _program, _name: json.dumps(("<div><a></a></div>", True, [[added]], [[added]], [], None)) 

135 ) 

136 is not None, 

137 } 

138 

139 

140class UnsupportedObserverCaseError(ValueError): 

141 """Keep malformed programs separate from failures during valid DOM operations.""" 

142 

143 

144class _ObserverCallbackError(RuntimeError): 

145 """Distinguish the expected callback error from an unrelated DOM failure.""" 

146 

147 

148_Record: TypeAlias = tuple[ 

149 str, int | None, tuple[int | None, ...], tuple[int | None, ...], int | None, int | None, str | None, str | None 

150] 

151_PROGRAMS: Final = ("reuse1", "reuse2", "reuse3", "replace", "nested", "error") 

152 

153 

154__all__ = [ 

155 "UnsupportedObserverCaseError", 

156 "observer_sequence_check", 

157 "observer_sequence_controls", 

158 "observer_sequence_generate", 

159 "observer_sequence_seeds", 

160]