Coverage for src/pytest_samples/plugin/_broker_base.py: 100%

1import logging as _logging 

2import pytest as _pytest 

3import random as _random 

4import time as _time 

5 

6from abc import ABC as _ABC 

7from arrow import Arrow as _Arrow 

8from datetime import timedelta as _timedelta 

9from pytest import Item as _Item 

10from typing import List as _List, Optional as _Optional 

11 

12from . import _meta 

13 

14 

15_logger = _logging.getLogger(__name__) 

16"""The logger for this module.""" 

17 

18 

19_skip_marker = _pytest.mark.skip( 

20 reason=f"{_meta.PLUGIN_FULL_NAME} timeout expired" 

21) 

22"""The marker to add to an item that should be skipped due to the soft 

23timeout. 

24""" 

25 

26 

27class SamplesBrokerBase(_ABC): 

28 """The base class for broker classes dermining which samples to 

29 execute and in which order. 

30 """ 

31 

32 __slots__ = ( 

33 "_soft_timeout", 

34 "_seed", 

35 "_rng", 

36 "_start_time", 

37 "_past_timeout" 

38 ) 

39 

40 def __init__( 

41 self, soft_timeout: _Optional[_timedelta], seed: _Optional[str] 

42 ) -> None: 

43 """Initialize a `SamplesBrokerBase`. 

44 

45 Args: 

46 soft_timeout (Optional[timedelta]): The time after which the 

47 timeout should occur or None, to disable it. 

48 seed (Optional[str]): The seed to use for the RNG or None 

49 to use a time based seed. 

50 """ 

51 

52 self._soft_timeout = soft_timeout 

53 """Stores the soft timeout timedelta.""" 

54 

55 if seed is None: 

56 _logger.info("No seed provided. Using Epoch.") 

57 seed = str(_time.time_ns()) 

58 _logger.info( 

59 "The RNG seed is %r of type %r.", 

60 seed, 

61 type(seed).__qualname__ 

62 ) 

63 

64 self._seed = seed 

65 """The seed for the RNG.""" 

66 

67 self._rng = _random.Random(seed) 

68 """The RNG to use for random processes.""" 

69 

70 self._start_time: _Optional[_Arrow] = None 

71 """Stores the start time of the test loop. May be unset if there 

72 were no tests. 

73 """ 

74 self._past_timeout: bool = False 

75 """Whether the timeout time has been reached.""" 

76 

77 def _shuffle_items(self, items: _List[_Item]) -> None: 

78 """Shuffle a list of items. 

79 

80 Args: 

81 items (List[Item]): The item list to shuffle. 

82 """ 

83 self._rng.shuffle(items) 

84 

85 @classmethod 

86 def _mark_skip(cls, item: _Item) -> None: 

87 """Mark an item as "to be skipped". 

88 

89 Args: 

90 item (Item): The item to mark. 

91 """ 

92 item.add_marker(_skip_marker) 

93 

94 def pytest_runtest_protocol(self, item: _Item) -> None: 

95 """Pytest hook called in the test loop to test an item. 

96 

97 Args: 

98 item (Item): The item to test. 

99 """ 

100 self._prepare_item_check_timeout(item) 

101 

102 def _prepare_item_check_timeout(self, item: _Item) -> None: 

103 """Called to prepare an item for the test and check if the 

104 timeout has expired. 

105 

106 Args: 

107 item (Item): The item to test. 

108 """ 

109 timeout = self._soft_timeout 

110 if timeout is None: 

111 return 

112 if self._past_timeout: 

113 self._mark_skip(item) 

114 return 

115 st = self._start_time 

116 if st is None: 

117 # This will be set before the first test is run. 

118 self._start_time = _Arrow.now() 

119 _logger.info("Setting start time at %s.", self._start_time) 

120 return 

121 if self.check_timeout_expired(st, timeout): 

122 _logger.info("Timeout after %s.", timeout) 

123 self._past_timeout = True 

124 self._mark_skip(item) 

125 

126 @classmethod 

127 def check_timeout_expired( 

128 cls, start_time: _Arrow, timeout: _timedelta 

129 ) -> bool: 

130 """Check if the timeout has expired given the start time and the 

131 timeout `timedelta`. 

132 

133 Args: 

134 start_time (Arrow): The start time. 

135 timeout (_timedelta): The timedelta after which the timeout 

136 should occur. 

137 

138 Returns: 

139 bool: Whether the timeout has expired. 

140 """ 

141 return (_Arrow.now() - start_time) >= timeout