CoCalc -- test_rtsock

GitHub Repository: freebsd/freebsd-src
Path: blob/main/tests/sys/net/routing/test_rtsock_multipath.py
³⁹⁶⁰⁴ views
1
import pytest
2
from atf_python.sys.net.rtsock import RtConst
3
from atf_python.sys.net.rtsock import Rtsock
4
from atf_python.sys.net.rtsock import RtsockRtMessage
5
from atf_python.sys.net.tools import ToolsHelper
6
from atf_python.sys.net.vnet import SingleVnetTestTemplate
7

8

9
class TestRtmMultipath(SingleVnetTestTemplate):
10
    def setup_method(self, method):
11
        method_name = method.__name__
12
        if "multipath4" in method_name:
13
            self.IPV4_PREFIXES = ["192.0.2.1/24"]
14
            self.PREFIX = "128.66.0.0/24"
15
        elif "multipath6" in method_name:
16
            self.IPV6_PREFIXES = ["2001:db8::1/64"]
17
            self.PREFIX = "2001:db8:0:ddbb::/64"
18
        super().setup_method(method)
19
        self.rtsock = Rtsock()
20

21
    def get_prefix_routes(self):
22
        family = "inet6" if ":" in self.PREFIX else "inet"
23
        routes = ToolsHelper.get_routes(family)
24
        return [r for r in routes if r["destination"] == self.PREFIX]
25

26
    @pytest.mark.parametrize(
27
        "gws",
28
        [
29
            pytest.param(["+.10=2", "+.5=3"], id="transition_multi"),
30
            pytest.param(["+.10=2", "+.5=3", "-.10=2"], id="transition_single1"),
31
            pytest.param(["+.10=2", "+.5=3", "-.5=3"], id="transition_single2"),
32
            pytest.param(
33
                ["+.10", "+.11", "+.50", "+.13", "+.145", "+.72"], id="correctness1"
34
            ),
35
            pytest.param(
36
                ["+.10", "+.11", "+.50", "-.50", "+.145", "+.72"], id="correctness2"
37
            ),
38
            pytest.param(["+.10=1", "+.5=2"], id="weight1"),
39
            pytest.param(["+.10=2", "+.5=7"], id="weight2"),
40
            pytest.param(["+.10=13", "+.5=21"], id="weight3_max"),
41
            pytest.param(["+.10=2", "+.5=3", "~.5=4"], id="change_new_weight1"),
42
            pytest.param(["+.10=2", "+.5=3", "~.10=3"], id="change_new_weight2"),
43
            pytest.param(
44
                ["+.10=2", "+.5=3", "+.7=4", "~.10=3"], id="change_new_weight3"
45
            ),
46
            pytest.param(["+.10=2", "+.5=3", "~.5=3"], id="change_same_weight1"),
47
            pytest.param(
48
                ["+.10=2", "+.5=3", "+.7=4", "~.5=3"], id="change_same_weight2"
49
            ),
50
        ],
51
    )
52
    @pytest.mark.require_user("root")
53
    def test_rtm_multipath4(self, gws):
54
        """Tests RTM_ADD with IPv4 dest transitioning to multipath"""
55
        self._test_rtm_multipath(gws, "192.0.2")
56

57
    @pytest.mark.parametrize(
58
        "gws",
59
        [
60
            pytest.param(["+:10=2", "+:5=3"], id="transition_multi"),
61
            pytest.param(["+:10=2", "+:5=3", "-:10=2"], id="transition_single1"),
62
            pytest.param(["+:10=2", "+:5=3", "-:5=3"], id="transition_single2"),
63
            pytest.param(
64
                ["+:10", "+:11", "+:50", "+:13", "+:145", "+:72"], id="correctness1"
65
            ),
66
            pytest.param(
67
                ["+:10", "+:11", "+:50", "-:50", "+:145", "+:72"], id="correctness2"
68
            ),
69
            pytest.param(["+:10=1", "+:5=2"], id="weight1"),
70
            pytest.param(["+:10=2", "+:5=7"], id="weight2"),
71
            pytest.param(["+:10=13", "+:5=21"], id="weight3_max"),
72
            pytest.param(["+:10=13", "+:5=21"], id="weight3_max"),
73
            pytest.param(["+:10=2", "+:5=3", "~:5=4"], id="change_new_weight1"),
74
            pytest.param(["+:10=2", "+:5=3", "~:10=3"], id="change_new_weight2"),
75
            pytest.param(
76
                ["+:10=2", "+:5=3", "+:7=4", "~:10=3"], id="change_new_weight3"
77
            ),
78
            pytest.param(["+:10=2", "+:5=3", "~:5=3"], id="change_same_weight1"),
79
            pytest.param(
80
                ["+:10=2", "+:5=3", "+:7=4", "~:5=3"], id="change_same_weight2"
81
            ),
82
        ],
83
    )
