Coverage for src / ocarina / dsl / testing_with_railway / match_page.py: 0.00%

1"""Conditional branching operator for adaptive pages. 

2 

3Provides match_page() and when() as a DSL extension for handling pages 

4that can be in multiple states (A/B tests, optional banners, random modals, 

5maintenance pages, etc.). 

6 

7Semantics: 

8 - Evaluates each when() condition in order (first match wins) 

9 - Executes the matching branch's ChainRunners sequentially 

10 - If no condition matches: Fail (on failure rail, short-circuits downstream) 

11 - If a branch's ChainRunner fails: Fail (same short-circuit behavior) 

12 - Returns a ChainRunner[Any] — integrates naturally in a scenario's list 

13 

14Railway behavior: 

15 match_page is a ChainRunner like drive_page. 

16 It composes in a scenario list and participates in the same 

17 short-circuit logic: if a previous ChainRunner has failed, 

18 match_page is never executed (caller is responsible for that, 

19 as with all ChainRunners in Ocarina). 

20 

21Example: 

22 >>> def scenario(driver: WebDriver, logger: ILogger) -> TestChain: 

23 ... page = SomePage(driver=driver) 

24 ... 

25 ... return [ 

26 ... drive_page( 

27 ... act(page, open_page) 

28 ... .failure(log_error("Failed to open...")) 

29 ... .success(log_success("Opened!")), 

30 ... ), 

31 ... match_page( 

32 ... branches=[ 

33 ... when( 

34 ... page.has_cookie_banner, 

35 ... name="has_cookie_banner", 

36 ... then=[ 

37 ... drive_page( 

38 ... act(page, dismiss_banner) 

39 ... .failure(log_error("Failed to dismiss...")) 

40 ... .success(log_success("Banner dismissed!")), 

41 ... ), 

42 ... ], 

43 ... ), 

44 ... when( 

45 ... lambda: not page.has_cookie_banner(), 

46 ... name="has_not_cookie_banner", 

47 ... then=[ 

48 ... drive_page( 

49 ... act(page, verify_page) 

50 ... .failure(log_error("Failed to verify...")) 

51 ... .success(log_success("Page verified!")), 

52 ... ), 

53 ... ], 

54 ... ), 

55 ... ], 

56 ... ), 

57 ... drive_page( 

58 ... act(page, do_next_thing) 

59 ... .failure(log_error("Failed...")) 

60 ... .success(log_success("Done!")), 

61 ... ), 

62 ... ] 

63 

64""" 

65 

66from dataclasses import dataclass 

67from typing import TYPE_CHECKING, Any, final 

68 

69from ocarina.custom_errors.test_framework.no_matching_branch import ( 

70 NoMatchingBranchError, 

71) 

72from ocarina.dsl.testing_with_railway.chain_actions import ChainRunner 

73from ocarina.dsl.testing_with_railway.internals.action_chain import ActionChain 

74from ocarina.opinionated.loggers.muted_logger import MutedLogger 

75from ocarina.railway.result import Fail, Ok 

76 

77if TYPE_CHECKING: 

78 from collections.abc import Sequence 

79 

80 from ocarina.custom_types.test_components import TestChain 

81 from ocarina.custom_types.thunk import Thunk 

82 from ocarina.ports.ilogger import ILogger 

83 

84 

85@final 

86@dataclass(frozen=True) 

87class When: 

88 """A conditional branch: a guard and the ChainRunners to execute if it matches. 

89 

90 Attributes: 

91 condition: A thunk returning True if this branch should execute. 

92 name: Name of the branch (logging purposes). 

93 then: ChainRunners to execute sequentially if condition is True. 

94 

95 """ 

96 

97 condition: Thunk[bool] 

98 then: TestChain 

99 name: str = "" 

100 

101 

102when = When 

103 

104 

105def _run_branch(runners: TestChain) -> ActionChain[Any]: 

106 """Execute a branch's ChainRunners sequentially, short-circuiting on failure. 

107 

108 Args: 

109 runners: ChainRunners to execute in order. 

110 

111 Returns: 

112 The final ActionChain — either the first failed one, or the last ok one. 

113 

114 """ 

115 last: ActionChain[Any] | None = None 

116 

117 for runner in runners: 

118 chain = runner.run() 

119 last = chain 

120 if chain.has_failed(): 

121 return chain 

122 

123 if last is None: 

124 return ActionChain(has_failed=False, result=Ok(value=None)) 

125 

126 return last 

127 

