| """ |
| Network Data Model Test |
| |
| 1) Create a slice1 |
| 2) Create sliver1 on slice1 |
| 3) Verify one quantum network created for sliver1 |
| 4) Create a private network, network1 |
| 5) Connect network1 to slice1 |
| 6) Create sliver1_2 on slice1 |
| 7) Verify two quantum networks created for sliver1_2 |
| """ |
| |
| import os |
| import json |
| import sys |
| import time |
| |
| sys.path.append("/opt/planetstack") |
| |
| os.environ.setdefault("DJANGO_SETTINGS_MODULE", "planetstack.settings") |
| from openstack.manager import OpenStackManager |
| from core.models import Slice, Sliver, ServiceClass, Reservation, Tag, Network, User, Node, Image, Deployment, Site, NetworkTemplate, NetworkSlice |
| |
| from planetstacktest import PlanetStackTest, fail_unless, fail |
| |
| class NetworkTest(PlanetStackTest): |
| def __init__(self): |
| PlanetStackTest.__init__(self) |
| |
| def wait_for_ports(self, sliver, count=1, max_time=120): |
| print "waiting for %d ports on %s" % (count, str(sliver)) |
| while max_time>0: |
| ports = self.manager.driver.shell.quantum.list_ports(device_id=sliver.instance_id)["ports"] |
| if len(ports)==count: |
| return ports |
| |
| fail_unless(len(ports)<=count, "too many ports") |
| |
| time.sleep(10) |
| max_time = max_time - 10 |
| |
| fail("timed out while waiting for port creation") |
| |
| def ports_to_networks(self, ports): |
| networks = [] |
| for port in ports: |
| port_networks = networks + self.manager.driver.shell.quantum.list_networks(id=port["network_id"])["networks"] |
| for network in port_networks: |
| if not (network in networks): |
| networks.append(network) |
| return networks |
| |
| def ports_to_network_names(self, ports): |
| network_names = [] |
| for network in self.ports_to_networks(ports): |
| network_names.append(network["name"]) |
| return network_names |
| |
| def verify_network_names(self, ports, network_names): |
| port_network_names = sorted(self.ports_to_network_names(ports)) |
| network_names = sorted(network_names) |
| fail_unless(port_network_names == network_names, "mismatched network names: %s != %s" % (str(port_network_names), str(network_names))) |
| print " verified network ports to", ",".join(port_network_names) |
| |
| def test_slice1(self): |
| slice1Name = self.make_slice_name() |
| slice1 = Slice(name = slice1Name, |
| omf_friendly=True, |
| site=self.testSite, |
| creator=self.testUser) |
| slice1=self.save_and_wait_for_enacted(slice1, nonempty_fields=["tenant_id"]) |
| |
| sliver1 = Sliver(image = self.testImage, |
| creator=self.testUser, |
| slice=slice1, |
| node=self.testNode, |
| deploymentNetwork=self.testDeployment) |
| sliver1=self.save_and_wait_for_enacted(sliver1, nonempty_fields=["instance_id", "ip"]) |
| |
| # sliver1 should have only one port, its private network |
| ports = self.wait_for_ports(sliver1, count=1) |
| self.verify_network_names(ports, [slice1.name]) |
| |
| network1 = Network(name = slice1Name + "-pvt", |
| template = self.get_network_template("private"), |
| owner = slice1) |
| network1=self.save_and_wait_for_enacted(network1, nonempty_fields=["network_id", "subnet_id", "router_id", "subnet"]) |
| |
| network1_slice1 = NetworkSlice(network=network1, slice=slice1) |
| network1_slice1.save() # does not need to be enacted |
| |
| sliver1_2 = Sliver(image = self.testImage, |
| creator=self.testUser, |
| slice=slice1, |
| node=self.testNode, |
| deploymentNetwork=self.testDeployment) |
| sliver1_2=self.save_and_wait_for_enacted(sliver1_2, nonempty_fields=["instance_id", "ip"]) |
| |
| ports = self.wait_for_ports(sliver1_2, count=2) |
| self.verify_network_names(ports, [slice1.name, network1.name]) |
| |
| self.slice1 = slice1 |
| self.network1 = network1 |
| |
| def test_slice2(self): |
| slice2Name = self.make_slice_name() |
| slice2 = Slice(name = slice2Name, |
| omf_friendly=True, |
| site=self.testSite, |
| creator=self.testUser) |
| slice2=self.save_and_wait_for_enacted(slice2, nonempty_fields=["tenant_id"]) |
| |
| network2 = Network(name = slice2Name + "-pvt", |
| template = self.get_network_template("private"), |
| owner = slice2) |
| network2=self.save_and_wait_for_enacted(network2, nonempty_fields=["network_id", "subnet_id", "router_id", "subnet"]) |
| |
| network2_slice2 = NetworkSlice(network=network2, slice=slice2) |
| network2_slice2.save() # does not need to be enacted |
| |
| sliver2_1 = Sliver(image = self.testImage, |
| creator=self.testUser, |
| slice=slice2, |
| node=self.testNode, |
| deploymentNetwork=self.testDeployment) |
| sliver2_1=self.save_and_wait_for_enacted(sliver2_1, nonempty_fields=["instance_id", "ip"]) |
| |
| ports = self.wait_for_ports(sliver2_1, count=2) |
| self.verify_network_names(ports, [slice2.name, network2.name]) |
| |
| self.slice2 = slice2 |
| self.network2 = network2 |
| |
| def test_shared_private_net(self): |
| # connect network2 to slice1 |
| self.network2.permittedSlices.add(self.slice1) |
| network2_slice1 = NetworkSlice(network=self.network2, slice=self.slice1) |
| network2_slice1.save() |
| |
| sliver1_3 = Sliver(image = self.testImage, |
| creator=self.testUser, |
| slice=self.slice1, |
| node=self.testNode, |
| deploymentNetwork=self.testDeployment) |
| sliver1_3=self.save_and_wait_for_enacted(sliver1_3, nonempty_fields=["instance_id", "ip"]) |
| |
| ports = self.wait_for_ports(sliver1_3, count=3) |
| self.verify_network_names(ports, [self.slice1.name, self.network1.name, self.network2.name]) |
| |
| def test_nat_net(self): |
| slice3Name = self.make_slice_name() |
| slice3 = Slice(name = slice3Name, |
| omf_friendly=True, |
| site=self.testSite, |
| creator=self.testUser) |
| slice3=self.save_and_wait_for_enacted(slice3, nonempty_fields=["tenant_id"]) |
| |
| network3 = Network(name = slice3Name + "-nat", |
| template = self.get_network_template("private-nat"), |
| owner = slice3) |
| # note that router_id will not be filled in for nat-net, since nat-net has no routers |
| network3=self.save_and_wait_for_enacted(network3, nonempty_fields=["network_id", "subnet_id", "subnet"]) |
| |
| network3_slice3 = NetworkSlice(network=network3, slice=slice3) |
| network3_slice3.save() # does not need to be enacted |
| |
| sliver3_1 = Sliver(image = self.testImage, |
| creator=self.testUser, |
| slice=slice3, |
| node=self.testNode, |
| deploymentNetwork=self.testDeployment) |
| sliver3_1=self.save_and_wait_for_enacted(sliver3_1, nonempty_fields=["instance_id", "ip"]) |
| |
| ports = self.wait_for_ports(sliver3_1, count=2) |
| self.verify_network_names(ports, [slice3.name, "nat-net"]) |
| |
| def run(self): |
| self.setup() |
| try: |
| self.test_slice1() |
| self.test_slice2() |
| self.test_shared_private_net() |
| self.test_nat_net() |
| print "SUCCESS" |
| finally: |
| self.cleanup() |
| |
| def main(): |
| NetworkTest().run() |
| |
| if __name__=="__main__": |
| main() |
| |
| |
| |