84
    @pytest.mark.require_user("root")
85
    def test_rtm_multipath6(self, gws):
86
        """Tests RTM_ADD with IPv6 dest transitioning to multipath"""
87
        self._test_rtm_multipath(gws, "2001:db8:")
88

89
    def _test_rtm_multipath(self, gws, gw_prefix: str):
90
        desired_map = {}
91
        for gw_act in gws:
92
            # GW format: <+-~>GW[=weight]
93
            if "=" in gw_act:
94
                arr = gw_act[1:].split("=")
95
                weight = int(arr[1])
96
                gw = gw_prefix + arr[0]
97
            else:
98
                weight = None
99
                gw = gw_prefix + gw_act[1:]
100
            if gw_act[0] == "+":
101
                msg = self.rtsock.new_rtm_add(self.PREFIX, gw)
102
                desired_map[gw] = self.rtsock.get_weight(weight)
103
            elif gw_act[0] == "-":
104
                msg = self.rtsock.new_rtm_del(self.PREFIX, gw)
105
                del desired_map[gw]
106
            else:
107
                msg = self.rtsock.new_rtm_change(self.PREFIX, gw)
108
                desired_map[gw] = self.rtsock.get_weight(weight)
109

110
            msg.rtm_flags = RtConst.RTF_GATEWAY
111
            if weight:
112
                msg.rtm_inits |= RtConst.RTV_WEIGHT
113
                msg.rtm_rmx.rmx_weight = weight
114
            # Prepare SAs to check for
115
            desired_sa = {
116
                RtConst.RTA_DST: msg.get_sa(RtConst.RTA_DST),
117
                RtConst.RTA_NETMASK: msg.get_sa(RtConst.RTA_NETMASK),
118
                RtConst.RTA_GATEWAY: msg.get_sa(RtConst.RTA_GATEWAY),
119
            }
120
            self.rtsock.write_message(msg)
121

122
            data = self.rtsock.read_data(msg.rtm_seq)
123
            msg_in = RtsockRtMessage.from_bytes(data)
124
            msg_in.print_in_message()
125
            msg_in.verify(msg.rtm_type, desired_sa)
126
            assert msg_in.rtm_rmx.rmx_weight == self.rtsock.get_weight(weight)
127

128
            routes = self.get_prefix_routes()
129
            derived_map = {r["gateway"]: r["weight"] for r in routes}
130
            assert derived_map == desired_map
131

132
    @pytest.mark.require_user("root")
133
    def test_rtm_multipath4_add_same_eexist(self):
134
        """Tests adding same IPv4 gw to the multipath group (EEXIST)"""
135
        gws = ["192.0.2.10", "192.0.2.11", "192.0.2.11"]
136
        self._test_rtm_multipath_add_same_eexist(gws)
137

138
    @pytest.mark.require_user("root")
139
    def test_rtm_multipath6_add_same_eexist(self):
140
        """Tests adding same IPv4 gw to the multipath group (EEXIST)"""
141
        gws = ["2001:db8::10", "2001:db8::11", "2001:db8::11"]
142
        self._test_rtm_multipath_add_same_eexist(gws)
143

144
    def _test_rtm_multipath_add_same_eexist(self, gws):
145
        for idx, gw in enumerate(gws):
146
            msg = self.rtsock.new_rtm_add(self.PREFIX, gw)
147
            msg.rtm_flags = RtConst.RTF_GATEWAY
148
            try:
149
                self.rtsock.write_message(msg)
150
            except FileExistsError as e:
151
                if idx != 2:
152
                    raise
153
                print("Succcessfully raised {}".format(e))
154

155
    @pytest.mark.require_user("root")
156
    def test_rtm_multipath4_del_unknown_esrch(self):
157
        """Tests deleting non-existing dest from the multipath group (ESRCH)"""
158
        gws = ["192.0.2.10", "192.0.2.11"]
159
        self._test_rtm_multipath_del_unknown_esrch(gws, "192.0.2.7")
160

161
    @pytest.mark.require_user("root")
162
    def test_rtm_multipath6_del_unknown_esrch(self):
163
        """Tests deleting non-existing dest from the multipath group (ESRCH)"""
164
        gws = ["2001:db8::10", "2001:db8::11"]
165
        self._test_rtm_multipath_del_unknown_esrch(gws, "2001:db8::7")
166

167
    @pytest.mark.require_user("root")
168
    def _test_rtm_multipath_del_unknown_esrch(self, gws, target_gw):
169
        for gw in gws:
170
            msg = self.rtsock.new_rtm_add(self.PREFIX, gw)
171
            msg.rtm_flags = RtConst.RTF_GATEWAY
172
            self.rtsock.write_message(msg)
173
        msg = self.rtsock.new_rtm_del(self.PREFIX, target_gw)
174
        msg.rtm_flags = RtConst.RTF_GATEWAY