128 

129def _match_page_builder( # noqa: ANN202 

130 *, 

131 raised_exceptions: tuple[type[BaseException], ...] = (), 

132): 

133 """Build a match_page function with exception policy. 

134 

135 Args: 

136 raised_exceptions: 

137 Tuple of exception classes that MUST be re-raised during 

138 condition evaluation. All other exceptions are treated as 

139 a failed condition (i.e. equivalent to False). 

140 

141 Returns: 

142 A match_page function. 

143 

144 """ 

145 

146 def _match_page(logger: ILogger, branches: Sequence[When]) -> ChainRunner[Any]: 

147 """Conditional branching operator for adaptive pages. 

148 

149 Evaluates each when() condition in declaration order. 

150 The first branch whose condition returns True is executed. 

151 If no branch matches, the ChainRunner fails with NoMatchingBranchError. 

152 

153 Exception handling policy: 

154 - Exceptions listed in `raise_exceptions` are re-raised immediately. 

155 - All other exceptions raised by branch.condition() are 

156 interpreted as a non-matching condition (False). 

157 

158 Args: 

159 logger: logger. 

160 branches: When instances declared with when(). Evaluated in order. 

161 

162 Returns: 

163 ChainRunner[Any] — composes naturally in a scenario's ChainRunner list. 

164 

165 Raises (via Fail on the railway): 

166 NoMatchingBranchError: 

167 If no when() condition returned True. 

168 

169 Any exception explicitly listed in `raise_exceptions`: 

170 Re-raised during condition evaluation. 

171 

172 """ 

173 

174 def _thunk() -> ActionChain[Any]: 

175 for index, branch in enumerate(branches): 

176 label = branch.name or f"branch[{index}]" 

177 

178 try: 

179 matched = branch.condition() 

180 msg = f"match_page: '{label}' -> {matched}" 

181 logger.debug(msg) 

182 

183 except raised_exceptions as exc: 

184 logger.exception("match_page: raising exception...", exc=exc) # type: ignore[arg-type] 

185 raise 

186 

187 except Exception as exc: 

188 msg = f"match_page: '{label}' raised" 

189 logger.exception(msg, exc=exc) 

190 matched = False 

191 

192 if matched: 

193 msg = f"match_page: '{label}' matched." 

194 logger.info(msg) 

195 return _run_branch(branch.then) 

196 

197 return ActionChain( 

198 has_failed=True, 

199 result=Fail( 

200 error=NoMatchingBranchError( 

201 f"No when() branch matched out of {len(branches)} candidate(s)." 

202 ) 

203 ), 

204 ) 

205 

206 return ChainRunner(thunk=_thunk) 

207 

208 return _match_page 

209 

210 

211def create_match_page( 

212 *, 

213 raised_exceptions: tuple[type[Exception], ...] = (), 

214): 

215 """Create a full-featured match_page function with exception and logging policy. 

216 

217 Args: 

218 raised_exceptions: 

219 Tuple of exception classes that MUST be re-raised during 

220 condition evaluation. All other exceptions are treated as 

221 a failed condition (i.e. equivalent to False). 

222 

223 Returns: 

224 A match_page function, injecting logger. 

225 

226 Raises (via Fail on the railway): 

227 NoMatchingBranchError: 

228 If no when() condition returned True. 

229 

230 Any exception explicitly listed in `raise_exceptions`: 

231 Re-raised during condition evaluation. 

232 

233 Example: 

234 >>> match_page = create_match_page( 

235 ... raise_exceptions=(ValueError, RuntimeError) 

236 ... ) 

237 >>> 

238 >>> runner = match_page( 

239 ... branches=[ 

240 ... when( 

241 ... page.is_in_state_a, 

242 ... name="s_a", 

243 ... then=[ 

244 ... drive_page(...) 

245 ... ] 

246 ... ), 

247 ... when( 

248 ... page.is_in_state_b, 

249 ... name="s_b", 

250 ... then=[ 

251 ... drive_page(...) 

252 ... ] 

253 ... ), 

254 ... ], 

255 ... ) 

256 

257 """ 

258 

259 def _match_page( # noqa: ANN202 

260 *, 

261 logger: ILogger | None = None, 

262 branches: Sequence[When], 

263 ): 

264 _logger = MutedLogger() if logger is None else logger 

265 return _match_page_builder(raised_exceptions=raised_exceptions)( 

266 logger=_logger, 

267 branches=branches, 

268 ) 

269 

270 return _match_page