Coverage for src/birdplan/bird_config/sections/protocols/pipe.py: 100%

1# 

2# SPDX-License-Identifier: GPL-3.0-or-later 

3# 

4# Copyright (c) 2019-2025, AllWorldIT 

5# 

6# This program is free software: you can redistribute it and/or modify 

7# it under the terms of the GNU General Public License as published by 

8# the Free Software Foundation, either version 3 of the License, or 

9# (at your option) any later version. 

10# 

11# This program is distributed in the hope that it will be useful, 

12# but WITHOUT ANY WARRANTY; without even the implied warranty of 

13# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

14# GNU General Public License for more details. 

15# 

16# You should have received a copy of the GNU General Public License 

17# along with this program. If not, see <http://www.gnu.org/licenses/>. 

18 

19"""BIRD pipe protocol configuration.""" 

20 

21from collections.abc import Callable 

22from enum import Enum 

23 

24from ...globals import BirdConfigGlobals 

25from ..base import SectionBase 

26 

27__all__ = ["ProtocolPipe", "ProtocolPipeFilterType"] 

28 

29 

30PipeTableNameType = str | Callable[[str], str] 

31 

32 

33class ProtocolPipeFilterType(Enum): 

34 """ 

35 Pipe protocol filter name type. 

36 

37 NO_FILTER - No filter will be applied. 

38 VERSIONED - Filter will be IP versioned. 

39 UNVERSIONED - Filter will not be IP versioned. 

40 

41 """ 

42 

43 NO_FILTER = 0 

44 VERSIONED = 1 

45 UNVERSIONED = 2 

46 

47 

48# This class cannot be a decendant of the SectionProtocolBase class or we'll have circular imports 

49class ProtocolPipe(SectionBase): # pylint: disable=too-many-instance-attributes 

50 """BIRD pipe protocol configuration.""" 

51 

52 _name_suffix: str 

53 

54 _table_from: PipeTableNameType 

55 _table_to: PipeTableNameType 

56 _table_export: str 

57 _table_import: str 

58 _export_filter_type: ProtocolPipeFilterType 

59 _import_filter_type: ProtocolPipeFilterType 

60 

61 # IP versions we're creating a pipe for 

62 _ipversions: list[str] 

63 

64 def __init__( # noqa: PLR0913,PLR0917 

65 self, 

66 birdconfig_globals: BirdConfigGlobals, 

67 table_from: PipeTableNameType, 

68 table_to: PipeTableNameType, 

69 name: str = "", 

70 table_export: str = "none", 

71 table_import: str = "none", 

72 export_filter_type: ProtocolPipeFilterType = ProtocolPipeFilterType.NO_FILTER, 

73 import_filter_type: ProtocolPipeFilterType = ProtocolPipeFilterType.NO_FILTER, 

74 has_ipv4: bool = True, # noqa: FBT001,FBT002 

75 has_ipv6: bool = True, # noqa: FBT001,FBT002 

76 ) -> None: 

77 """Initialize the object.""" 

78 super().__init__(birdconfig_globals) 

79 

80 # Add a suffix if we have a name 

81 self._name_suffix = "" 

82 if name: 

83 self._name_suffix = f"_{name}" 

84 

85 # Grab table information 

86 self._table_from = table_from 

87 self._table_to = table_to 

88 self._table_export = table_export 

89 self._table_import = table_import 

90 self._export_filter_type = export_filter_type 

91 self._import_filter_type = import_filter_type 

92 

93 # Are we excluding anything? 

94 self._ipversions = [] 

95 if has_ipv4: 

96 self._ipversions.append("4") 

97 if has_ipv6: 

98 self._ipversions.append("6") 

99 

100 def configure(self) -> None: 

101 """Create a pipe protocol.""" 

102 super().configure() 

103 

104 for ipv in self._ipversions: 

105 self.conf.add(f"protocol pipe p_{self.table_from(ipv)}_to_{self.table_to(ipv)}{self.name_suffix} {{") 

