Coverage for src / ocarina / dsl / testing_with_railway / chain_actions.py: 38.89%

1"""Chain actions operator for composing Railway action sequences. 

2 

3Provides utilities for composing multiple Railway actions into a single 

4lazy execution unit, solving the "parenthesis hell" problem. 

5 

6Solution: 

7 >>> # Flat, readable syntax 

8 >>> runner = chain_actions( 

9 ... action1.failure(h1).success(h2), 

10 ... action2.failure(h3).success(h4), 

11 ... action3.failure(h5).success(h6) 

12 ... ) 

13 >>> chain = runner.run() # Execute when ready 

14 

15Benefits: 

16- Lazy evaluation: Build chain without executing 

17- Flat syntax: No deep nesting 

18- Automatic short-circuiting: Stops on first failure 

19- Composability: ChainRunner is a value 

20 

21Example: 

22 >>> # Test usage 

23 ... def test_scenario(driver, logger): 

24 ... return [ 

25 ... chain_actions( 

26 ... act(page, step1).failure(log_error).success(log_success), 

27 ... act(page, step2).failure(log_error).success(log_success) 

28 ... ) 

29 ... ] 

30 

31""" 

32 

33from functools import reduce 

34from typing import TYPE_CHECKING, final 

35 

36if TYPE_CHECKING: 

37 from ocarina.custom_types.thunk import Thunk 

38 from ocarina.dsl.testing_with_railway.internals.action_chain import ( 

39 ActionChain, 

40 ActionSuccess, 

41 ) 

42 

43 

44@final 

45class ChainRunner[T]: 

46 """Lazy execution wrapper for Railway action chains. 

47 

48 Encapsulates a sequence of Railway actions that have been composed 

49 but not yet executed. Stores a thunk that executes the chain when 

50 run() is called. 

51 

52 Example: 

53 >>> runner = chain_actions(action1, action2) 

54 >>> chain = runner.run() # Execute when ready 

55 >>> if chain.has_failed(): 

56 ... logger.error("Chain failed") 

57 

58 """ 

59 

60 def __init__(self, *, thunk: Thunk[ActionChain[T]]) -> None: 

61 """Initialize with lazy computation.""" 

62 self._thunk = thunk 

63 

64 def run(self) -> ActionChain[T]: 

65 """Execute the lazy action chain and return result. 

66 

67 Returns: 

68 ActionChain[T] with final state after execution. 

69 

70 Example: 

71 >>> chain = runner.run() 

72 >>> if chain.is_ok(): 

73 ... logger.success("All succeeded") 

74 

75 """ 

76 return self._thunk() 

77 

78 

79def chain_actions[T]( 

80 first: ActionSuccess[T], *rest: ActionSuccess[T] 

81) -> ChainRunner[T]: 

82 """Compose multiple configured actions into lazy execution chain. 

83 

84 Takes ActionSuccess instances (with both handlers configured) and 

85 composes them into a ChainRunner that executes sequentially with 

86 automatic short-circuiting on failure. 

87 

88 Solves "parenthesis hell" by flattening nested .then().execute() chains. 

89 

90 Execution: 

91 1. Execute first action with handlers 

92 2. If failed: short-circuit, return failed chain 

93 3. If succeeded: chain next action with .then() 

94 4. Repeat for all actions 

95 5. Return final chain state 

96 

97 Args: 

98 first: First action (with handlers configured). 

99 *rest: Additional actions in sequence. 

100 

101 Returns: 

102 ChainRunner[T] that lazily executes when run() is called. 

103 

104 Example: 

105 >>> # With short-circuit 

106 ... runner = chain_actions( 

107 ... failing_action.failure(h1).success(h2), 

108 ... never_runs.failure(h3).success(h4) # Skipped 

109 ... ) 

110 >>> chain = runner.run() 

111 >>> assert chain.has_failed() 

112 

113 Example: 

114 >>> runner = chain_actions( 

115 ... action1.failure(h1).success(h2), 

116 ... action2.failure(h3).success(h4) 

117 ... ) 

118 >>> chain = runner.run() 

119 

120 """ 

121 

122 def thunk() -> ActionChain[T]: 

123 """Lazy computation executing the action sequence.""" 

124 

125 def reducer(chain: ActionChain[T], step: ActionSuccess[T]) -> ActionChain[T]: 

126 """Fold function chaining actions with short-circuit logic.""" 

127 if chain.has_failed(): 

128 return chain 

129 

130 return ( 

131 chain.then(step.__action__) 

132 .failure(step.__failure_handler__) 

133 .success(step.__success_handler__) 

134 .execute() 

135 ) 

136 

137 return reduce(reducer, rest, first.execute()) 

138 

139 return ChainRunner(thunk=thunk)