Coverage for britney2/transaction.py: 92%

1from collections.abc import Iterator 

2from typing import TYPE_CHECKING, Optional, TypedDict 

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 

7 

8class UndoItem(TypedDict): 

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

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

11 virtual: dict[tuple[str, str], set[tuple[str, str]] | None] 

12 

13 

14class MigrationTransactionState: 

15 def __init__( 

16 self, 

17 suite_info: "Suites", 

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

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

20 ) -> None: 

21 self._suite_info = suite_info 

22 self._all_binaries = all_binaries 

23 self.parent_transaction = parent 

24 self._is_rolled_back = False 

25 self._is_committed = False 

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

27 self._pending_child = False 

28 if self.parent_transaction: 

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

30 assert not self.parent_transaction._pending_child 

31 self.parent_transaction._pending_child = True 

32 

33 def add_undo_item( 

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

35 ) -> None: 

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

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

38 assert not self._pending_child 

39 self._assert_open_transaction() 

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

41 

42 def _assert_open_transaction(self) -> None: 

43 assert not self._is_rolled_back and not self._is_committed 

44 if p := self.parent_transaction: 

45 p._assert_open_transaction() 

46 

47 @property 

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

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

50 yield from self._undo_items 

51 

52 def commit(self) -> None: 

53 """Commit the transaction 

54 

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

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

57 still be rolled back). 

58 """ 

59 self._assert_open_transaction() 

60 self._is_committed = True 

61 if self.parent_transaction: 

62 self.parent_transaction._pending_child = False 

63 for undo_item in self._undo_items: 

64 self.parent_transaction.add_undo_item(*undo_item) 

65 

66 def rollback(self) -> None: 

67 """Rollback all recorded changes by this transaction 

68 

69 The parent transaction (if any) will remain unchanged 

70 """ 

71 

72 self._assert_open_transaction() 

73 self._is_rolled_back = True 

74 lundo = self._undo_items 

75 lundo.reverse() 

76 

77 all_binary_packages = self._all_binaries 

78 target_suite = self._suite_info.target_suite 

79 sources_t = target_suite.sources 

80 binaries_t = target_suite.binaries 

81 provides_t = target_suite.provides_table 

82 

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

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

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

86 # 

87 # see commit:ef71f0e33a7c3d8ef223ec9ad5e9843777e68133 and 

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

89 # 

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

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

92 # then it is now feasible. 

93 

94 # STEP 1 

95 # undo all the changes for sources 

96 for undo, updated_binaries in lundo: 

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

98 if v is None: 

99 del sources_t[k] 

100 else: 

101 sources_t[k] = v 

102 

103 # STEP 2 

104 # undo all new/updated binaries 

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

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

107 for _, updated_binaries in lundo: 

108 for pkg_id in updated_binaries: 

109 pkg_name, _, pkg_arch = pkg_id 

110 try: 

111 del binaries_t[pkg_arch][pkg_name] 

112 except KeyError: 

113 continue 

114 

115 target_suite.remove_binary(pkg_id) 

116 

117 # STEP 3 

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

119 for undo, updated_binaries in lundo: 

120 for p in undo["binaries"]: 

121 binary, arch = p 

122 binaries_t_a = binaries_t[arch] 

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

124 binaries_t_a[binary] = pkgdata 

125 target_suite.add_binary(pkgdata.pkg_id) 

126 

127 # STEP 4 

128 # undo all changes to virtual packages 

129 for undo, _ in lundo: 

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

131 provided_pkg, arch = p 

132 if value is None: 132 ↛ 133line 132 didn't jump to line 133 because the condition on line 132 was never true

133 del provides_t[arch][provided_pkg] 

134 else: 

135 provides_t[arch][provided_pkg] = value 

136 

137 if self.parent_transaction: 137 ↛ 138line 137 didn't jump to line 138 because the condition on line 137 was never true

138 self.parent_transaction._pending_child = False 

139 

140 @property 

141 def is_rolled_back(self) -> bool: 

142 return self._is_rolled_back 

143 

144 @property 

145 def is_committed(self) -> bool: 

146 return self._is_committed