]>
Commit | Line | Data |
---|---|---|
cb747ed9 | 1 | #!/usr/bin/env python |
acddc0ed | 2 | # SPDX-License-Identifier: ISC |
cb747ed9 DA |
3 | |
4 | # Copyright (c) 2021 by | |
5 | # Donatas Abraitis <donatas.abraitis@gmail.com> | |
6 | # | |
cb747ed9 DA |
7 | |
8 | """ | |
9 | Test if AddPath RX direction is not negotiated via AddPath capability. | |
10 | """ | |
11 | ||
12 | import os | |
13 | import sys | |
14 | import json | |
cb747ed9 DA |
15 | import pytest |
16 | import functools | |
17 | ||
18 | CWD = os.path.dirname(os.path.realpath(__file__)) | |
19 | sys.path.append(os.path.join(CWD, "../")) | |
20 | ||
21 | # pylint: disable=C0413 | |
22 | from lib import topotest | |
23 | from lib.topogen import Topogen, TopoRouter, get_topogen | |
cb747ed9 DA |
24 | from lib.common_config import step |
25 | ||
26 | pytestmark = [pytest.mark.bgpd] | |
27 | ||
28 | ||
e82b531d CH |
29 | def build_topo(tgen): |
30 | for routern in range(1, 5): | |
31 | tgen.add_router("r{}".format(routern)) | |
cb747ed9 | 32 | |
e82b531d CH |
33 | switch = tgen.add_switch("s1") |
34 | switch.add_link(tgen.gears["r1"]) | |
35 | switch.add_link(tgen.gears["r2"]) | |
cb747ed9 | 36 | |
e82b531d CH |
37 | switch = tgen.add_switch("s2") |
38 | switch.add_link(tgen.gears["r2"]) | |
39 | switch.add_link(tgen.gears["r3"]) | |
40 | switch.add_link(tgen.gears["r4"]) | |
cb747ed9 DA |
41 | |
42 | ||
43 | def setup_module(mod): | |
e82b531d | 44 | tgen = Topogen(build_topo, mod.__name__) |
cb747ed9 DA |
45 | tgen.start_topology() |
46 | ||
47 | router_list = tgen.routers() | |
48 | ||
49 | for i, (rname, router) in enumerate(router_list.items(), 1): | |
50 | router.load_config( | |
51 | TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) | |
52 | ) | |
53 | router.load_config( | |
54 | TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname)) | |
55 | ) | |
56 | ||
57 | tgen.start_router() | |
58 | ||
59 | ||
60 | def teardown_module(mod): | |
61 | tgen = get_topogen() | |
62 | tgen.stop_topology() | |
63 | ||
64 | ||
65 | def test_bgp_disable_addpath_rx(): | |
66 | tgen = get_topogen() | |
67 | ||
68 | if tgen.routers_have_failure(): | |
69 | pytest.skip(tgen.errors) | |
70 | ||
71 | r1 = tgen.gears["r1"] | |
72 | r2 = tgen.gears["r2"] | |
73 | ||
74 | step( | |
75 | "Check if r2 advertised only 2 paths to r1 (despite addpath-tx-all-paths enabled on r2)." | |
76 | ) | |
77 | ||
78 | def check_bgp_advertised_routes(router): | |
79 | output = json.loads( | |
80 | router.vtysh_cmd( | |
81 | "show bgp ipv4 unicast neighbor 192.168.1.1 advertised-routes json" | |
82 | ) | |
83 | ) | |
84 | expected = { | |
85 | "advertisedRoutes": { | |
86 | "172.16.16.254/32": { | |
87 | "addrPrefix": "172.16.16.254", | |
88 | "prefixLen": 32, | |
89 | }, | |
90 | "192.168.2.0/24": { | |
91 | "addrPrefix": "192.168.2.0", | |
92 | "prefixLen": 24, | |
93 | }, | |
94 | }, | |
95 | "totalPrefixCounter": 2, | |
96 | } | |
97 | ||
98 | return topotest.json_cmp(output, expected) | |
99 | ||
100 | test_func = functools.partial(check_bgp_advertised_routes, r2) | |
101 | success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5) | |
102 | assert result is None, "AddPath TX not working." | |
103 | ||
104 | step("Check if AddPath RX is disabled on r1 and we receive only 2 paths.") | |
105 | ||
106 | def check_bgp_disabled_addpath_rx(router): | |
107 | output = json.loads(router.vtysh_cmd("show bgp neighbor 192.168.1.2 json")) | |
108 | expected = { | |
109 | "192.168.1.2": { | |
110 | "bgpState": "Established", | |
111 | "neighborCapabilities": { | |
112 | "addPath": { | |
113 | "ipv4Unicast": {"txReceived": True, "rxReceived": True} | |
114 | }, | |
115 | }, | |
116 | "addressFamilyInfo": {"ipv4Unicast": {"acceptedPrefixCounter": 2}}, | |
117 | } | |
118 | } | |
119 | ||
120 | return topotest.json_cmp(output, expected) | |
121 | ||
122 | test_func = functools.partial(check_bgp_disabled_addpath_rx, r1) | |
123 | success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5) | |
124 | assert result is None, "AddPath RX advertised, but should not." | |
125 | ||
126 | ||
127 | if __name__ == "__main__": | |
128 | args = ["-s"] + sys.argv[1:] | |
129 | sys.exit(pytest.main(args)) |