1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from openstack.network.v2 import bgp_peer as _bgp_peer
from openstack.network.v2 import bgp_speaker as _bgp_speaker
from openstack.tests.functional import base
class TestBGPSpeaker(base.BaseFunctionalTest):
def setUp(self):
super().setUp()
self.LOCAL_AS = 101
self.IP_VERSION = 4
self.REMOTE_AS = 42
self.PEER_IP = '172.200.12.3'
self.SPEAKER_NAME = 'my_speaker' + self.getUniqueString()
self.PEER_NAME = 'my_peer' + self.getUniqueString()
if not self.user_cloud.network.find_extension("bgp"):
self.skipTest("Neutron BGP Dynamic Routing Extension disabled")
bgp_speaker = self.operator_cloud.network.create_bgp_speaker(
ip_version=self.IP_VERSION,
local_as=self.LOCAL_AS,
name=self.SPEAKER_NAME,
)
assert isinstance(bgp_speaker, _bgp_speaker.BgpSpeaker)
self.SPEAKER = bgp_speaker
bgp_peer = self.operator_cloud.network.create_bgp_peer(
name=self.PEER_NAME,
auth_type='none',
remote_as=self.REMOTE_AS,
peer_ip=self.PEER_IP,
)
assert isinstance(bgp_peer, _bgp_peer.BgpPeer)
self.PEER = bgp_peer
def tearDown(self):
sot = self.operator_cloud.network.delete_bgp_peer(self.PEER.id)
self.assertIsNone(sot)
sot = self.operator_cloud.network.delete_bgp_speaker(self.SPEAKER.id)
self.assertIsNone(sot)
super().tearDown()
def test_find_bgp_speaker(self):
sot = self.operator_cloud.network.find_bgp_speaker(self.SPEAKER.name)
self.assertEqual(self.IP_VERSION, sot.ip_version)
self.assertEqual(self.LOCAL_AS, sot.local_as)
# Check defaults
self.assertTrue(sot.advertise_floating_ip_host_routes)
self.assertTrue(sot.advertise_tenant_networks)
def test_get_bgp_speaker(self):
sot = self.operator_cloud.network.get_bgp_speaker(self.SPEAKER.id)
self.assertEqual(self.IP_VERSION, sot.ip_version)
self.assertEqual(self.LOCAL_AS, sot.local_as)
def test_list_bgp_speakers(self):
speaker_ids = [
sp.id for sp in self.operator_cloud.network.bgp_speakers()
]
self.assertIn(self.SPEAKER.id, speaker_ids)
def test_update_bgp_speaker(self):
sot = self.operator_cloud.network.update_bgp_speaker(
self.SPEAKER.id, advertise_floating_ip_host_routes=False
)
self.assertFalse(sot.advertise_floating_ip_host_routes)
def test_find_bgp_peer(self):
sot = self.operator_cloud.network.find_bgp_peer(self.PEER.name)
self.assertEqual(self.PEER_IP, sot.peer_ip)
self.assertEqual(self.REMOTE_AS, sot.remote_as)
def test_get_bgp_peer(self):
sot = self.operator_cloud.network.get_bgp_peer(self.PEER.id)
self.assertEqual(self.PEER_IP, sot.peer_ip)
self.assertEqual(self.REMOTE_AS, sot.remote_as)
def test_list_bgp_peers(self):
peer_ids = [pe.id for pe in self.operator_cloud.network.bgp_peers()]
self.assertIn(self.PEER.id, peer_ids)
def test_update_bgp_peer(self):
name = 'new_peer_name' + self.getUniqueString()
sot = self.operator_cloud.network.update_bgp_peer(
self.PEER.id, name=name
)
self.assertEqual(name, sot.name)
def test_add_remove_peer_to_speaker(self):
self.operator_cloud.network.add_bgp_peer_to_speaker(
self.SPEAKER.id, self.PEER.id
)
sot = self.operator_cloud.network.get_bgp_speaker(self.SPEAKER.id)
self.assertEqual([self.PEER.id], sot.peers)
# Remove the peer
self.operator_cloud.network.remove_bgp_peer_from_speaker(
self.SPEAKER.id, self.PEER.id
)
sot = self.operator_cloud.network.get_bgp_speaker(self.SPEAKER.id)
self.assertEqual([], sot.peers)
def test_add_remove_gw_network_to_speaker(self):
net_name = 'my_network' + self.getUniqueString()
net = self.user_cloud.create_network(name=net_name)
self.operator_cloud.network.add_gateway_network_to_speaker(
self.SPEAKER.id, net.id
)
sot = self.operator_cloud.network.get_bgp_speaker(self.SPEAKER.id)
self.assertEqual([net.id], sot.networks)
# Remove the network
self.operator_cloud.network.remove_gateway_network_from_speaker(
self.SPEAKER.id, net.id
)
sot = self.operator_cloud.network.get_bgp_speaker(self.SPEAKER.id)
self.assertEqual([], sot.networks)
def test_get_advertised_routes_of_speaker(self):
sot = self.operator_cloud.network.get_advertised_routes_of_speaker(
self.SPEAKER.id
)
self.assertEqual({'advertised_routes': []}, sot)
|
