blob: 131db5ab5781692570094ff8aa0f10811e056fc8 [file] [log] [blame]
Scott Bakerab23fa92013-08-25 16:53:49 -07001"""
2 Network Data Model Test
3
4 1) Create a slice1
5 2) Create sliver1 on slice1
6 3) Verify one quantum network created for sliver1
7 4) Create a private network, network1
8 5) Connect network1 to slice1
9 6) Create sliver1_2 on slice1
10 7) Verify two quantum networks created for sliver1_2
11"""
12
13import os
14import json
15import sys
16import time
17
Scott Baker166f4b82015-02-09 11:18:46 -080018sys.path.append("/opt/xos")
Scott Bakerab23fa92013-08-25 16:53:49 -070019
20os.environ.setdefault("DJANGO_SETTINGS_MODULE", "planetstack.settings")
21from openstack.manager import OpenStackManager
22from core.models import Slice, Sliver, ServiceClass, Reservation, Tag, Network, User, Node, Image, Deployment, Site, NetworkTemplate, NetworkSlice
23
24from planetstacktest import PlanetStackTest, fail_unless, fail
25
26class NetworkTest(PlanetStackTest):
27 def __init__(self):
28 PlanetStackTest.__init__(self)
29
30 def wait_for_ports(self, sliver, count=1, max_time=120):
31 print "waiting for %d ports on %s" % (count, str(sliver))
32 while max_time>0:
33 ports = self.manager.driver.shell.quantum.list_ports(device_id=sliver.instance_id)["ports"]
34 if len(ports)==count:
35 return ports
36
37 fail_unless(len(ports)<=count, "too many ports")
38
39 time.sleep(10)
40 max_time = max_time - 10
41
42 fail("timed out while waiting for port creation")
43
44 def ports_to_networks(self, ports):
45 networks = []
46 for port in ports:
47 port_networks = networks + self.manager.driver.shell.quantum.list_networks(id=port["network_id"])["networks"]
48 for network in port_networks:
49 if not (network in networks):
50 networks.append(network)
51 return networks
52
53 def ports_to_network_names(self, ports):
54 network_names = []
55 for network in self.ports_to_networks(ports):
56 network_names.append(network["name"])
57 return network_names
58
59 def verify_network_names(self, ports, network_names):
60 port_network_names = sorted(self.ports_to_network_names(ports))
61 network_names = sorted(network_names)
62 fail_unless(port_network_names == network_names, "mismatched network names: %s != %s" % (str(port_network_names), str(network_names)))
63 print " verified network ports to", ",".join(port_network_names)
64
65 def test_slice1(self):
66 slice1Name = self.make_slice_name()
67 slice1 = Slice(name = slice1Name,
68 omf_friendly=True,
69 site=self.testSite,
70 creator=self.testUser)
71 slice1=self.save_and_wait_for_enacted(slice1, nonempty_fields=["tenant_id"])
72
73 sliver1 = Sliver(image = self.testImage,
74 creator=self.testUser,
75 slice=slice1,
76 node=self.testNode,
77 deploymentNetwork=self.testDeployment)
78 sliver1=self.save_and_wait_for_enacted(sliver1, nonempty_fields=["instance_id", "ip"])
79
80 # sliver1 should have only one port, its private network
81 ports = self.wait_for_ports(sliver1, count=1)
82 self.verify_network_names(ports, [slice1.name])
83
84 network1 = Network(name = slice1Name + "-pvt",
85 template = self.get_network_template("private"),
86 owner = slice1)
87 network1=self.save_and_wait_for_enacted(network1, nonempty_fields=["network_id", "subnet_id", "router_id", "subnet"])
88
89 network1_slice1 = NetworkSlice(network=network1, slice=slice1)
90 network1_slice1.save() # does not need to be enacted
91
92 sliver1_2 = Sliver(image = self.testImage,
93 creator=self.testUser,
94 slice=slice1,
95 node=self.testNode,
96 deploymentNetwork=self.testDeployment)
97 sliver1_2=self.save_and_wait_for_enacted(sliver1_2, nonempty_fields=["instance_id", "ip"])
98
99 ports = self.wait_for_ports(sliver1_2, count=2)
100 self.verify_network_names(ports, [slice1.name, network1.name])
101
102 self.slice1 = slice1
103 self.network1 = network1
104
105 def test_slice2(self):
106 slice2Name = self.make_slice_name()
107 slice2 = Slice(name = slice2Name,
108 omf_friendly=True,
109 site=self.testSite,
110 creator=self.testUser)
111 slice2=self.save_and_wait_for_enacted(slice2, nonempty_fields=["tenant_id"])
112
113 network2 = Network(name = slice2Name + "-pvt",
114 template = self.get_network_template("private"),
115 owner = slice2)
116 network2=self.save_and_wait_for_enacted(network2, nonempty_fields=["network_id", "subnet_id", "router_id", "subnet"])
117
118 network2_slice2 = NetworkSlice(network=network2, slice=slice2)
119 network2_slice2.save() # does not need to be enacted
120
121 sliver2_1 = Sliver(image = self.testImage,
122 creator=self.testUser,
123 slice=slice2,
124 node=self.testNode,
125 deploymentNetwork=self.testDeployment)
126 sliver2_1=self.save_and_wait_for_enacted(sliver2_1, nonempty_fields=["instance_id", "ip"])
127
128 ports = self.wait_for_ports(sliver2_1, count=2)
129 self.verify_network_names(ports, [slice2.name, network2.name])
130
131 self.slice2 = slice2
132 self.network2 = network2
133
134 def test_shared_private_net(self):
135 # connect network2 to slice1
136 self.network2.permittedSlices.add(self.slice1)
137 network2_slice1 = NetworkSlice(network=self.network2, slice=self.slice1)
138 network2_slice1.save()
139
140 sliver1_3 = Sliver(image = self.testImage,
141 creator=self.testUser,
142 slice=self.slice1,
143 node=self.testNode,
144 deploymentNetwork=self.testDeployment)
145 sliver1_3=self.save_and_wait_for_enacted(sliver1_3, nonempty_fields=["instance_id", "ip"])
146
147 ports = self.wait_for_ports(sliver1_3, count=3)
148 self.verify_network_names(ports, [self.slice1.name, self.network1.name, self.network2.name])
149
150 def test_nat_net(self):
151 slice3Name = self.make_slice_name()
152 slice3 = Slice(name = slice3Name,
153 omf_friendly=True,
154 site=self.testSite,
155 creator=self.testUser)
156 slice3=self.save_and_wait_for_enacted(slice3, nonempty_fields=["tenant_id"])
157
158 network3 = Network(name = slice3Name + "-nat",
159 template = self.get_network_template("private-nat"),
160 owner = slice3)
161 # note that router_id will not be filled in for nat-net, since nat-net has no routers
162 network3=self.save_and_wait_for_enacted(network3, nonempty_fields=["network_id", "subnet_id", "subnet"])
163
164 network3_slice3 = NetworkSlice(network=network3, slice=slice3)
165 network3_slice3.save() # does not need to be enacted
166
167 sliver3_1 = Sliver(image = self.testImage,
168 creator=self.testUser,
169 slice=slice3,
170 node=self.testNode,
171 deploymentNetwork=self.testDeployment)
172 sliver3_1=self.save_and_wait_for_enacted(sliver3_1, nonempty_fields=["instance_id", "ip"])
173
174 ports = self.wait_for_ports(sliver3_1, count=2)
175 self.verify_network_names(ports, [slice3.name, "nat-net"])
176
177 def run(self):
178 self.setup()
179 try:
180 self.test_slice1()
181 self.test_slice2()
182 self.test_shared_private_net()
183 self.test_nat_net()
184 print "SUCCESS"
185 finally:
186 self.cleanup()
187
188def main():
189 NetworkTest().run()
190
191if __name__=="__main__":
192 main()
193
194
195