<s> """<STR_LIT>""" <EOL> import os <EOL> import sys <EOL> import re <EOL> import unittest <EOL> import socket <EOL> from ganeti import constants <EOL> from ganeti import objects <EOL> from ganeti import compat <EOL> from ganeti import utils <EOL> from ganeti import errors <EOL> from ganeti import impexpd <EOL> import testutils <EOL> class CmdBuilderConfig ( objects . ConfigObject ) : <EOL> __slots__ = [ <EOL> "<STR_LIT>" , <EOL> "<STR_LIT:key>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT:host>" , <EOL> "<STR_LIT:port>" , <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> def CheckCmdWord ( cmd , word ) : <EOL> wre = re . compile ( r"<STR_LIT>" % re . escape ( word ) ) <EOL> return compat . any ( wre . search ( i ) for i in cmd ) <EOL> class TestCommandBuilder ( unittest . TestCase ) : <EOL> def test ( self ) : <EOL> compress_import = { <EOL> constants . IEC_GZIP : "<STR_LIT>" , <EOL> constants . IEC_GZIP_FAST : "<STR_LIT>" , <EOL> constants . IEC_GZIP_SLOW : "<STR_LIT>" , <EOL> constants . IEC_LZOP : "<STR_LIT>" , <EOL> } <EOL> compress_export = { <EOL> constants . IEC_GZIP : "<STR_LIT>" , <EOL> constants . IEC_GZIP_FAST : "<STR_LIT>" , <EOL> constants . IEC_GZIP_SLOW : "<STR_LIT>" , <EOL> constants . IEC_LZOP : "<STR_LIT>" , <EOL> } <EOL> for mode in [ constants . IEM_IMPORT , constants . IEM_EXPORT ] : <EOL> if mode == constants . IEM_IMPORT : <EOL> compress_dict = compress_import <EOL> elif mode == constants . IEM_EXPORT : <EOL> compress_dict = compress_export <EOL> for compress in constants . IEC_ALL : <EOL> for magic in [ None , <NUM_LIT:10> * "<STR_LIT:->" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" ] : <EOL> opts = CmdBuilderConfig ( magic = magic , compress = compress ) <EOL> builder = impexpd . CommandBuilder ( mode , opts , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ) <EOL> magic_cmd = builder . _GetMagicCommand ( ) <EOL> dd_cmd = builder . _GetDdCommand ( ) <EOL> if magic : <EOL> self . assert_ ( ( "<STR_LIT>" % magic ) in magic_cmd ) <EOL> self . assert_ ( ( "<STR_LIT>" % magic ) in dd_cmd ) <EOL> else : <EOL> self . assertFalse ( magic_cmd ) <EOL> for host in [ "<STR_LIT:localhost>" , "<STR_LIT>" , "<STR_LIT>" ] : <EOL> for port in [ <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] : <EOL> for cmd_prefix in [ None , "<STR_LIT>" , <EOL> "<STR_LIT>" ] : <EOL> for cmd_suffix in [ None , "<STR_LIT>" , <EOL> "<STR_LIT>" ] : <EOL> opts = CmdBuilderConfig ( host = host , port = port , compress = compress , <EOL> cmd_prefix = cmd_prefix , <EOL> cmd_suffix = cmd_suffix ) <EOL> builder = impexpd . CommandBuilder ( mode , opts , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ) <EOL> cmd = builder . GetCommand ( ) <EOL> self . assert_ ( isinstance ( cmd , list ) ) <EOL> if compress != constants . IEC_NONE : <EOL> self . assert_ ( CheckCmdWord ( cmd , compress_dict [ compress ] ) ) <EOL> if cmd_prefix is not None : <EOL> self . assert_ ( compat . any ( cmd_prefix in i for i in cmd ) ) <EOL> if cmd_suffix is not None : <EOL> self . assert_ ( compat . any ( cmd_suffix in i for i in cmd ) ) <EOL> socat_cmd = builder . _GetSocatCommand ( ) <EOL> if mode == constants . IEM_IMPORT : <EOL> ssl_addr = socat_cmd [ - <NUM_LIT:2> ] . split ( "<STR_LIT:U+002C>" ) <EOL> self . assert_ ( ( "<STR_LIT>" % port ) in ssl_addr ) <EOL> elif mode == constants . IEM_EXPORT : <EOL> ssl_addr = socat_cmd [ - <NUM_LIT:1> ] . split ( "<STR_LIT:U+002C>" ) <EOL> self . assert_ ( ( "<STR_LIT>" % ( host , port ) ) in ssl_addr ) <EOL> self . assert_ ( "<STR_LIT>" in ssl_addr ) <EOL> def testIPv6 ( self ) : <EOL> for mode in [ constants . IEM_IMPORT , constants . IEM_EXPORT ] : <EOL> opts = CmdBuilderConfig ( host = "<STR_LIT:localhost>" , port = <NUM_LIT> , <EOL> ipv4 = False , ipv6 = False ) <EOL> builder = impexpd . CommandBuilder ( mode , opts , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ) <EOL> cmd = builder . _GetSocatCommand ( ) <EOL> self . assert_ ( compat . all ( "<STR_LIT>" not in i for i in cmd ) ) <EOL> opts = CmdBuilderConfig ( host = "<STR_LIT:localhost>" , port = <NUM_LIT> , <EOL> ipv4 = True , ipv6 = False ) <EOL> builder = impexpd . CommandBuilder ( mode , opts , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ) <EOL> cmd = builder . _GetSocatCommand ( ) <EOL> self . assert_ ( compat . any ( "<STR_LIT>" in i for i in cmd ) ) <EOL> opts = CmdBuilderConfig ( host = "<STR_LIT:localhost>" , port = <NUM_LIT> , <EOL> ipv4 = False , ipv6 = True ) <EOL> builder = impexpd . CommandBuilder ( mode , opts , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ) <EOL> cmd = builder . _GetSocatCommand ( ) <EOL> self . assert_ ( compat . any ( "<STR_LIT>" in i for i in cmd ) ) <EOL> opts = CmdBuilderConfig ( host = "<STR_LIT:localhost>" , port = <NUM_LIT> , <EOL> ipv4 = True , ipv6 = True ) <EOL> builder = impexpd . CommandBuilder ( mode , opts , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ) <EOL> self . assertRaises ( AssertionError , builder . _GetSocatCommand ) <EOL> def testCommaError ( self ) : <EOL> opts = CmdBuilderConfig ( host = "<STR_LIT:localhost>" , port = <NUM_LIT> , <EOL> ca = "<STR_LIT>" ) <EOL> for mode in [ constants . IEM_IMPORT , constants . IEM_EXPORT ] : <EOL> builder = impexpd . CommandBuilder ( mode , opts , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ) <EOL> self . assertRaises ( errors . GenericError , builder . GetCommand ) <EOL> def testOptionLengthError ( self ) : <EOL> testopts = [ <EOL> CmdBuilderConfig ( bind = "<STR_LIT>" + ( "<STR_LIT:A>" * impexpd . SOCAT_OPTION_MAXLEN ) , <EOL> port = <NUM_LIT> , ca = "<STR_LIT>" ) , <EOL> CmdBuilderConfig ( host = "<STR_LIT:localhost>" , port = <NUM_LIT> , <EOL> ca = "<STR_LIT>" + ( "<STR_LIT:B>" * impexpd . SOCAT_OPTION_MAXLEN ) ) , <EOL> CmdBuilderConfig ( host = "<STR_LIT:localhost>" , port = <NUM_LIT> , <EOL> key = "<STR_LIT>" + ( "<STR_LIT:B>" * impexpd . SOCAT_OPTION_MAXLEN ) ) , <EOL> ] <EOL> for opts in testopts : <EOL> for mode in [ constants . IEM_IMPORT , constants . IEM_EXPORT ] : <EOL> builder = impexpd . CommandBuilder ( mode , opts , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ) <EOL> self . assertRaises ( errors . GenericError , builder . GetCommand ) <EOL> opts . host = "<STR_LIT:localhost>" + ( "<STR_LIT:A>" * impexpd . SOCAT_OPTION_MAXLEN ) <EOL> builder = impexpd . CommandBuilder ( constants . IEM_EXPORT , opts , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ) <EOL> self . assertRaises ( errors . GenericError , builder . GetCommand ) <EOL> def testModeError ( self ) : <EOL> mode = "<STR_LIT>" <EOL> assert mode not in [ constants . IEM_IMPORT , constants . IEM_EXPORT ] <EOL> opts = CmdBuilderConfig ( host = "<STR_LIT:localhost>" , port = <NUM_LIT> ) <EOL> builder = impexpd . CommandBuilder ( mode , opts , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ) <EOL> self . assertRaises ( errors . GenericError , builder . GetCommand ) <EOL> class TestVerifyListening ( unittest . TestCase ) : <EOL> def test ( self ) : <EOL> self . assertEqual ( impexpd . _VerifyListening ( socket . AF_INET , <EOL> "<STR_LIT>" , <NUM_LIT> ) , <EOL> ( "<STR_LIT>" , <NUM_LIT> ) ) <EOL> self . assertEqual ( impexpd . _VerifyListening ( socket . AF_INET6 , "<STR_LIT>" , <NUM_LIT> ) , <EOL> ( "<STR_LIT>" , <NUM_LIT> ) ) <EOL> self . assertEqual ( impexpd . _VerifyListening ( socket . AF_INET6 , "<STR_LIT>" , <NUM_LIT> ) , <EOL> ( "<STR_LIT>" , <NUM_LIT> ) ) <EOL> self . assertEqual ( impexpd . _VerifyListening ( socket . AF_INET6 , <EOL> "<STR_LIT>" , <NUM_LIT> ) , <EOL> ( "<STR_LIT>" , <NUM_LIT> ) ) <EOL> def testError ( self ) : <EOL> for family in [ socket . AF_UNIX , socket . AF_INET , socket . AF_INET6 ] : <EOL> self . assertRaises ( errors . GenericError , impexpd . _VerifyListening , <EOL> family , "<STR_LIT>" , <NUM_LIT> ) <EOL> self . assertRaises ( errors . GenericError , impexpd . _VerifyListening , <EOL> family , "<STR_LIT>" , <NUM_LIT> ) <EOL> for family in [ socket . AF_UNIX , socket . AF_INET6 ] : <EOL> self . assertRaises ( errors . GenericError , impexpd . _VerifyListening , <EOL> family , "<STR_LIT>" , <NUM_LIT> ) <EOL> self . assertRaises ( errors . GenericError , impexpd . _VerifyListening , <EOL> family , "<STR_LIT>" , <NUM_LIT> ) <EOL> self . assertRaises ( errors . GenericError , impexpd . _VerifyListening , <EOL> family , "<STR_LIT>" , <NUM_LIT> ) <EOL> for family in [ socket . AF_UNIX , socket . AF_INET ] : <EOL> self . assertRaises ( errors . GenericError , impexpd . _VerifyListening , <EOL> family , "<STR_LIT>" , <NUM_LIT> ) <EOL> class TestCalcThroughput ( unittest . TestCase ) : <EOL> def test ( self ) : <EOL> self . assertEqual ( impexpd . _CalcThroughput ( [ ] ) , None ) <EOL> self . assertEqual ( impexpd . _CalcThroughput ( [ ( <NUM_LIT:0> , <NUM_LIT:0> ) ] ) , None ) <EOL> samples = [ <EOL> ( <NUM_LIT:0.0> , <NUM_LIT:0.0> ) , <EOL> ( <NUM_LIT> , <NUM_LIT> ) , <EOL> ] <EOL> self . assertAlmostEqual ( impexpd . _CalcThroughput ( samples ) , <NUM_LIT> , <NUM_LIT:3> ) <EOL> samples = [ <EOL> ( <NUM_LIT> , <NUM_LIT> ) , <EOL> ( <NUM_LIT> , <NUM_LIT> ) , <EOL> ( <NUM_LIT> , <NUM_LIT> ) , <EOL> ] <EOL> self . assertAlmostEqual ( impexpd . _CalcThroughput ( samples ) , <NUM_LIT> , <NUM_LIT:3> ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> testutils . GanetiTestProgram ( ) </s>
