Repository navigation
Expand file tree
/
Copy pathpolicyScale.py
More file actions
executable file
·120 lines (90 loc) · 3.1 KB
/
Copy pathpolicyScale.py
File metadata and controls
executable file
·120 lines (90 loc) · 3.1 KB
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
#!/usr/bin/python
# sanity tests
import api.tbed
import api.tutils
import time
import sys
import api.objmodel
# create lot of policies
def createPolicies(numPolicy, numRulesPerPolicy):
tenant = api.objmodel.tenant('default')
for pid in range(numPolicy):
pname = 'policy' + str(pid + 1)
# Create policy
policy = tenant.newPolicy(pname)
# create default deny Rule
policy.addRule('1', direction="in", protocol="tcp", action="deny")
# Create Rules
for rid in range(numRulesPerPolicy):
# Create allow port xxx Rule
policy.addRule(str(2 + rid), direction="in", priority=10, protocol="tcp", port=(8000 + rid), action="accept")
# Add the policy to epg
epgName = "srv" + str(pid) + ".private"
group = tenant.newGroup(epgName, policies=[pname])
# Test connections
def testConnections(testbed, numContainer):
# start containers
containers = testbed.runContainers(numContainer)
# start netcast listeners
testbed.startListeners(containers, [8000, 7999])
# Check connection to all containers
if testbed.checkConnections(containers, 8000, True) != True:
api.tutils.exit("Connection failed")
if testbed.checkConnections(containers, 7999, False) != False:
api.tutils.exit("Connection succeded while expecting it to fail")
# stop netcast listeners
testbed.stopListeners(containers)
# remove containers
testbed.removeContainers(containers)
# Cleanup all policies
def cleanupPolicies(numPolicy, numRulesPerPolicy):
tenant = api.objmodel.tenant('default')
for pid in range(numPolicy):
pname = 'policy' + str(pid + 1)
policy = tenant.newPolicy(pname)
# Remove policy from epg and delete epg
epgName = "srv" + str(pid) + ".private"
group = tenant.newGroup(epgName, policies=[])
tenant.deleteGroup(epgName)
# Remove the policy and rules
tenant.deletePolicy(pname)
# Test policy scale
def testPolicyScale(testbed, numCntr, numIter, numPolicies, numRulesPerPolicy):
for iter in range(numIter):
# Create policies
createPolicies(numPolicies, numRulesPerPolicy)
# Verify policy is working
testConnections(testbed, numCntr)
# Cleanup policies
cleanupPolicies(numPolicies, numRulesPerPolicy)
api.tutils.info("testPolicyScale Iteration " + str(iter) + " Passed")
if __name__ == "__main__":
# Parse command line args
if len(sys.argv) <= 3:
print "Usage: " + sys.argv[0] + " <create|delete|test> <numPolicies> <numRulesPerPolicy> [<ip-addr> <ip-addr> ...]"
sys.exit(1)
command = sys.argv[1]
numPolicies = int(sys.argv[2])
numRulesPerPolicy = int(sys.argv[3])
# Create policies
if command == "create":
createPolicies(numPolicies, numRulesPerPolicy)
elif command == "delete":
cleanupPolicies(numPolicies, numRulesPerPolicy)
elif command == "test":
# Form address list
addrList = []
for addr in sys.argv[4:]:
addrList.append(addr)
# Setup testbed
testbed = api.tbed.Testbed(addrList)
time.sleep(5)
numCntr = testbed.numNodes() * 2
numIter = 4
# Perform the test
testPolicyScale(testbed, numCntr, numIter, numPolicies, numRulesPerPolicy)
# Cleanup testbed
testbed.cleanup()
api.tutils.info("Policy Scale test PASSES")
else:
print "Unknown command " + command