Coverage for src / ocarina / dsl / testing_with_railway / constructors / create_act.py: 20.00%

1"""Low-level POM action operator for Railway Oriented Programming. 

2 

3Provides create_act() as a building block for project-specific action wrappers. 

4NOT meant to be used directly - wrap it in your own act() function with 

5project-specific failure handling. 

6 

7Recommended usage pattern: 

8 >>> # In your project, create a wrapper with custom logic 

9 >>> def act(pom: TPOM, action: Callable[[TPOM], TPOM]) -> ActionStart[TPOM]: 

10 ... def failure_hook(pom: TPOM, exc: Exception) -> Fail: 

11 ... # Detect HTTP error pages 

12 ... title = pom.get_current_title() 

13 ... if ERROR_PAGE_REGEX.match(title): 

14 ... return Fail(error=HttpErrorPageReachedError(title)) 

15 ... return Fail(error=exc) 

16 ... 

17 ... return create_act( 

18 ... pom, action, 

19 ... on_failure=failure_hook, 

20 ... on_run_effect=increment_step_counter 

21 ... ) 

22 ... 

23 >>> # Then use your wrapper 

24 >>> act(page, lambda p: p.click_button()) 

25 ... .failure(log_error) 

26 ... .success(log_success) 

27 ... .execute() 

28""" 

29 

30from typing import TYPE_CHECKING 

31 

32from ocarina.dsl.testing_with_railway.internals.action_chain import ActionStart 

33from ocarina.opinionated.infra.act_counter import ActCounter as ThreadsBasedActCounter 

34from ocarina.railway.result import Fail, Ok, Result 

35 

36if TYPE_CHECKING: 

37 from collections.abc import Callable 

38 

39 from ocarina.custom_types.effect import Effect 

40 from ocarina.custom_types.tpom import TPOM 

41 

42 

43def create_act( 

44 pom: TPOM, 

45 action: Callable[[TPOM], TPOM], 

46 *, 

47 on_failure: Callable[[TPOM, Exception], Fail] | None = None, 

48 on_run_effect: Effect | None = None, 

49 act_counter_effect: Effect | None = None, 

50) -> ActionStart[TPOM]: 

51 """Low-level POM action wrapper for Railway pattern. 

52 

53 Building block for project-specific action wrappers. Do NOT use directly - 

54 create your own act() function that wraps this with project logic. 

55 

56 Wraps a Page Object Model action in Railway Result type: 

57 1. Execute optional side effect (e.g., step counting) 

58 2. Run action on POM 

59 3. On success: Return Ok[TPOM] 

60 4. On failure: Return Fail (or custom failure handler result) 

61 5. Wrap in ActionStart for Railway chain 

62 

63 Args: 

64 pom: Page Object Model instance to operate on. 

65 action: Function performing action on POM, returns POM. 

66 Signature: (TPOM) -> TPOM 

67 on_failure: Optional custom failure handler. 

68 Called when action raises exception. 

69 Signature: (TPOM, Exception) -> Fail 

70 If None, returns Fail(error=exc). 

71 on_run_effect: Optional side effect before action. 

72 Signature: () -> None 

73 act_counter_effect: Optional side effect before action. 

74 Signature: () -> None 

75 

76 Returns: 

77 ActionStart[TPOM]: Railway builder for handler configuration. 

78 

79 Example: 

80 >>> # Project-specific wrapper - RECOMMENDED 

81 ... def act(pom: TPOM, action: Callable[[TPOM], TPOM]) -> ActionStart[TPOM]: 

82 ... def failure_hook(pom: TPOM, exc: Exception) -> Fail: 

83 ... if "404" in pom.get_current_title(): 

84 ... return Fail(error=PageNotFoundError()) 

85 ... return Fail(error=exc) 

86 ... return create_act(pom, action, on_failure=failure_hook) 

87 

88 Note: 

89 All exceptions are caught and converted to Fail, 

90 but you can still use a custom on_failure hook to transform them. 

91 

92 """ 

93 

94 def run_action() -> Result[TPOM]: 

95 """Execute action with error handling.""" 

96 try: 

97 if act_counter_effect: 

98 act_counter_effect() 

99 else: 

100 ThreadsBasedActCounter().incr_act_call_count() 

101 

102 if on_run_effect: 

103 on_run_effect() 

104 result_pom = action(pom) 

105 return Ok(result_pom) 

106 except Exception as exc: # noqa: BLE001 

107 if on_failure: 

108 return on_failure(pom, exc) 

109 return Fail(error=exc) 

110 

111 return ActionStart(run_action)