Attachment 1451460 Details for Bug 1580254 – static_route_test.py

static_route_test.py

static_route_test.py (text/x-python), 12.54 KB, created by Roni on 2018-06-14 15:21:43 UTC

(hide)

Description:

Filename:

MIME Type:

Creator: Roni

Created: 2018-06-14 15:21:43 UTC

Size: 12.54 KB

patch

obsolete

>#! /usr/bin/python
># -*- coding: utf-8 -*-
>
>from neutronclient.v2_0 import client
>from keystoneauth1 import identity, session
>from neutronclient.common.exceptions import BadRequest, NeutronClientException
>
>
>HOST_NAME = "network-ge-1.scl.lab.tlv.redhat.com"
>
>class OvnProvider(client.Client):
>    """
>    Manage OVN provider
>    """
>    def __init__(self, username, password, host_name):
>        """
>        Initialized connection to the provider.
>
>        Args:
>            username (str): Provider username
>            password (str): Provider password
>            host_name (str): Provider IP or Name
>        """
>        keystone_url = "https://{}:35357/v2.0".format(host_name)
>        auth = identity.Password(
>            auth_url=keystone_url, username=username, password=password
>        )
>        sess = session.Session(auth=auth, verify=False)
>        super(OvnProvider, self).__init__(session=sess)
>
>
>    def add_router(self, properties):
>        """
>        Add router
>
>        Args:
>            properties (dict): Router properties
>
>        Returns:
>            str: Router ID or empty string if error occurred
>        """
>        router_name = properties.get("name")
>        try:
>            ret = self.create_router({"router": properties})
>            print "create_router() return: {}".format(ret)
>        except NeutronClientException:
>            print NeutronClientException.message
>            return ""
>
>        id = self.get_router_id(router=router_name)
>        return id
>
>
>    def delete_router_by_name(self, router):
>        """
>        Delete router by name
>
>        Args:
>            router (str): Router name
>
>        Returns:
>            bool: True if deleted successfully, False otherwise
>        """
>        router_id = self.get_router_id(router=router)
>        if not router_id:
>            return False
>
>        try:
>            self.delete_router(router=router_id)
>        except NeutronClientException:
>            return False
>        return True
>
>
>    def get_network_id(self, network):
>        """
>        Get network ID
>
>        Args:
>            network (str): Network name
>
>        Returns:
>            str: Network ID
>        """
>        networks = self.get_all_networks()
>        return self.get_id_by_name(name=network, collection=networks)
>
>
>    def get_all_routers(self):
>        """
>        Get all routers
>
>        Returns:
>            list: List of all routers
>        """
>        return self.list_routers().get("routers")
>
>
>    def get_all_networks(self):
>        """
>        Get all networks
>
>        Returns:
>            list: List of all networks
>        """
>        return self.list_networks().get("networks")
>
>
>    def get_router_id(self, router):
>        """
>        Get network ID
>
>        Args:
>            router (str): Router name
>
>        Returns:
>            str: Router ID
>        """
>        routers = self.get_all_routers()
>        return self.get_id_by_name(name=router, collection=routers)
>
>
>    def delete_network_by_name(self, network):
>        """
>        Delete network by name
>
>        Args:
>            network (str): Network name
>
>        Returns:
>            bool: True if deleted successfully, False otherwise
>        """
>        network_id = self.get_network_id(network=network)
>        if not network_id:
>            return False
>
>        try:
>            self.delete_network(network=network_id)
>        except BadRequest:
>            return False
>        return True
>
>
>    def add_network(self, network):
>        """
>        Create network
>
>        Args:
>            network (dict): Network dict to create
>
>        Returns:
>            str: Network ID or empty string
>        """
>        network_name = network.get("name")
>        try:
>            self.create_network({"network": network})
>        except BadRequest:
>            return ""
>
>        return self.get_network_id(network=network_name)
>
>
>    def update_network_properties(self, network, properties):
>        """
>        Update network properties
>
>        Args:
>            network (str): Network name
>            properties (dict): Network properties to update
>
>        Returns:
>            bool: True if updated successfully, False otherwise
>        """
>        network_id = self.get_network_id(network=network)
>        if not network_id:
>            return False
>
>        try:
>            self.update_network(
>                network=network_id, body={"network": properties}
>            )
>        except BadRequest:
>            return False
>        return True
>
>
>    def update_router_properties(self, router, properties):
>        """
>        Update router properties
>
>        Args:
>            router (str): Router name
>            properties (dict): Router properties
>
>        Returns:
>            bool: True if updated successfully, False otherwise
>        """
>        router_id = self.get_router_id(router=router)
>        if not router_id:
>            return False
>
>        try:
>            self.update_router(router=router_id, body={"router": properties})
>        except NeutronClientException:
>            return False
>        return True
>
>
>    def add_subnet(self, subnet, network=None):
>        """
>        Create subnet
>
>        Args:
>            subnet (dict): Subnet dict to create
>            network (str): Network name for the subnet
>
>        Returns:
>            str: Subnet ID or empty string
>        """
>        subnet_name = subnet.get("name")
>        network_id = self.get_network_id(network=network)
>
>        if not network_id:
>            return ""
>
>        subnet["network_id"] = network_id
>        try:
>            self.create_subnet({"subnet": subnet})
>        except BadRequest:
>            return ""
>
>        return self.get_subnet_id(subnet=subnet_name)
>
>
>    def get_subnet_id(self, subnet):
>        """
>        Get subnet ID
>
>        Args:
>            subnet (str): Subnet name
>
>        Returns:
>            str: Subnet ID
>        """
>        subnets = self.get_all_subnets()
>        return self.get_id_by_name(name=subnet, collection=subnets)
>
>
>    def get_id_by_name(self, name, collection):
>        """
>        Get object ID by name from collection
>
>        Args:
>            name (str): Name to search for
>            collection (list): List of objects
>
>        Returns:
>            str: ID if name found, or empty string otherwise
>        """
>        id_ = [
>            col.get("id") for col in collection if col.get("name") == name
>        ]
>        return id_[0] if id_ else ""
>
>
>    def get_all_subnets(self):
>        """
>        Get all networks
>
>        Returns:
>            list: List of all networks
>        """
>        return self.list_subnets().get("subnets")
>
>
>def scenario(operation=None):
>    """
>    Run the scenario described at:
>    https://bugzilla.redhat.com/show_bug.cgi?id=1580254
>
>    To change/choose environment edit the HOST_NAME variable at the top
>    
>    The script should be executed 5 times as followed:
>    0. Manually clean OVN data
>        ovn-nbctl lr-list
>        ovn-nbctl lr-del <IDs from command above>
>       
>        ovn-nbctl ls-list
>        ovn-nbctl ls-del <IDs from command above>
>       
>        ovn-nbctl dhcp-options-list
>        ovn-nbctl dhcp-options-del <IDs from command above>
>
>    1. Run the script 1th time
>        - manually check static routes: ovn-nbctl list Logical_Router_Static_Route
>        - there should only be one static route
>            
>    2. Run the script 2th time:
>        This will change 1.1.1.111 to 1.1.1.222, run 'ovn-nbctl show' to verify 
>        - manually check static routes: ovn-nbctl list Logical_Router_Static_Route
>        - there should only be one static route, the old should be removed, a new one added
>    
>    3. Run the script with "new_network" as first parameter:
>        this will do the following: 
>        NETWORK_NAME = "test_network_1"   # will replace 1 to 2 
>        SUBNET_NAME  = "test_subnet_1"    # will replace 1 to 2
>        EXTERNAL_IP  = "1.1.1.111"        # will replace 111 to 133
>        
>        - manually check static routes: ovn-nbctl list Logical_Router_Static_Route
>        - there should only be one static route, the old should be removed, a new one added
>    
>    4. Run the script with "del_router" as first parameter:
>        This will delete "test_router_1"
>        
>        - manually check static routes: ovn-nbctl list Logical_Router_Static_Route
>        - all static routes should be removed
>
>    Arges:
>      operation (str): describe special function behavior
>                    new_network - create a new network if one already exists
>                    del_router - delete a router and exit
>    """
>    provider = OvnProvider(
>        username="admin@internal",
>        password="123456",
>        host_name=HOST_NAME
>    )
>
>    NETWORK_NAME = "test_network_1"
>    SUBNET_NAME = "test_subnet_1"
>    ROUTER_NAME = "test_router_1"
>    EXTERNAL_IP = "1.1.1.111"
>
>    # if Network already exists then enter "modify" mode, section #2 above
>    action = "modify" if provider.get_network_id(network=NETWORK_NAME) else "add"
>
>    # if operation include 'new_network', see section #3 above
>    new_network = True if operation and "new_network" in operation else False
>
>    # if operation include 'del_router', see section #4 above
>    del_router = True if operation and "del_router" in operation else False
>
>    if new_network:
>        action = "add"
>        NETWORK_NAME = NETWORK_NAME.replace("1", "2")
>        SUBNET_NAME  = SUBNET_NAME.replace("1", "2")
>        EXTERNAL_IP  = EXTERNAL_IP.replace("111", "133")
>
>    if del_router:
>        provider.delete_router_by_name(router=ROUTER_NAME)
>        return
>
>    if action == "modify":
>        EXTERNAL_IP = "1.1.1.222"
>        network_id = provider.get_network_id(network=NETWORK_NAME)
>    else:
>        network_id = provider.add_network(network={"name": NETWORK_NAME})
>
>    OVN_SUBNET = {
>        "name": SUBNET_NAME,
>        "cidr": "1.1.1.0/24",
>        "enable_dhcp": True,
>        "ip_version": 4,
>        "network_id": str(network_id),
>        "gateway_ip": "1.1.1.254"
>    }
>
>    if action == "modify":
>        subnet_id = provider.get_subnet_id(subnet=SUBNET_NAME)
>    else:
>        subnet_id = provider.add_subnet(OVN_SUBNET, network=NETWORK_NAME)
>
>
>    EX = {
>        "name": ROUTER_NAME,
>        "external_gateway_info":
>            {
>                "network_id": str(network_id),
>                "enable_snat": False,
>                "external_fixed_ips": [
>                    {
>                        "subnet_id": str(subnet_id),
>                        "ip_address": EXTERNAL_IP
>                    }
>                ]
>            }
>    }
>
>    if action == "modify" or new_network:
>        router_id = provider.update_router_properties(router=ROUTER_NAME, properties=EX)
>    else:
>        router_id = provider.add_router(properties=EX)
>
>    assert router_id
>
>
>if __name__ == "__main__":
>
>    raw_input ("\n"
>       "Manually clean OVN data\n"
>       "ovn-nbctl lr-list\n"
>       "ovn-nbctl lr-del <IDs from command above>\n"
>       "ovn-nbctl ls-list\n"
>       "ovn-nbctl ls-del <IDs from command above>\n"
>       "ovn-nbctl dhcp-options-list\n"
>       "ovn-nbctl dhcp-options-del <IDs from command above>\n"
>       "Press any key to run step #1...\n"
>    )
>    scenario()
>    print ("\n"
>           "Manually run: 'ovn-nbctl show'\n"
>           "External gateway: 1.1.1.111\n"
>           "Manually check static routes:\n"
>           "ovn-nbctl list Logical_Router_Static_Route\n"
>           "There should be only one static route\n"
>    )
>
>
>    raw_input("Press any key to run step #2...")
>    scenario()
>    print ("\n"
>           "Manually run: 'ovn-nbctl show'\n"
>           "External gateway: 1.1.1.111 should be changed to 1.1.1.222\n"
>           "Manually check static routes:\n"
>           "ovn-nbctl list Logical_Router_Static_Route\n"
>           "There should be only one static route, the old should be removed, a new one added\n"
>           "_uuid was changed\n"
>    )
>
>
>    raw_input("Press any key to run step #3...")
>    scenario("new_network")
>    print ("Manually run: 'ovn-nbctl show'\n"
>           "New network was created: 'test_network_2'\n" 
>           "New subnet was created: 'test_subnet_2'\n"     
>           "External IP changed from 1.1.1.222 to 1.1.1.133\n"
>           "manually check static routes:\n"
>           "ovn-nbctl list Logical_Router_Static_Route\n"
>           "There should only be one static route, the old should be removed, a new one added\n"
>           "_uuid was changed\n"
>    )
>
>
>    raw_input("Press any key to run step #4...")
>    scenario("del_router")
>    print("Manually run: 'ovn-nbctl show'\n"
>          "There should be no router only two networks 1 & 2\n"
>          "Manually check static routes:\n"
>          "ovn-nbctl list Logical_Router_Static_Route\n"
>          "All static routes should be removed\n"
>    )

Actions: View

Attachments on bug 1580254: 1451460