/
test_fabric_forwarding.py
292 lines (275 loc) · 13.4 KB
/
test_fabric_forwarding.py
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
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
from common.k8s.base import BaseK8sTest
from tcutils.wrappers import preposttest_wrapper
from tcutils.util import get_random_name
import test
import time
class TestFabricFWD(BaseK8sTest):
@classmethod
def setUpClass(cls):
try:
super(TestFabricFWD, cls).setUpClass()
cls.setup_fabric_gw()
except:
cls.tearDownClass()
raise
@classmethod
def tearDownClass(cls):
cls.cleanup_fabric_gw()
super(TestFabricFWD, cls).tearDownClass()
def is_test_applicable(self):
'''verify the fabroic gateway info
'''
if not self.inputs.fabric_gw_info:
return (False , "Fabric gateway is needed for the test run")
return (True , None)
def setup_namespaces_pods_for_fabric_test(self, isolation=False,ip_fabric_forwarding=False):
""" common routine to create the namesapces and the pods by enabling the fabric forwarding
1.create 2 namespaces (ns1,ns2 enable fabric forwarding)
2.create pods in each namespace and verify(ns1:pod1,pod2(c1,c2), ns2:pod1,default:pod1)
"""
namespace1_name = get_random_name("ns1")
namespace2_name = get_random_name("ns2")
namespace1 = self.setup_namespace(name = namespace1_name, isolation = isolation,
ip_fabric_forwarding = ip_fabric_forwarding)
namespace2 = self.setup_namespace(name = namespace2_name, isolation = isolation,
ip_fabric_forwarding = ip_fabric_forwarding)
#verifying namespaces have been created
assert namespace1.verify_on_setup()
assert namespace2.verify_on_setup()
label = "fabric"
#create a pod in default namespaces
pod1_in_default_ns = self.setup_ubuntuapp_pod()
#create a two pods in fabric forwarding enabled namespace
#pod1 with one container and pod2 with two containers
pod1_in_ns1 = self.setup_ubuntuapp_pod(namespace=namespace1_name,
labels={'app': label})
spec = {
'containers': [
{'image': 'ubuntu-upstart',
"name": "c1",
'command': ['sleep', '1000000'],
'image_pull_policy': 'IfNotPresent'
},
{'image': 'ubuntu-upstart',
"name": "c2",
'command': ['sleep', '1000000'],
'image_pull_policy': 'IfNotPresent'
}
]
}
pod2_in_ns1 = self.setup_pod(namespace=namespace1_name,
spec=spec,
labels={'app': label})
#create pod 3 without associating any label
pod3_in_ns1 = self.setup_ubuntuapp_pod(namespace=namespace1_name)
#create a pod in fabric forwarding enabled namespace
pod1_in_ns2 = self.setup_ubuntuapp_pod(namespace=namespace2_name,
labels={'app': label})
assert pod1_in_default_ns.verify_on_setup()
assert pod1_in_ns1.verify_on_setup()
assert pod2_in_ns1.verify_on_setup()
assert pod3_in_ns1.verify_on_setup()
assert pod1_in_ns2.verify_on_setup()
client1 = [pod1_in_ns1, pod2_in_ns1, pod3_in_ns1, namespace1]
client2 = [pod1_in_ns2, namespace2]
client3 = [pod1_in_default_ns]
return (client1, client2, client3)
#end setup_namespaces_pods_for_fabric_test
@preposttest_wrapper
def test_pod_publicreachability_with_fabric_forwarding_enabled(self):
"""
Verify the pods reachability to public network when ip fabric forwarding is enables
1.create 2 isolated namespaces with ip fabd forwarding enabled
2.create pods in thos namespaces(ns1:pod1,pod2(c1,c2), ns2:pod1 ,default:pod1)
3.ping ns1:pod1 to public host dns should PASS
4.ping from ns1:pod2:c1 and ns1:pod2:c2 to the public dns should PASS
5.ping from ns1:pod1 to ns2:pod1 should FAIL
6.ping from ns1:pod1 to dafeult:pod1 should FAIL
"""
client1, client2, client3 = self.setup_namespaces_pods_for_fabric_test(isolation=True,
ip_fabric_forwarding=True)
assert client1[0].ping_to_ip(self.inputs.public_host)
assert client1[1].ping_to_ip(self.inputs.public_host,container="c1")
assert client1[1].ping_to_ip(self.inputs.public_host,container="c2")
assert client2[0].ping_to_ip(self.inputs.public_host)
assert client1[0].ping_to_ip(client1[2].pod_ip)
assert client1[0].ping_to_ip(client2[0].pod_ip, expectation=False)
assert client1[0].ping_to_ip(client3[0].pod_ip, expectation=False)
#end test_pod_publicreachability_with_fabric_forwarding_enabled
@preposttest_wrapper
def test_fabric_forwarding_disabled_by_default(self):
"""
IP Fabric Forwaring is disabled by default
1.create an isolated namespace with out ip fabric forwarding enabled and verify
2.create a pod in the namespace created in step 1 and verify
3.public reachability should fail
4.delete the pod and namespace created in step1 and step2
5.recreate the namespace with ip fabric enabled
6.create a pod in the namespace created in step 5
7.now verifu the public reachability ,should PASS
"""
namespace1_name = "ns1"
namespace1 = self.setup_namespace(name=namespace1_name, isolation=True)
assert namespace1.verify_on_setup()
pod1_in_ns1 = self.setup_ubuntuapp_pod(namespace=namespace1_name)
assert pod1_in_ns1.verify_on_setup()
assert pod1_in_ns1.ping_to_ip(self.inputs.public_host, expectation=False)
self.perform_cleanup(namespace1)
#recreate the same namespace with ip fabric enabled this time
#time.sleep(2)
namespace1 = self.setup_namespace(name=namespace1_name, isolation=True,
ip_fabric_forwarding=True)
pod1_in_ns1 = self.setup_ubuntuapp_pod(namespace=namespace1_name)
assert pod1_in_ns1.verify_on_setup()
assert pod1_in_ns1.ping_to_ip(self.inputs.public_host)
#end test_fabric_forwarding_disabled_by_default
@preposttest_wrapper
def test_ping_with_jumbo_frame(self):
"""
IP Fabric Forwaring should support jumbo frames
1.create 2 isolated namespaces with ip fabd forwarding enabled
2.create pods in thos namespaces(ns1:pod1,pod2(c1,c2), ns2:pod1 ,default:pod1)
3.ping ns1:pod1 to public host dns should PASS with jumbo frame
4.ping from ns1:pod2:c1 and ns1:pod2:c2 to the public dns should PASS with jumbo frame
5.ping from ns1:pod1 to ns2:pod1 should FAIL with jumbo frame
6.ping from ns1:pod1 to dafeult:pod1 should FAIL with jumbo frame
"""
jumbo_frame_size="4000"
client1, client2, client3 = self.setup_namespaces_pods_for_fabric_test(isolation=True,
ip_fabric_forwarding=True)
assert client1[0].ping_to_ip(self.inputs.cfgm_ip, jumboframe=jumbo_frame_size)
assert client1[0].ping_to_ip(client2[0].pod_ip, jumboframe=jumbo_frame_size,
expectation=False)
assert client1[1].ping_to_ip(self.inputs.cfgm_ip,jumboframe=jumbo_frame_size,container="c2")
assert client1[0].ping_to_ip(client3[0].pod_ip, jumboframe=jumbo_frame_size,
expectation=False)
#end test_ping_with_jumbo_frame
def test_delete_and_recreate_the_namespace_with_fabric_fwd_enabled(self):
"""
Verify deletion and recreation of namespace works fien with ip fabric enable
1.create a namespace with ip fabric ennabled
2.create a pod in the namespace created in step1
3.verify the pubilc reahcability from the pod
4.delete the namespace and pod
5.recreate the same namespce names with ip fabric enabled
6.create the pod
7.verify the pubilc reahcbility
"""
namespace1_name = "test"
namespace1 = self.setup_namespace(name=namespace1_name, isolation=True,
ip_fabric_forwarding=True)
assert namespace1.verify_on_setup()
pod1 = self.setup_busybox_pod(namespace=namespace1_name)
assert pod1.verify_on_setup()
assert pod1.ping_to_ip(self.inputs.public_host)
self.perform_cleanup(namespace1)
namespace1 = self.setup_namespace(name=namespace1_name, isolation=True,
ip_fabric_forwarding=True)
assert namespace1.verify_on_setup()
pod1 = self.setup_busybox_pod(namespace=namespace1_name)
assert pod1.verify_on_setup()
assert pod1.ping_to_ip(self.inputs.public_host)
@test.attr(type=['k8s_sanity'])
@preposttest_wrapper
def test_deployment_with_replica_update_for_fabric_fwd(self):
'''
Verifies fabric forwarding is enabled though deployment object
1.Create a deployment with n replicas with ip fabric forwardingenabled
2.verify the replicas able to reach the public network
3.update the pod replicas
4.should be able to reach pubic network from each pod
'''
labels = {'app': 'test'}
replicas = len(self.inputs.compute_ips)*2
new_replicas = len(self.inputs.compute_ips)*3
namespace1_name = get_random_name("ns1")
namespace1 = self.setup_namespace(name=namespace1_name, isolation=True,
ip_fabric_forwarding=True)
assert namespace1.verify_on_setup()
metadata = {}
spec = {}
name = get_random_name('ubuntu-dep')
template_metadata = {}
template_metadata['labels'] = labels
template_spec = {
'containers': [
{'image': 'ubuntu-upstart',
"name": "c1",
'command': ['sleep', '1000000'],
'image_pull_policy': 'IfNotPresent',
}
]
}
spec.update({
'replicas': replicas,
'template': {
'metadata': template_metadata,
'spec': template_spec
}
})
dep_1 = self.setup_deployment(name=name, namespace=namespace1_name,
metadata=metadata, spec=spec)
assert dep_1.verify_on_setup()
s_pod_fixtures = []
server_pods = dep_1.get_pods_list()
for x in server_pods:
s_pod_fixture = self.setup_ubuntuapp_pod(name=x.metadata.name,
namespace=namespace1_name)
s_pod_fixture.verify_on_setup()
assert s_pod_fixture.ping_to_ip(self.inputs.public_host)
dep_1.set_replicas(new_replicas)
assert dep_1.verify_on_setup()
s_pod_fixtures = []
server_pods = dep_1.get_pods_list()
for x in server_pods:
s_pod_fixture = self.setup_ubuntuapp_pod(name=x.metadata.name,
namespace=namespace1_name)
assert s_pod_fixture.verify_on_setup()
assert s_pod_fixture.ping_to_ip(self.inputs.public_host)
#end test_deployment_with_replica_update_for_fabric_fwd
@preposttest_wrapper
def test_deployment_with_fabric_fwd(self):
'''
Verifies fabric forwarding is enabled though deployment object
1.Create a deployment with n replicas with ip fabric forwardingenabled
2.verify the replicas able to reach the public network
'''
labels = {'app': 'test'}
replicas = len(self.inputs.compute_ips)*2
namespace1_name = get_random_name("ns1")
namespace1 = self.setup_namespace(name=namespace1_name, isolation=True,
ip_fabric_forwarding=True)
assert namespace1.verify_on_setup()
metadata = {}
spec = {}
name = get_random_name('ubuntu-dep')
template_metadata = {}
template_metadata['labels'] =labels
template_spec = {
'containers': [
{'image': 'ubuntu-upstart',
"name": "c1",
'command': ['sleep', '1000000'],
'image_pull_policy': 'IfNotPresent',
}
]
}
#spec.update({'replicas': replicas})
spec.update({
'replicas': replicas,
'template': {
'metadata': template_metadata,
'spec': template_spec
}
})
dep_1 = self.setup_deployment(name=name, namespace=namespace1_name,
metadata=metadata, spec=spec)
assert dep_1.verify_on_setup()
s_pod_fixtures = []
server_pods = dep_1.get_pods_list()
for x in server_pods:
s_pod_fixture = self.setup_ubuntuapp_pod(name=x.metadata.name,
namespace=namespace1_name)
s_pod_fixture.verify_on_setup()
assert s_pod_fixture.ping_to_ip(self.inputs.public_host)
#end test_deployment_with_fabric_fwd