Coverage for britney2/transaction.py: 95%

1from collections.abc import Iterator 

2from dataclasses import dataclass, field 

3from typing import TYPE_CHECKING, Optional 

4 

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

6 from . import BinaryPackage, BinaryPackageId, SourcePackage, Suites 

7 

8 

9@dataclass(slots=True, frozen=True) 

10class UndoItem: 

11 sources: dict[str, Optional["SourcePackage"]] = field(default_factory=dict) 

12 binaries: dict[tuple[str, str], "BinaryPackageId"] = field(default_factory=dict) 

13 virtual: dict[tuple[str, str], set[tuple[str, str]] | None] = field( 

14 default_factory=dict 

15 ) 

16 

17 

18class MigrationTransactionState: 

19 def __init__( 

20 self, 

21 suite_info: "Suites", 

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

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

24 ) -> None: 

25 self._suite_info = suite_info 

26 self._all_binaries = all_binaries 

27 self.parent_transaction = parent 

28 self._is_rolled_back = False 

29 self._is_committed = False 

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

31 self._pending_child = False 

32 if self.parent_transaction: 

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

34 assert not self.parent_transaction._pending_child 

35 self.parent_transaction._pending_child = True 

36 

37 def add_undo_item( 

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

39 ) -> None: 

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

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

42 assert not self._pending_child 

43 self._assert_open_transaction() 

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

45 

46 def _assert_open_transaction(self) -> None: 

47 assert not self._is_rolled_back and not self._is_committed 

48 if p := self.parent_transaction: 

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.package_name, pkg_id.architecture 

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