<s> """<STR_LIT>""" <EOL> import os <EOL> import random <EOL> import unittest <EOL> from ganeti import compat <EOL> from ganeti import constants <EOL> from ganeti import errors <EOL> from ganeti import objects <EOL> from ganeti import utils <EOL> from ganeti . storage import bdev <EOL> import testutils <EOL> class TestRADOSBlockDevice ( testutils . GanetiTestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> testutils . GanetiTestCase . setUp ( self ) <EOL> self . plain_output_old_ok = testutils . ReadTestData ( "<STR_LIT>" ) <EOL> self . plain_output_old_no_matches = testutils . ReadTestData ( "<STR_LIT>" ) <EOL> self . plain_output_old_extra_matches = testutils . ReadTestData ( "<STR_LIT>" ) <EOL> self . plain_output_old_empty = testutils . ReadTestData ( "<STR_LIT>" ) <EOL> self . plain_output_new_ok = testutils . ReadTestData ( "<STR_LIT>" ) <EOL> self . plain_output_new_no_matches = testutils . ReadTestData ( "<STR_LIT>" ) <EOL> self . plain_output_new_extra_matches = testutils . ReadTestData ( "<STR_LIT>" ) <EOL> self . plain_output_new_empty = "<STR_LIT>" <EOL> self . json_output_ok = testutils . ReadTestData ( "<STR_LIT>" ) <EOL> self . json_output_no_matches = testutils . ReadTestData ( "<STR_LIT>" ) <EOL> self . json_output_extra_matches = testutils . ReadTestData ( "<STR_LIT>" ) <EOL> self . json_output_empty = testutils . ReadTestData ( "<STR_LIT>" ) <EOL> self . output_invalid = testutils . ReadTestData ( "<STR_LIT>" ) <EOL> self . volume_name = "<STR_LIT>" <EOL> self . test_unique_id = ( "<STR_LIT>" , self . volume_name ) <EOL> self . test_params = { <EOL> constants . LDP_POOL : "<STR_LIT>" <EOL> } <EOL> def test_ParseRbdShowmappedJson ( self ) : <EOL> parse_function = bdev . RADOSBlockDevice . _ParseRbdShowmappedJson <EOL> self . assertEqual ( parse_function ( self . json_output_ok , self . volume_name ) , <EOL> "<STR_LIT>" ) <EOL> self . assertEqual ( parse_function ( self . json_output_empty , self . volume_name ) , <EOL> None ) <EOL> self . assertEqual ( parse_function ( self . json_output_no_matches , <EOL> self . volume_name ) , None ) <EOL> self . assertRaises ( errors . BlockDeviceError , parse_function , <EOL> self . json_output_extra_matches , self . volume_name ) <EOL> self . assertRaises ( errors . BlockDeviceError , parse_function , <EOL> self . output_invalid , self . volume_name ) <EOL> def test_ParseRbdShowmappedPlain ( self ) : <EOL> parse_function = bdev . RADOSBlockDevice . _ParseRbdShowmappedPlain <EOL> self . assertEqual ( parse_function ( self . plain_output_new_ok , <EOL> self . volume_name ) , "<STR_LIT>" ) <EOL> self . assertEqual ( parse_function ( self . plain_output_old_ok , <EOL> self . volume_name ) , "<STR_LIT>" ) <EOL> self . assertEqual ( parse_function ( self . plain_output_new_empty , <EOL> self . volume_name ) , None ) <EOL> self . assertEqual ( parse_function ( self . plain_output_old_empty , <EOL> self . volume_name ) , None ) <EOL> self . assertEqual ( parse_function ( self . plain_output_new_no_matches , <EOL> self . volume_name ) , None ) <EOL> self . assertEqual ( parse_function ( self . plain_output_old_no_matches , <EOL> self . volume_name ) , None ) <EOL> self . assertRaises ( errors . BlockDeviceError , parse_function , <EOL> self . plain_output_new_extra_matches , self . volume_name ) <EOL> self . assertRaises ( errors . BlockDeviceError , parse_function , <EOL> self . plain_output_old_extra_matches , self . volume_name ) <EOL> self . assertRaises ( errors . BlockDeviceError , parse_function , <EOL> self . output_invalid , self . volume_name ) <EOL> @ testutils . patch_object ( utils , "<STR_LIT>" ) <EOL> @ testutils . patch_object ( bdev . RADOSBlockDevice , "<STR_LIT>" ) <EOL> @ testutils . patch_object ( bdev . RADOSBlockDevice , "<STR_LIT>" ) <EOL> def testRADOSBlockDeviceImport ( self , attach_mock , unmap_mock , run_cmd_mock ) : <EOL> """<STR_LIT>""" <EOL> attach_mock . return_value = True <EOL> run_cmd_mock . return_value = utils . RunResult ( <NUM_LIT:0> , None , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , utils . process . _TIMEOUT_NONE , <NUM_LIT:0> ) <EOL> inst = bdev . RADOSBlockDevice ( self . test_unique_id , [ ] , <NUM_LIT> , <EOL> self . test_params , { } ) <EOL> import_cmd = [ constants . RBD_CMD , "<STR_LIT>" , <EOL> "<STR_LIT>" , inst . rbd_pool , <EOL> "<STR_LIT:->" , inst . rbd_name ] <EOL> self . assertEqual ( inst . Import ( ) , import_cmd ) <EOL> @ testutils . patch_object ( bdev . RADOSBlockDevice , "<STR_LIT>" ) <EOL> def testRADOSBlockDeviceExport ( self , attach_mock ) : <EOL> """<STR_LIT>""" <EOL> attach_mock . return_value = True <EOL> inst = bdev . RADOSBlockDevice ( self . test_unique_id , [ ] , <NUM_LIT> , <EOL> self . test_params , { } ) <EOL> export_cmd = [ constants . RBD_CMD , "<STR_LIT>" , <EOL> "<STR_LIT>" , inst . rbd_pool , <EOL> inst . rbd_name , "<STR_LIT:->" ] <EOL> self . assertEqual ( inst . Export ( ) , export_cmd ) <EOL> class TestExclusiveStoragePvs ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> _EPS = <NUM_LIT> <EOL> _MARGIN = constants . PART_MARGIN + constants . PART_RESERVED + _EPS <EOL> @ staticmethod <EOL> def _GenerateRandomPvInfo ( rnd , name , vg ) : <EOL> size = rnd . randint ( <NUM_LIT> * <NUM_LIT:100> , <NUM_LIT:10> * <NUM_LIT> * <NUM_LIT> * <NUM_LIT:100> ) <EOL> if rnd . choice ( [ False , True ] ) : <EOL> free = float ( rnd . randint ( <NUM_LIT:0> , size ) ) / <NUM_LIT> <EOL> else : <EOL> free = float ( size ) / <NUM_LIT> <EOL> size = float ( size ) / <NUM_LIT> <EOL> attr = "<STR_LIT>" <EOL> return objects . LvmPvInfo ( name = name , vg_name = vg , size = size , free = free , <EOL> attributes = attr ) <EOL> def testGetStdPvSize ( self ) : <EOL> """<STR_LIT>""" <EOL> rnd = random . Random ( <NUM_LIT> ) <EOL> for _ in range ( <NUM_LIT:0> , <NUM_LIT:50> ) : <EOL> pvi = self . _GenerateRandomPvInfo ( rnd , "<STR_LIT>" , "<STR_LIT>" ) <EOL> onesize = bdev . LogicalVolume . _GetStdPvSize ( [ pvi ] ) <EOL> self . assertTrue ( onesize <= pvi . size ) <EOL> self . assertTrue ( onesize > pvi . size * ( <NUM_LIT:1> - self . _MARGIN ) ) <EOL> for length in range ( <NUM_LIT:2> , <NUM_LIT:10> ) : <EOL> n_size = bdev . LogicalVolume . _GetStdPvSize ( [ pvi ] * length ) <EOL> self . assertEqual ( onesize , n_size ) <EOL> for length in range ( <NUM_LIT:1> , <NUM_LIT:10> ) : <EOL> pvlist = [ self . _GenerateRandomPvInfo ( rnd , "<STR_LIT>" , "<STR_LIT>" ) <EOL> for _ in range ( <NUM_LIT:0> , length ) ] <EOL> std_size = bdev . LogicalVolume . _GetStdPvSize ( pvlist ) <EOL> self . assertTrue ( compat . all ( std_size <= pvi . size for pvi in pvlist ) ) <EOL> self . assertTrue ( compat . any ( std_size > pvi . size * ( <NUM_LIT:1> - self . _MARGIN ) <EOL> for pvi in pvlist ) ) <EOL> pvlist . append ( pvlist [ <NUM_LIT:0> ] ) <EOL> p1_size = bdev . LogicalVolume . _GetStdPvSize ( pvlist ) <EOL> self . assertEqual ( std_size , p1_size ) <EOL> def testComputeNumPvs ( self ) : <EOL> """<STR_LIT>""" <EOL> rnd = random . Random ( <NUM_LIT> ) <EOL> for _ in range ( <NUM_LIT:0> , <NUM_LIT:1000> ) : <EOL> pvlist = [ self . _GenerateRandomPvInfo ( rnd , "<STR_LIT>" , "<STR_LIT>" ) ] <EOL> lv_size = float ( rnd . randint ( <NUM_LIT:10> * <NUM_LIT:100> , <NUM_LIT> * <NUM_LIT> * <NUM_LIT:100> ) ) / <NUM_LIT> <EOL> num_pv = bdev . LogicalVolume . _ComputeNumPvs ( lv_size , pvlist ) <EOL> std_size = bdev . LogicalVolume . _GetStdPvSize ( pvlist ) <EOL> self . assertTrue ( num_pv >= <NUM_LIT:1> ) <EOL> self . assertTrue ( num_pv * std_size >= lv_size ) <EOL> self . assertTrue ( ( num_pv - <NUM_LIT:1> ) * std_size < lv_size * ( <NUM_LIT:1> + self . _EPS ) ) <EOL> def testGetEmptyPvNames ( self ) : <EOL> """<STR_LIT>""" <EOL> rnd = random . Random ( <NUM_LIT> ) <EOL> for _ in range ( <NUM_LIT:0> , <NUM_LIT:100> ) : <EOL> num_pvs = rnd . randint ( <NUM_LIT:1> , <NUM_LIT:20> ) <EOL> pvlist = [ self . _GenerateRandomPvInfo ( rnd , "<STR_LIT>" % n , "<STR_LIT>" ) <EOL> for n in range ( <NUM_LIT:0> , num_pvs ) ] <EOL> for num_req in range ( <NUM_LIT:1> , num_pvs + <NUM_LIT:2> ) : <EOL> epvs = bdev . LogicalVolume . _GetEmptyPvNames ( pvlist , num_req ) <EOL> epvs_set = compat . UniqueFrozenset ( epvs ) <EOL> if len ( epvs ) > <NUM_LIT:1> : <EOL> self . assertEqual ( len ( epvs ) , len ( epvs_set ) ) <EOL> for pvi in pvlist : <EOL> if pvi . name in epvs_set : <EOL> self . assertEqual ( pvi . size , pvi . free ) <EOL> else : <EOL> self . assertTrue ( len ( epvs ) == num_req or pvi . free != pvi . size ) <EOL> class TestLogicalVolume ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def testParseLvInfoLine ( self ) : <EOL> """<STR_LIT>""" <EOL> broken_lines = [ <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> ] <EOL> for broken in broken_lines : <EOL> self . assertRaises ( errors . BlockDeviceError , <EOL> bdev . LogicalVolume . _ParseLvInfoLine , broken , "<STR_LIT:#>" ) <EOL> true_out = [ <EOL> ( "<STR_LIT>" , <NUM_LIT> , <NUM_LIT:3> , <NUM_LIT> , <NUM_LIT:2> , [ "<STR_LIT>" ] ) , <EOL> ( "<STR_LIT>" , <NUM_LIT> , <NUM_LIT:7> , <NUM_LIT> , <NUM_LIT:4> , [ "<STR_LIT>" ] ) , <EOL> ( "<STR_LIT>" , <NUM_LIT> , <NUM_LIT:4> , <NUM_LIT> , <NUM_LIT:5> , [ "<STR_LIT>" , "<STR_LIT>" ] ) , <EOL> ( "<STR_LIT>" , <NUM_LIT:15> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT:32> , [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] ) , <EOL> ( "<STR_LIT>" , <NUM_LIT:15> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT:32> , [ ] ) , <EOL> ] <EOL> for exp in true_out : <EOL> for sep in "<STR_LIT>" : <EOL> pvs = "<STR_LIT:U+002C>" . join ( "<STR_LIT>" % ( d , i * <NUM_LIT:12> ) for ( i , d ) in enumerate ( exp [ - <NUM_LIT:1> ] ) ) <EOL> lvs_line = ( sep . join ( ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:%s>" ) ) % <EOL> ( exp [ <NUM_LIT:0> : - <NUM_LIT:1> ] + ( pvs , ) ) ) <EOL> parsed = bdev . LogicalVolume . _ParseLvInfoLine ( lvs_line , sep ) <EOL> self . assertEqual ( parsed , exp ) <EOL> @ staticmethod <EOL> def _FakeRunCmd ( success , stdout ) : <EOL> if success : <EOL> exit_code = <NUM_LIT:0> <EOL> else : <EOL> exit_code = <NUM_LIT:1> <EOL> return lambda cmd : utils . RunResult ( exit_code , None , stdout , "<STR_LIT>" , cmd , <EOL> utils . process . _TIMEOUT_NONE , <NUM_LIT:5> ) <EOL> def testGetLvInfo ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertRaises ( errors . BlockDeviceError , bdev . LogicalVolume . _GetLvInfo , <EOL> "<STR_LIT>" , <EOL> _run_cmd = self . _FakeRunCmd ( False , "<STR_LIT>" ) ) <EOL> self . assertRaises ( errors . BlockDeviceError , bdev . LogicalVolume . _GetLvInfo , <EOL> "<STR_LIT>" , _run_cmd = self . _FakeRunCmd ( True , "<STR_LIT>" ) ) <EOL> self . assertRaises ( errors . BlockDeviceError , bdev . LogicalVolume . _GetLvInfo , <EOL> "<STR_LIT>" , _run_cmd = self . _FakeRunCmd ( True , "<STR_LIT>" ) ) <EOL> good_line = "<STR_LIT>" <EOL> fake_cmd = self . _FakeRunCmd ( True , good_line ) <EOL> good_res = bdev . LogicalVolume . _GetLvInfo ( "<STR_LIT>" , _run_cmd = fake_cmd ) <EOL> for lines in [ <EOL> [ good_line ] * <NUM_LIT:2> , <EOL> [ good_line ] * <NUM_LIT:3> , <EOL> ] : <EOL> fake_cmd = self . _FakeRunCmd ( True , "<STR_LIT:\n>" . join ( lines ) ) <EOL> same_res = bdev . LogicalVolume . _GetLvInfo ( "<STR_LIT>" , fake_cmd ) <EOL> self . assertEqual ( same_res , good_res ) <EOL> one_line = "<STR_LIT>" <EOL> fake_cmd = self . _FakeRunCmd ( True , one_line ) <EOL> one_res = bdev . LogicalVolume . _GetLvInfo ( "<STR_LIT>" , _run_cmd = fake_cmd ) <EOL> for multi_lines in [ <EOL> ( "<STR_LIT>" <EOL> "<STR_LIT>" ) , <EOL> ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) , <EOL> ( "<STR_LIT>" <EOL> "<STR_LIT>" ) , <EOL> ] : <EOL> fake_cmd = self . _FakeRunCmd ( True , multi_lines ) <EOL> multi_res = bdev . LogicalVolume . _GetLvInfo ( "<STR_LIT>" , _run_cmd = fake_cmd ) <EOL> self . assertEqual ( multi_res , one_res ) <EOL> @ testutils . patch_object ( bdev . LogicalVolume , "<STR_LIT>" ) <EOL> def testLogicalVolumeImport ( self , attach_mock ) : <EOL> """<STR_LIT>""" <EOL> attach_mock . return_value = True <EOL> test_unique_id = ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> inst = bdev . LogicalVolume ( test_unique_id , [ ] , <NUM_LIT> , { } , { } ) <EOL> import_cmd = [ constants . DD_CMD , <EOL> "<STR_LIT>" % inst . dev_path , <EOL> "<STR_LIT>" % constants . DD_BLOCK_SIZE , <EOL> "<STR_LIT>" , "<STR_LIT>" ] <EOL> self . assertEqual ( inst . Import ( ) , import_cmd ) <EOL> @ testutils . patch_object ( bdev . LogicalVolume , "<STR_LIT>" ) <EOL> def testLogicalVolumeExport ( self , attach_mock ) : <EOL> """<STR_LIT>""" <EOL> attach_mock . return_value = True <EOL> test_unique_id = ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> inst = bdev . LogicalVolume ( test_unique_id , [ ] , <NUM_LIT> , { } , { } ) <EOL> export_cmd = [ constants . DD_CMD , <EOL> "<STR_LIT>" % inst . dev_path , <EOL> "<STR_LIT>" % constants . DD_BLOCK_SIZE , <EOL> "<STR_LIT>" % inst . size , <EOL> "<STR_LIT>" ] <EOL> self . assertEqual ( inst . Export ( ) , export_cmd ) <EOL> class TestPersistentBlockDevice ( testutils . GanetiTestCase ) : <EOL> """<STR_LIT>""" <EOL> def testPersistentBlockDeviceImport ( self ) : <EOL> """<STR_LIT>""" <EOL> test_unique_id = ( constants . BLOCKDEV_DRIVER_MANUAL , "<STR_LIT>" ) <EOL> inst = bdev . PersistentBlockDevice ( test_unique_id , [ ] , <NUM_LIT> , { } , { } ) <EOL> self . assertRaises ( errors . BlockDeviceError , <EOL> bdev . PersistentBlockDevice . Import , inst ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> testutils . GanetiTestProgram ( ) </s>