106 self.conf.add(f' description "Pipe from {self.t_table_from(ipv)} to {self.t_table_to(ipv)}{self.name_suffix}";') 

107 self.conf.add("") 

108 self.conf.add(f" vrf {self.birdconfig_globals.vrf};") 

109 self.conf.add("") 

110 self.conf.add(f" table {self.t_table_from(ipv)};") 

111 self.conf.add(f" peer table {self.t_table_to(ipv)}{self.name_suffix};") 

112 self.conf.add("") 

113 

114 # Check if we're doing export filtering 

115 if self.export_filter_type == ProtocolPipeFilterType.VERSIONED: 

116 self.conf.add(f" export filter f_{self.table_from(ipv)}_{self.table_to(ipv)}_export;") 

117 elif self.export_filter_type == ProtocolPipeFilterType.UNVERSIONED: 

118 self.conf.add(f" export filter f_{self.table_from()}_{self.table_to()}_export;") 

119 # If not add per normal 

120 else: 

121 self.conf.add(f" export {self.table_export};") 

122 

123 # Check if we're doing import filtering 

124 if self.import_filter_type == ProtocolPipeFilterType.VERSIONED: 

125 self.conf.add(f" import filter f_{self.table_from(ipv)}_{self.table_to(ipv)}_import;") 

126 elif self.import_filter_type == ProtocolPipeFilterType.UNVERSIONED: 

127 self.conf.add(f" import filter f_{self.table_from()}_{self.table_to()}_import;") 

128 # If not add per normal 

129 else: 

130 self.conf.add(f" import {self.table_import};") 

131 

132 self.conf.add("};") 

133 self.conf.add("") 

134 

135 def table_from(self, ipv: str = "") -> str: 

136 """Return table_from with IP version included.""" 

137 

138 # If the table is callable, call it and get the name 

139 table_from = self._table_from(ipv) if callable(self._table_from) else f"{self._table_from}{ipv}" 

140 

141 # If it starts with t_, we need to remove t_ for now 

142 return table_from.removeprefix("t_") 

143 

144 def table_to(self, ipv: str = "") -> str: 

145 """Return table_to with IP version included.""" 

146 

147 # If the table is callable, call it and get the name 

148 table_to = self._table_to(ipv) if callable(self._table_to) else f"{self._table_to}{ipv}" 

149 

150 # If it starts with t_, we need to remove t_ for now 

151 return table_to.removeprefix("t_") 

152 

153 def t_table_from(self, ipv: str) -> str: 

154 r"""Return table_from, some tables don't have t\_ prefixes.""" 

155 table_name = self.table_from(ipv) 

156 # If it is not a master table, add t_ 

157 if not table_name.startswith("master"): 

158 table_name = "t_" + table_name 

159 return table_name 

160 

161 def t_table_to(self, ipv: str) -> str: 

162 r"""Return table_to, some tables don't have t\_ prefixes.""" 

163 table_name = self.table_to(ipv) 

164 # If it is not a master table, add t_ 

165 if not table_name.startswith("master"): 

166 table_name = "t_" + table_name 

167 return table_name 

168 

169 @property 

170 def name_suffix(self) -> str: 

171 """Return our name suffix.""" 

172 return self._name_suffix 

173 

174 @property 

175 def table_export(self) -> str: 

176 """Return that state of us exporting the table.""" 

177 return self._table_export 

178 

179 @property 

180 def table_import(self) -> str: 

181 """Return that state of us importing the table.""" 

182 return self._table_import 

183 

184 @property 

185 def export_filter_type(self) -> ProtocolPipeFilterType: 

186 """Return that state of us exporting the table filtered.""" 

187 return self._export_filter_type 

188 

189 @property 

190 def import_filter_type(self) -> ProtocolPipeFilterType: 

191 """Return that state of us importing the table filtered.""" 

192 return self._import_filter_type