Attachment 1450517 Details for Bug 1590359 – python_script_to_simulate_the_error

python_script_to_simulate_the_error

ext_gw_test.py (text/x-python), 10.61 KB, created by Roni on 2018-06-12 13:27:32 UTC

(hide)

Description:

Filename:

MIME Type:

Creator: Roni

Created: 2018-06-12 13:27:32 UTC

Size: 10.61 KB

patch

obsolete

>#! /usr/bin/python
># -*- coding: utf-8 -*-
>
>"""
>External Network Provider fixtures
>"""
>
>from neutronclient.v2_0 import client
>from keystoneauth1 import identity, session
>from neutronclient.common.exceptions import BadRequest, NeutronClientException
>#from art.rhevm_api.tests_lib.high_level.extrenal_providers import OvnProvider
>
>
>#from art.rhevm_api.tests_lib.high_level import (
>#    clusters as hl_clusters,
>#    extrenal_providers as hl_extrenal_providers,
>#    vms as hl_vms,
>#)
>
>HOST_URL = "network-ge-1.scl.lab.tlv.redhat.com"
>#NETWORK_ID = "35c394b8-8c18-4b67-a7cf-0b4a23ffcc1e"
>NETWORK_ID = "1a1093c9-e572-4d8a-b3bc-399bc42080d8"
>SUBNET_ID = "1892a82b-0acc-4e13-a5b9-07eeab136e12"
>
>NETWORK_NAME = "roni_test"
>SUBNET_NAME  = "roni_subnet_test"
>
>class OvnProvider(client.Client):
>    """
>    Manage OVN provider
>    """
>    def __init__(self, username, password, auth_url):
>        """
>        Initialized connection to the provider.
>
>        Args:
>            username (str): Provider username
>            password (str): Provider password
>            auth_url (str): Provider authentication URL
>        """
>        auth = identity.Password(
>            auth_url=auth_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 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 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 get_provider_connection():
>    """
>    Get connection object to OVN provider
>    """
>    host_name = HOST_URL
>    username = "admin@internal"
>    password = "123456"
>
>    keystone_url = "https://{}:35357/v2.0".format(host_name)
>
>    #provider_class = hl_extrenal_providers.OvnProvider(
>    provider_class = OvnProvider(
>        username=username,
>        password=password,
>        auth_url=keystone_url
>    )
>    return provider_class
>
>
># -------------------------------------------------------------------------
># Main
># -------------------------------------------------------------------------
>if __name__ == "__main__":
>
>    """
>    The script should be run 3 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
>        - check static routes: ovn-nbctl list Logical_Router_Static_Route
>        - there should only be one static route
>        
>    2. Run the script 2th time:
>       - 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. Modify as follow and run the script 3th time:
>        NETWORK_NAME = "roni_test_3"            #change 3 to 4 
>        SUBNET_NAME  = "roni_subnet_test_3"     #change 3 to 4     
>        ROUTER_NAME  = "ovn_router_test_3"      #change 3 to 4
>        #IP_ADDR = "1.1.1.144"                  #change 111 to 144
>        IP_ADDR      = "1.1.1.111"
>        
>         - 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. Remove router
>       - check static routes: ovn-nbctl list Logical_Router_Static_Route
>       - all static routes should be removed
>    """
>
>    provider_class = get_provider_connection()
>
>    NETWORK_NAME = "roni_test_3"
>    SUBNET_NAME  = "roni_subnet_test_3"
>    ROUTER_NAME  = "ovn_router_test_3"
>    #IP_ADDR = "1.1.1.144"
>    IP_ADDR      = "1.1.1.111"
>
>    # if Network already exists then enter "modify" mode, section #2 above
>    action = "modify" if provider_class.get_router_id(router=ROUTER_NAME) else "add"
>
>    if action == "modify":
>        IP_ADDR = "1.1.1.222"
>        network_id = provider_class.get_network_id(network=NETWORK_NAME)
>    else:
>        network_id = provider_class.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_class.get_subnet_id(subnet=SUBNET_NAME)
>    else:
>        subnet_id = provider_class.add_subnet(OVN_SUBNET, network=NETWORK_NAME)
>
>    # if router already exists then change IP
>    #ip_adder  = "1.1.1.111" if action == "add" else "1.1.1.222"
>
>    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": IP_ADDR
>                    }
>                ]
>            }
>    }
>
>    if action == "modify":
>        router_id = provider_class.update_network_properties(network=NETWORK_NAME, properties=EX)
>    else:
>        router_id = provider_class.add_router(properties=EX)
>
>    print router_id
>

Actions: View

Attachments on bug 1590359: 1450517