175
        try:
176
            self.rtsock.write_message(msg)
177
        except ProcessLookupError as e:
178
            print("Succcessfully raised {}".format(e))
179

180
    @pytest.mark.require_user("root")
181
    def test_rtm_multipath4_change_unknown_esrch(self):
182
        """Tests changing non-existing dest in the multipath group (ESRCH)"""
183
        gws = ["192.0.2.10", "192.0.2.11"]
184
        self._test_rtm_multipath_change_unknown_esrch(gws, "192.0.2.7")
185

186
    @pytest.mark.require_user("root")
187
    def test_rtm_multipath6_change_unknown_esrch(self):
188
        """Tests changing non-existing dest in the multipath group (ESRCH)"""
189
        gws = ["2001:db8::10", "2001:db8::11"]
190
        self._test_rtm_multipath_change_unknown_esrch(gws, "2001:db8::7")
191

192
    @pytest.mark.require_user("root")
193
    def _test_rtm_multipath_change_unknown_esrch(self, gws, target_gw):
194
        for gw in gws:
195
            msg = self.rtsock.new_rtm_add(self.PREFIX, gw)
196
            msg.rtm_flags = RtConst.RTF_GATEWAY
197
            self.rtsock.write_message(msg)
198
        msg = self.rtsock.new_rtm_change(self.PREFIX, target_gw)
199
        msg.rtm_flags = RtConst.RTF_GATEWAY
200
        try:
201
            self.rtsock.write_message(msg)
202
        except ProcessLookupError as e:
203
            print("Succcessfully raised {}".format(e))
204

205
    @pytest.mark.require_user("root")
206
    def test_rtm_multipath4_add_zero_weight(self):
207
        """Tests RTM_ADD with dest transitioning to multipath"""
208

209
        desired_map = {}
210
        for gw in ["192.0.2.10", "192.0.2.11", "192.0.2.13"]:
211
            msg = self.rtsock.new_rtm_add(self.PREFIX, gw)
212
            msg.rtm_flags = RtConst.RTF_GATEWAY
213
            msg.rtm_rmx.rmx_weight = 0
214
            msg.rtm_inits |= RtConst.RTV_WEIGHT
215
            self.rtsock.write_message(msg)
216
            desired_map[gw] = self.rtsock.get_weight(0)
217

218
        routes = self.get_prefix_routes()
219
        derived_map = {r["gateway"]: r["weight"] for r in routes}
220
        assert derived_map == desired_map
221

222
    @pytest.mark.require_user("root")
223
    def test_rtm_multipath4_getroute(self):
224
        """Tests RTM_GET with exact prefix lookup on the multipath group"""
225
        gws = ["192.0.2.10", "192.0.2.11", "192.0.2.13"]
226
        return self._test_rtm_multipath_getroute(gws)
227

228
    @pytest.mark.require_user("root")
229
    def test_rtm_multipath6_getroute(self):
230
        """Tests RTM_GET with exact prefix lookup on the multipath group"""
231
        gws = ["2001:db8::10", "2001:db8::11", "2001:db8::13"]
232
        return self._test_rtm_multipath_getroute(gws)
233

234
    def _test_rtm_multipath_getroute(self, gws):
235
        valid_gws = []
236
        for gw in gws:
237
            msg = self.rtsock.new_rtm_add(self.PREFIX, gw)
238
            msg.rtm_flags = RtConst.RTF_GATEWAY
239
            self.rtsock.write_message(msg)
240

241
            desired_sa = {
242
                RtConst.RTA_DST: msg.get_sa(RtConst.RTA_DST),
243
                RtConst.RTA_NETMASK: msg.get_sa(RtConst.RTA_NETMASK),
244
            }
245
            valid_gws.append(msg.get_sa(RtConst.RTA_GATEWAY))
246

247
            msg_get = RtsockRtMessage(
248
                RtConst.RTM_GET,
249
                self.rtsock.get_seq(),
250
                msg.get_sa(RtConst.RTA_DST),
251
                msg.get_sa(RtConst.RTA_NETMASK),
252
            )
253
            self.rtsock.write_message(msg_get)
254

255
            data = self.rtsock.read_data(msg_get.rtm_seq)
256
            msg_in = RtsockRtMessage.from_bytes(data)
257
            msg_in.print_in_message()
258
            msg_in.verify(RtConst.RTM_GET, desired_sa)
259

260
            # Additionally, check that the gateway is among the valid
261
            # gateways
262
            gw_found = False
263
            gw_in = msg_in.get_sa(RtConst.RTA_GATEWAY)
264
            for valid_gw in valid_gws:
265
                try:
266
                    assert valid_gw == gw_in
267
                    gw_found = True
268
                    break
269
                except AssertionError:
270
                    pass
271
            assert gw_found is True
272

273
Product

Resources

Company
