<s> from dragonflow . tests . common import utils <EOL> from dragonflow . tests . fullstack import test_base <EOL> from dragonflow . tests . fullstack import test_objects as objects <EOL> from neutron . agent . linux . utils import wait_until_true <EOL> import time <EOL> class TestOVSFlowsForDHCP ( test_base . DFTestBase ) : <EOL> def setUp ( self ) : <EOL> super ( TestOVSFlowsForDHCP , self ) . setUp ( ) <EOL> def check_dhcp_rule ( self , flows , dhcp_srv ) : <EOL> for flow in flows : <EOL> if flow [ '<STR_LIT>' ] == '<STR_LIT>' and flow [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> if ( '<STR_LIT>' + dhcp_srv + '<STR_LIT>' <EOL> in flow [ '<STR_LIT>' ] ) : <EOL> return True <EOL> return False <EOL> def get_dhcp_ip ( self , network_id , subnet_id ) : <EOL> ports = self . neutron . list_ports ( network_id = network_id ) <EOL> ports = ports [ '<STR_LIT>' ] <EOL> for port in ports : <EOL> if port [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> ips = port [ '<STR_LIT>' ] <EOL> for ip in ips : <EOL> if ip [ '<STR_LIT>' ] == subnet_id : <EOL> return ip [ '<STR_LIT>' ] <EOL> return None <EOL> def test_broadcast_dhcp_rule ( self ) : <EOL> found_dhcp_cast_flow = False <EOL> ovs = utils . OvsFlowsParser ( ) <EOL> flows = ovs . dump ( ) <EOL> for flow in flows : <EOL> if flow [ '<STR_LIT>' ] == '<STR_LIT>' and flow [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> if ( '<STR_LIT>' <EOL> in flow [ '<STR_LIT>' ] ) : <EOL> found_dhcp_cast_flow = True <EOL> break <EOL> self . assertTrue ( found_dhcp_cast_flow ) <EOL> def test_create_update_subnet_with_dhcp ( self ) : <EOL> ovs = utils . OvsFlowsParser ( ) <EOL> flows_before_change = ovs . dump ( ) <EOL> network = self . store ( objects . NetworkTestObj ( self . neutron , self . nb_api ) ) <EOL> network_id = network . create ( ) <EOL> subnet = { '<STR_LIT>' : network_id , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:4> , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : True } <EOL> subnet = self . neutron . create_subnet ( { '<STR_LIT>' : subnet } ) <EOL> subnet_id = subnet [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] <EOL> dhcp_ip = utils . wait_until_is_and_return ( <EOL> lambda : self . get_dhcp_ip ( network_id , subnet_id ) , <EOL> exception = Exception ( '<STR_LIT>' ) <EOL> ) <EOL> self . assertFalse ( self . check_dhcp_rule ( flows_before_change , dhcp_ip ) ) <EOL> wait_until_true ( <EOL> lambda : self . check_dhcp_rule ( ovs . dump ( ) , dhcp_ip ) , <EOL> exception = Exception ( '<STR_LIT>' ) , <EOL> timeout = <NUM_LIT:5> <EOL> ) <EOL> updated_subnet = { '<STR_LIT>' : False } <EOL> self . neutron . update_subnet ( subnet_id , { '<STR_LIT>' : updated_subnet } ) <EOL> time . sleep ( utils . DEFAULT_CMD_TIMEOUT ) <EOL> flows_after_update = ovs . dump ( ) <EOL> self . assertFalse ( self . check_dhcp_rule ( flows_after_update , dhcp_ip ) ) <EOL> network . close ( ) <EOL> def test_create_update_subnet_without_dhcp ( self ) : <EOL> ovs = utils . OvsFlowsParser ( ) <EOL> flows_before_change = ovs . dump ( ) <EOL> network = self . store ( objects . NetworkTestObj ( self . neutron , self . nb_api ) ) <EOL> network_id = network . create ( ) <EOL> subnet = { '<STR_LIT>' : network_id , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:4> , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : False } <EOL> subnet = self . neutron . create_subnet ( { '<STR_LIT>' : subnet } ) <EOL> subnet_id = subnet [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] <EOL> time . sleep ( utils . DEFAULT_CMD_TIMEOUT ) <EOL> flows_after_change = ovs . dump ( ) <EOL> updated_subnet = { '<STR_LIT>' : True } <EOL> self . neutron . update_subnet ( subnet_id , { '<STR_LIT>' : updated_subnet } ) <EOL> dhcp_ip = utils . wait_until_is_and_return ( <EOL> lambda : self . get_dhcp_ip ( network_id , subnet_id ) , <EOL> exception = Exception ( '<STR_LIT>' ) <EOL> ) <EOL> self . assertFalse ( self . check_dhcp_rule ( flows_before_change , dhcp_ip ) ) <EOL> self . assertFalse ( self . check_dhcp_rule ( flows_after_change , dhcp_ip ) ) <EOL> wait_until_true ( <EOL> lambda : self . check_dhcp_rule ( ovs . dump ( ) , dhcp_ip ) , <EOL> exception = Exception ( '<STR_LIT>' ) , <EOL> timeout = <NUM_LIT:5> <EOL> ) <EOL> network . close ( ) <EOL> utils . wait_until_none ( <EOL> lambda : self . check_dhcp_rule ( ovs . dump ( ) , dhcp_ip ) , <EOL> exception = Exception ( '<STR_LIT>' ) , <EOL> timeout = <NUM_LIT:30> <EOL> ) <EOL> def test_create_router_interface ( self ) : <EOL> ovs = utils . OvsFlowsParser ( ) <EOL> flows_before_change = ovs . dump ( ) <EOL> router = self . store ( objects . RouterTestObj ( self . neutron , self . nb_api ) ) <EOL> network = self . store ( objects . NetworkTestObj ( self . neutron , self . nb_api ) ) <EOL> network_id = network . create ( ) <EOL> subnet = { '<STR_LIT>' : network_id , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:4> , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : True } <EOL> subnet = self . neutron . create_subnet ( { '<STR_LIT>' : subnet } ) <EOL> subnet_id = subnet [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] <EOL> router_id = router . create ( ) <EOL> self . assertTrue ( router . exists ( ) ) <EOL> subnet_msg = { '<STR_LIT>' : subnet_id } <EOL> time . sleep ( utils . DEFAULT_CMD_TIMEOUT ) <EOL> self . neutron . add_interface_router ( router_id , body = subnet_msg ) <EOL> dhcp_ip = utils . wait_until_is_and_return ( <EOL> lambda : self . get_dhcp_ip ( network_id , subnet_id ) , <EOL> exception = Exception ( '<STR_LIT>' ) <EOL> ) <EOL> flows_after_change = ovs . dump ( ) <EOL> self . assertFalse ( self . check_dhcp_rule ( flows_before_change , dhcp_ip ) ) <EOL> self . assertTrue ( self . check_dhcp_rule ( flows_after_change , dhcp_ip ) ) <EOL> self . neutron . remove_interface_router ( router_id , body = subnet_msg ) <EOL> router . close ( ) <EOL> network . close ( ) <EOL> utils . wait_until_none ( <EOL> lambda : self . check_dhcp_rule ( ovs . dump ( ) , dhcp_ip ) , <EOL> exception = Exception ( '<STR_LIT>' ) , <EOL> timeout = <NUM_LIT:30> <EOL> ) </s>
