Coverage for britney2/transaction.py: 95%

1from typing import TYPE_CHECKING, Optional, TypedDict 

2from collections.abc import Iterator 

3 

4if TYPE_CHECKING: 4 ↛ 5line 4 didn't jump to line 5, because the condition on line 4 was never true

5 from . import BinaryPackage, BinaryPackageId, SourcePackage, Suites 

6 

7UndoItem = TypedDict( 

8 "UndoItem", 

9 { 

10 "sources": dict[str, Optional["SourcePackage"]], 

11 "binaries": dict[tuple[str, str], "BinaryPackageId"], 

12 "virtual": dict[tuple[str, str], Optional[set[tuple[str, str]]]], 

13 }, 

14) 

15 

16 

17class MigrationTransactionState(object): 

18 def __init__( 

19 self, 

20 suite_info: "Suites", 

21 all_binaries: dict["BinaryPackageId", "BinaryPackage"], 

22 parent: Optional["MigrationTransactionState"] = None, 

23 ) -> None: 

24 self._suite_info = suite_info 

25 self._all_binaries = all_binaries 

26 self.parent_transaction = parent 

27 self._is_rolled_back = False 

28 self._is_committed = False 

29 self._undo_items: list[tuple[UndoItem, set["BinaryPackageId"]]] = [] 

30 self._pending_child = False 

31 if self.parent_transaction: 

32 # Transactions can only support one child transaction at a time 

33 assert not self.parent_transaction._pending_child 

34 self.parent_transaction._pending_child = True 

35 

36 def add_undo_item( 

37 self, undo: UndoItem, updated_binaries: set["BinaryPackageId"] 

38 ) -> None: 

39 # We do not accept any changes to this transaction while it has a child transaction 

40 # (the undo code does not handle that case correctly) 

41 assert not self._pending_child 

42 self._assert_open_transaction() 

43 self._undo_items.append((undo, updated_binaries)) 

44 

45 def _assert_open_transaction(self) -> None: 

46 assert not self._is_rolled_back and not self._is_committed 

47 p = self.parent_transaction 

48 if p: 

49 p._assert_open_transaction() 

50 

51 @property 

52 def undo_items(self) -> Iterator[tuple[UndoItem, set["BinaryPackageId"]]]: 

53 """Only needed by a _apply_item_to_target_suite for the "hint"-hint case""" 

54 yield from self._undo_items 

55 

56 def commit(self) -> None: 

57 """Commit the transaction 

58 

59 After this call, it is not possible to roll these changes 

60 back (except if there is a parent transaction, which can 

61 still be rolled back). 

62 """ 

63 self._assert_open_transaction() 

64 self._is_committed = True 

65 if self.parent_transaction: 

66 self.parent_transaction._pending_child = False 

67 for undo_item in self._undo_items: 

68 self.parent_transaction.add_undo_item(*undo_item) 

69 

70 def rollback(self) -> None: 

71 """Rollback all recorded changes by this transaction 

72 

73 The parent transaction (if any) will remain unchanged 

74 """ 

75 

76 self._assert_open_transaction() 

77 self._is_rolled_back = True 

78 lundo = self._undo_items 

79 lundo.reverse() 

80 

81 all_binary_packages = self._all_binaries 

82 target_suite = self._suite_info.target_suite 

83 sources_t = target_suite.sources 

84 binaries_t = target_suite.binaries 

85 provides_t = target_suite.provides_table 

86 

87 # Historically, we have done the undo process in "4 steps" 

88 # with the rule that each step must be fully completed for 

89 # each undo-item before starting on the next. 

90 # 

91 # see commit:ef71f0e33a7c3d8ef223ec9ad5e9843777e68133 and 

92 # #624716 for the issues we had when we did not do this. 

93 # 

94 # Today, only STEP 2 and STEP 3 are known to potentially 

95 # clash. If there is a point in merging the loops/steps, 

96 # then it is now feasible. 

97 

98 # STEP 1 

99 # undo all the changes for sources 

100 for undo, updated_binaries in lundo: 

101 for k, v in undo["sources"].items(): 

102 if v is None: 

103 del sources_t[k] 

104 else: 

105 sources_t[k] = v 

106 

107 # STEP 2 

108 # undo all new/updated binaries 

109 # Note this must be completed fully before starting STEP 3 

110 # as it potentially breaks STEP 3 if the two are interleaved. 

111 for _, updated_binaries in lundo: 

112 for pkg_id in updated_binaries: 

113 pkg_name, _, pkg_arch = pkg_id 

114 try: 

115 del binaries_t[pkg_arch][pkg_name] 

116 except KeyError: 

117 continue 

118 

119 target_suite.remove_binary(pkg_id) 

120 

121 # STEP 3 

122 # undo all other binary package changes (except virtual packages) 

123 for undo, updated_binaries in lundo: 

124 for p in undo["binaries"]: 

125 binary, arch = p 

126 binaries_t_a = binaries_t[arch] 

127 pkgdata = all_binary_packages[undo["binaries"][p]] 

128 binaries_t_a[binary] = pkgdata 

129 target_suite.add_binary(pkgdata.pkg_id) 

130 

131 # STEP 4 

132 # undo all changes to virtual packages 

133 for undo, _ in lundo: 

134 for p, value in undo["virtual"].items(): 

135 provided_pkg, arch = p 

136 if value is None: 136 ↛ 137line 136 didn't jump to line 137, because the condition on line 136 was never true

137 del provides_t[arch][provided_pkg] 

138 else: 

139 provides_t[arch][provided_pkg] = value 

140 

141 if self.parent_transaction: 

142 self.parent_transaction._pending_child = False 

143 

144 @property 

145 def is_rolled_back(self) -> bool: 

146 return self._is_rolled_back 

147 

148 @property 

149 def is_committed(self) -> bool: 

150 return self._is_committed