<s> import unittest <EOL> from yaml import safe_load <EOL> import os <EOL> import tempfile <EOL> from swiftlm . cli . supervisor import generate_delta , rebalance <EOL> from swiftlm . rings . ring_model import RingSpecifications , RingSpecification , DeviceInfo , DriveConfiguration , DriveConfigurations , SwiftModelException , KeystoneRegionRings <EOL> from swiftlm . rings . hlm_model import InputModel , DeviceInfo <EOL> from swiftlm . rings . ring_builder import RingDelta , RingBuilder <EOL> from tests . data . ring_padawan import padawan_input_model , padawan_net_hosts , padawan_swf_rng_consumes , padawan_drive_configurations , expected_cmds <EOL> class FakeRingBuilder ( RingBuilder ) : <EOL> def __init__ ( self , builder_dir , region_name , fake_rings , replica_count ) : <EOL> super ( FakeRingBuilder , self ) . __init__ ( builder_dir , False ) <EOL> for ring_name in fake_rings : <EOL> self . register_ring ( region_name , ring_name , replica_count , <NUM_LIT:100> ) <EOL> def load_fake_ring_data ( self , delta ) : <EOL> for device_info in delta . delta_devices : <EOL> self . flat_device_list . append ( device_info ) <EOL> def fake_set_weights ( self , new_weight ) : <EOL> for device_info in self . flat_device_list : <EOL> device_info [ '<STR_LIT>' ] = new_weight <EOL> class DummyInputOptions ( object ) : <EOL> def __init__ ( self ) : <EOL> self . size_to_weight = float ( <NUM_LIT> * <NUM_LIT> * <NUM_LIT> ) <EOL> self . allow_partitions = False <EOL> self . stop_on_warnings = True <EOL> self . weight_step = None <EOL> def dummy_osconfig_load ( text ) : <EOL> model = safe_load ( text ) <EOL> drive_configurations = DriveConfigurations ( ) <EOL> for dc_model in model . get ( '<STR_LIT>' ) : <EOL> drive_configuration = DriveConfiguration ( ) <EOL> drive_configuration . load_model ( dc_model ) <EOL> drive_configurations . add ( drive_configuration ) <EOL> return drive_configurations <EOL> def cmd_key ( cmd ) : <EOL> """<STR_LIT>""" <EOL> verb , ringname , args = cmd <EOL> return ringname + '<STR_LIT:.>' + verb + '<STR_LIT:.>' + '<STR_LIT:.>' . join ( args ) <EOL> def cmd_parse ( text ) : <EOL> """<STR_LIT>""" <EOL> if text == '<STR_LIT>' : <EOL> return None <EOL> words = text . split ( ) <EOL> if len ( words ) == <NUM_LIT:0> : <EOL> return None <EOL> ringname = os . path . basename ( words [ <NUM_LIT:1> ] ) <EOL> verb = words [ <NUM_LIT:2> ] <EOL> args = words [ <NUM_LIT:3> : ] <EOL> return ( verb , ringname , args ) <EOL> def assert_cmds_are_same ( test_case , expected , actual ) : <EOL> expected_args = [ ] <EOL> actual_args = [ ] <EOL> for cmd in expected : <EOL> cmd_tuple = cmd_parse ( cmd ) <EOL> if cmd_tuple : <EOL> expected_args . append ( cmd_tuple ) <EOL> for cmd in actual : <EOL> cmd_tuple = cmd_parse ( cmd ) <EOL> if cmd_tuple : <EOL> actual_args . append ( cmd_tuple ) <EOL> expected_args . sort ( key = cmd_key ) <EOL> actual_args . sort ( key = cmd_key ) <EOL> test_case . assertEquals ( expected_args , actual_args ) <EOL> def verb_ringname_args_in_cmds ( verb , ringname , args , cmds ) : <EOL> for cmd in cmds : <EOL> cmdverb , cmdringname , cmdargs = cmd_parse ( cmd ) <EOL> if cmdverb == verb and cmdringname == ringname : <EOL> if not cmdargs : <EOL> return True <EOL> if cmdargs == args : <EOL> return True <EOL> return False <EOL> def cleanup ( path ) : <EOL> for name in os . listdir ( path ) : <EOL> if os . path . isdir ( os . path . join ( path , name ) ) : <EOL> cleanup ( os . path . join ( path , name ) ) <EOL> else : <EOL> os . unlink ( os . path . join ( path , name ) ) <EOL> os . rmdir ( path ) <EOL> class TestPadawan ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> tf = tempfile . NamedTemporaryFile ( ) <EOL> self . builder_dir = os . path . basename ( tf . name ) + '<STR_LIT>' <EOL> os . mkdir ( self . builder_dir ) <EOL> def tearDown ( self ) : <EOL> cleanup ( self . builder_dir ) <EOL> def test_build_rings ( self ) : <EOL> options = DummyInputOptions ( ) <EOL> input_model = InputModel ( config = safe_load ( padawan_input_model ) , <EOL> hosts_fd = padawan_net_hosts , <EOL> consumes = padawan_swf_rng_consumes ) <EOL> ring_model = RingSpecifications ( model = safe_load ( <EOL> padawan_input_model ) ) <EOL> rings = RingBuilder ( self . builder_dir , False ) <EOL> drive_configurations = dummy_osconfig_load ( <EOL> padawan_drive_configurations ) <EOL> delta = RingDelta ( ) <EOL> generate_delta ( input_model , ring_model , rings , <EOL> drive_configurations , options , delta ) <EOL> cmds = rebalance ( delta , rings , True ) <EOL> assert_cmds_are_same ( self , expected_cmds , cmds ) <EOL> def test_missmatched_region_names ( self ) : <EOL> options = DummyInputOptions ( ) <EOL> input_model = InputModel ( config = safe_load ( padawan_input_model ) , <EOL> hosts_fd = padawan_net_hosts , <EOL> consumes = padawan_swf_rng_consumes ) <EOL> ring_model = RingSpecifications ( model = safe_load ( padawan_input_model ) ) <EOL> rings = RingBuilder ( self . builder_dir , False ) <EOL> drive_configurations = dummy_osconfig_load ( <EOL> padawan_drive_configurations ) <EOL> delta = RingDelta ( ) <EOL> for region_rings in ring_model . keystone_ring_specifications : <EOL> region_rings . region_name = '<STR_LIT>' <EOL> self . assertRaises ( SwiftModelException , generate_delta , input_model , <EOL> ring_model , rings , drive_configurations , options , <EOL> delta ) <EOL> ring_model = RingSpecifications ( model = safe_load ( padawan_input_model ) ) <EOL> keystone_region_rings = KeystoneRegionRings ( self ) <EOL> keystone_region_rings . region_name = '<STR_LIT>' <EOL> ring_model . keystone_ring_specifications . append ( keystone_region_rings ) <EOL> options . stop_on_warnings = False <EOL> generate_delta ( input_model , ring_model , rings , <EOL> drive_configurations , options , delta ) <EOL> options . stop_on_warnings = True <EOL> self . assertRaises ( SwiftModelException , generate_delta , input_model , <EOL> ring_model , rings , drive_configurations , options , <EOL> delta ) <EOL> def test_change_replica_count ( self ) : <EOL> options = DummyInputOptions ( ) <EOL> input_model = InputModel ( config = safe_load ( padawan_input_model ) , <EOL> hosts_fd = padawan_net_hosts , <EOL> consumes = padawan_swf_rng_consumes ) <EOL> ring_model = RingSpecifications ( model = safe_load ( <EOL> padawan_input_model ) ) <EOL> rings = FakeRingBuilder ( self . builder_dir , '<STR_LIT>' , [ '<STR_LIT>' ] , <EOL> <NUM_LIT> ) <EOL> drive_configurations = dummy_osconfig_load ( <EOL> padawan_drive_configurations ) <EOL> delta = RingDelta ( ) <EOL> generate_delta ( input_model , ring_model , rings , <EOL> drive_configurations , options , delta ) <EOL> cmds = rebalance ( delta , rings , True ) <EOL> self . assertTrue ( verb_ringname_args_in_cmds ( '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> [ '<STR_LIT>' ] , cmds ) ) <EOL> self . assertTrue ( not verb_ringname_args_in_cmds ( '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> None , cmds ) ) <EOL> self . assertTrue ( verb_ringname_args_in_cmds ( '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , cmds ) ) <EOL> self . assertTrue ( verb_ringname_args_in_cmds ( '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , cmds ) ) <EOL> def test_noop ( self ) : <EOL> options = DummyInputOptions ( ) <EOL> input_model = InputModel ( config = safe_load ( padawan_input_model ) , <EOL> hosts_fd = padawan_net_hosts , <EOL> consumes = padawan_swf_rng_consumes ) <EOL> ring_model = RingSpecifications ( model = safe_load ( <EOL> padawan_input_model ) ) <EOL> rings = FakeRingBuilder ( self . builder_dir , '<STR_LIT>' , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] , <EOL> <NUM_LIT> ) <EOL> drive_configurations = dummy_osconfig_load ( <EOL> padawan_drive_configurations ) <EOL> delta = RingDelta ( ) <EOL> generate_delta ( input_model , ring_model , rings , <EOL> drive_configurations , options , delta ) <EOL> rings . load_fake_ring_data ( delta ) <EOL> delta = RingDelta ( ) <EOL> generate_delta ( input_model , ring_model , rings , <EOL> drive_configurations , options , delta ) <EOL> cmds = rebalance ( delta , rings , True ) <EOL> self . assertTrue ( len ( cmds ) == <NUM_LIT:7> ) <EOL> def test_set_weight_no_step ( self ) : <EOL> options = DummyInputOptions ( ) <EOL> input_model = InputModel ( config = safe_load ( padawan_input_model ) , <EOL> hosts_fd = padawan_net_hosts , <EOL> consumes = padawan_swf_rng_consumes ) <EOL> ring_model = RingSpecifications ( model = safe_load ( <EOL> padawan_input_model ) ) <EOL> rings = FakeRingBuilder ( self . builder_dir , '<STR_LIT>' , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> <NUM_LIT> ) <EOL> drive_configurations = dummy_osconfig_load ( <EOL> padawan_drive_configurations ) <EOL> delta = RingDelta ( ) <EOL> generate_delta ( input_model , ring_model , rings , <EOL> drive_configurations , options , delta ) <EOL> rings . load_fake_ring_data ( delta ) <EOL> rings . fake_set_weights ( <NUM_LIT:1.0> ) <EOL> delta = RingDelta ( ) <EOL> generate_delta ( input_model , ring_model , rings , <EOL> drive_configurations , options , delta ) <EOL> cmds = rebalance ( delta , rings , True ) <EOL> self . assertTrue ( '<STR_LIT>' <EOL> '<STR_LIT>' in '<STR_LIT:U+0020>' . join ( cmds ) ) <EOL> def test_set_weight_with_step ( self ) : <EOL> options = DummyInputOptions ( ) <EOL> input_model = InputModel ( config = safe_load ( padawan_input_model ) , <EOL> hosts_fd = padawan_net_hosts , <EOL> consumes = padawan_swf_rng_consumes ) <EOL> ring_model = RingSpecifications ( model = safe_load ( <EOL> padawan_input_model ) ) <EOL> rings = FakeRingBuilder ( self . builder_dir , '<STR_LIT>' , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> <NUM_LIT> ) <EOL> drive_configurations = dummy_osconfig_load ( <EOL> padawan_drive_configurations ) <EOL> delta = RingDelta ( ) <EOL> generate_delta ( input_model , ring_model , rings , <EOL> drive_configurations , options , delta ) <EOL> rings . load_fake_ring_data ( delta ) <EOL> rings . fake_set_weights ( <NUM_LIT:1.0> ) <EOL> options . weight_step = '<STR_LIT>' <EOL> delta = RingDelta ( ) <EOL> generate_delta ( input_model , ring_model , rings , <EOL> drive_configurations , options , delta ) <EOL> cmds = rebalance ( delta , rings , True ) <EOL> self . assertTrue ( '<STR_LIT>' <EOL> '<STR_LIT>' in '<STR_LIT:U+0020>' . join ( cmds ) ) <EOL> rings = FakeRingBuilder ( self . builder_dir , '<STR_LIT>' , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> <NUM_LIT> ) <EOL> rings . load_fake_ring_data ( delta ) <EOL> options . weight_step = '<STR_LIT:1.0>' <EOL> delta = RingDelta ( ) <EOL> generate_delta ( input_model , ring_model , rings , <EOL> drive_configurations , options , delta ) <EOL> cmds = rebalance ( delta , rings , True ) <EOL> self . assertTrue ( '<STR_LIT>' <EOL> '<STR_LIT>' in '<STR_LIT:U+0020>' . join ( cmds ) ) <EOL> rings = FakeRingBuilder ( self . builder_dir , '<STR_LIT>' , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> <NUM_LIT> ) <EOL> rings . load_fake_ring_data ( delta ) <EOL> options . weight_step = '<STR_LIT>' <EOL> delta = RingDelta ( ) <EOL> generate_delta ( input_model , ring_model , rings , <EOL> drive_configurations , options , delta ) <EOL> cmds = rebalance ( delta , rings , True ) <EOL> self . assertTrue ( '<STR_LIT>' <EOL> '<STR_LIT>' in '<STR_LIT:U+0020>' . join ( cmds ) ) <EOL> def test_set_weight_down ( self ) : <EOL> options = DummyInputOptions ( ) <EOL> input_model = InputModel ( config = safe_load ( padawan_input_model ) , <EOL> hosts_fd = padawan_net_hosts , <EOL> consumes = padawan_swf_rng_consumes ) <EOL> ring_model = RingSpecifications ( model = safe_load ( <EOL> padawan_input_model ) ) <EOL> rings = FakeRingBuilder ( self . builder_dir , '<STR_LIT>' , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> <NUM_LIT> ) <EOL> drive_configurations = dummy_osconfig_load ( <EOL> padawan_drive_configurations ) <EOL> delta = RingDelta ( ) <EOL> generate_delta ( input_model , ring_model , rings , <EOL> drive_configurations , options , delta ) <EOL> rings . load_fake_ring_data ( delta ) <EOL> rings . fake_set_weights ( <NUM_LIT> ) <EOL> options . weight_step = '<STR_LIT>' <EOL> delta = RingDelta ( ) <EOL> generate_delta ( input_model , ring_model , rings , <EOL> drive_configurations , options , delta ) <EOL> cmds = rebalance ( delta , rings , True ) <EOL> self . assertTrue ( '<STR_LIT>' <EOL> '<STR_LIT>' in '<STR_LIT:U+0020>' . join ( cmds ) ) <EOL> rings = FakeRingBuilder ( self . builder_dir , '<STR_LIT>' , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> <NUM_LIT> ) <EOL> rings . load_fake_ring_data ( delta ) <EOL> options . weight_step = '<STR_LIT:1.0>' <EOL> delta = RingDelta ( ) <EOL> generate_delta ( input_model , ring_model , rings , <EOL> drive_configurations , options , delta ) <EOL> cmds = rebalance ( delta , rings , True ) <EOL> self . assertTrue ( '<STR_LIT>' <EOL> '<STR_LIT>' in '<STR_LIT:U+0020>' . join ( cmds ) ) <EOL> rings = FakeRingBuilder ( self . builder_dir , '<STR_LIT>' , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> <NUM_LIT> ) <EOL> rings . load_fake_ring_data ( delta ) <EOL> options . weight_step = '<STR_LIT>' <EOL> delta = RingDelta ( ) <EOL> generate_delta ( input_model , ring_model , rings , <EOL> drive_configurations , options , delta ) <EOL> cmds = rebalance ( delta , rings , True ) <EOL> self . assertTrue ( '<STR_LIT>' <EOL> '<STR_LIT>' in '<STR_LIT:U+0020>' . join ( cmds ) ) <EOL> def test_add_servers ( self ) : <EOL> options = DummyInputOptions ( ) <EOL> input_model = InputModel ( config = safe_load ( padawan_input_model ) , <EOL> hosts_fd = padawan_net_hosts , <EOL> consumes = padawan_swf_rng_consumes ) <EOL> ring_model = RingSpecifications ( model = safe_load ( <EOL> padawan_input_model ) ) <EOL> rings = RingBuilder ( self . builder_dir , False ) <EOL> drive_configurations = dummy_osconfig_load ( <EOL> padawan_drive_configurations ) <EOL> options . weight_step = '<STR_LIT>' <EOL> delta = RingDelta ( ) <EOL> generate_delta ( input_model , ring_model , rings , <EOL> drive_configurations , options , delta ) <EOL> cmds = rebalance ( delta , rings , True ) <EOL> self . assertTrue ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' in '<STR_LIT:U+0020>' . join ( cmds ) ) <EOL> self . assertTrue ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' in '<STR_LIT:U+0020>' . join ( cmds ) ) </s>
