<s> import mock <EOL> from oslo_db import exception as exc <EOL> from sqlalchemy . orm import query <EOL> import neutron . db . api as db <EOL> from neutron . plugins . ml2 . drivers import type_vlan <EOL> from neutron . tests . unit import testlib_api <EOL> TENANT_NET = '<STR_LIT>' <EOL> VLAN_MIN = <NUM_LIT:200> <EOL> VLAN_MAX = <NUM_LIT> <EOL> VLAN_OUTSIDE = <NUM_LIT:100> <EOL> NETWORK_VLAN_RANGES = { <EOL> TENANT_NET : [ ( VLAN_MIN , VLAN_MAX ) ] , <EOL> } <EOL> class HelpersTest ( testlib_api . SqlTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( HelpersTest , self ) . setUp ( ) <EOL> self . driver = type_vlan . VlanTypeDriver ( ) <EOL> self . driver . network_vlan_ranges = NETWORK_VLAN_RANGES <EOL> self . driver . _sync_vlan_allocations ( ) <EOL> self . session = db . get_session ( ) <EOL> def check_raw_segment ( self , expected , observed ) : <EOL> for key , value in expected . items ( ) : <EOL> self . assertEqual ( value , observed [ key ] ) <EOL> def test_primary_keys ( self ) : <EOL> self . assertEqual ( set ( [ '<STR_LIT>' , '<STR_LIT>' ] ) , <EOL> self . driver . primary_keys ) <EOL> def test_allocate_specific_unallocated_segment_in_pools ( self ) : <EOL> expected = dict ( physical_network = TENANT_NET , vlan_id = VLAN_MIN ) <EOL> observed = self . driver . allocate_fully_specified_segment ( self . session , <EOL> ** expected ) <EOL> self . check_raw_segment ( expected , observed ) <EOL> def test_allocate_specific_allocated_segment_in_pools ( self ) : <EOL> raw_segment = dict ( physical_network = TENANT_NET , vlan_id = VLAN_MIN ) <EOL> self . driver . allocate_fully_specified_segment ( self . session , <EOL> ** raw_segment ) <EOL> observed = self . driver . allocate_fully_specified_segment ( self . session , <EOL> ** raw_segment ) <EOL> self . assertIsNone ( observed ) <EOL> def test_allocate_specific_finally_allocated_segment_in_pools ( self ) : <EOL> raw_segment = dict ( physical_network = TENANT_NET , vlan_id = VLAN_MIN ) <EOL> with mock . patch . object ( query . Query , '<STR_LIT>' , return_value = <NUM_LIT:0> ) : <EOL> observed = self . driver . allocate_fully_specified_segment ( <EOL> self . session , ** raw_segment ) <EOL> self . assertIsNone ( observed ) <EOL> def test_allocate_specific_unallocated_segment_outside_pools ( self ) : <EOL> expected = dict ( physical_network = TENANT_NET , vlan_id = VLAN_OUTSIDE ) <EOL> observed = self . driver . allocate_fully_specified_segment ( self . session , <EOL> ** expected ) <EOL> self . check_raw_segment ( expected , observed ) <EOL> def test_allocate_specific_allocated_segment_outside_pools ( self ) : <EOL> raw_segment = dict ( physical_network = TENANT_NET , vlan_id = VLAN_OUTSIDE ) <EOL> self . driver . allocate_fully_specified_segment ( self . session , <EOL> ** raw_segment ) <EOL> observed = self . driver . allocate_fully_specified_segment ( self . session , <EOL> ** raw_segment ) <EOL> self . assertIsNone ( observed ) <EOL> def test_allocate_specific_finally_unallocated_segment_outside_pools ( self ) : <EOL> expected = dict ( physical_network = TENANT_NET , vlan_id = VLAN_MIN ) <EOL> with mock . patch . object ( self . driver . model , '<STR_LIT>' ) : <EOL> observed = self . driver . allocate_fully_specified_segment ( <EOL> self . session , ** expected ) <EOL> self . check_raw_segment ( expected , observed ) <EOL> def test_allocate_partial_segment_without_filters ( self ) : <EOL> expected = dict ( physical_network = TENANT_NET ) <EOL> observed = self . driver . allocate_partially_specified_segment ( <EOL> self . session ) <EOL> self . check_raw_segment ( expected , observed ) <EOL> def test_allocate_partial_segment_with_filter ( self ) : <EOL> expected = dict ( physical_network = TENANT_NET ) <EOL> observed = self . driver . allocate_partially_specified_segment ( <EOL> self . session , ** expected ) <EOL> self . check_raw_segment ( expected , observed ) <EOL> def test_allocate_partial_segment_no_resource_available ( self ) : <EOL> for i in range ( VLAN_MIN , VLAN_MAX + <NUM_LIT:1> ) : <EOL> self . driver . allocate_partially_specified_segment ( self . session ) <EOL> observed = self . driver . allocate_partially_specified_segment ( <EOL> self . session ) <EOL> self . assertIsNone ( observed ) <EOL> def test_allocate_partial_segment_outside_pools ( self ) : <EOL> raw_segment = dict ( physical_network = '<STR_LIT>' ) <EOL> observed = self . driver . allocate_partially_specified_segment ( <EOL> self . session , ** raw_segment ) <EOL> self . assertIsNone ( observed ) <EOL> def test_allocate_partial_segment_first_attempt_fails ( self ) : <EOL> expected = dict ( physical_network = TENANT_NET ) <EOL> with mock . patch . object ( query . Query , '<STR_LIT>' , side_effect = [ <NUM_LIT:0> , <NUM_LIT:1> ] ) : <EOL> self . assertRaises ( <EOL> exc . RetryRequest , <EOL> self . driver . allocate_partially_specified_segment , <EOL> self . session , ** expected ) <EOL> observed = self . driver . allocate_partially_specified_segment ( <EOL> self . session , ** expected ) <EOL> self . check_raw_segment ( expected , observed ) </s>
