<s> """<STR_LIT>""" <EOL> import os <EOL> import sys <EOL> import unittest <EOL> from ganeti import constants <EOL> from ganeti import errors <EOL> from ganeti import utils <EOL> from ganeti import masterd <EOL> from ganeti . masterd . instance import ImportExportTimeouts , _DiskImportExportBase , ComputeRemoteExportHandshake , CheckRemoteExportHandshake , ComputeRemoteImportDiskInfo , CheckRemoteExportDiskInfo , FormatProgress <EOL> import testutils <EOL> class TestMisc ( unittest . TestCase ) : <EOL> def testTimeouts ( self ) : <EOL> tmo = ImportExportTimeouts ( <NUM_LIT:0> ) <EOL> self . assertEqual ( tmo . connect , <NUM_LIT:0> ) <EOL> self . assertEqual ( tmo . listen , ImportExportTimeouts . DEFAULT_LISTEN_TIMEOUT ) <EOL> self . assertEqual ( tmo . ready , ImportExportTimeouts . DEFAULT_READY_TIMEOUT ) <EOL> self . assertEqual ( tmo . error , ImportExportTimeouts . DEFAULT_ERROR_TIMEOUT ) <EOL> self . assertEqual ( tmo . progress , <EOL> ImportExportTimeouts . DEFAULT_PROGRESS_INTERVAL ) <EOL> tmo = ImportExportTimeouts ( <NUM_LIT> ) <EOL> self . assertEqual ( tmo . connect , <NUM_LIT> ) <EOL> tmo = ImportExportTimeouts ( <NUM_LIT:1> , listen = <NUM_LIT:2> , error = <NUM_LIT:3> , ready = <NUM_LIT:4> , progress = <NUM_LIT:5> ) <EOL> self . assertEqual ( tmo . connect , <NUM_LIT:1> ) <EOL> self . assertEqual ( tmo . listen , <NUM_LIT:2> ) <EOL> self . assertEqual ( tmo . error , <NUM_LIT:3> ) <EOL> self . assertEqual ( tmo . ready , <NUM_LIT:4> ) <EOL> self . assertEqual ( tmo . progress , <NUM_LIT:5> ) <EOL> def testTimeoutExpired ( self ) : <EOL> self . assert_ ( utils . TimeoutExpired ( <NUM_LIT:100> , <NUM_LIT> , _time_fn = lambda : <NUM_LIT> ) ) <EOL> self . assertFalse ( utils . TimeoutExpired ( <NUM_LIT:100> , <NUM_LIT> , _time_fn = lambda : <NUM_LIT:0> ) ) <EOL> self . assertFalse ( utils . TimeoutExpired ( <NUM_LIT:100> , <NUM_LIT> , _time_fn = lambda : <NUM_LIT:100> ) ) <EOL> self . assertFalse ( utils . TimeoutExpired ( <NUM_LIT:100> , <NUM_LIT> , _time_fn = lambda : <NUM_LIT> ) ) <EOL> def testDiskImportExportBaseDirect ( self ) : <EOL> self . assertRaises ( AssertionError , _DiskImportExportBase , <EOL> None , None , None , None , None , None , None ) <EOL> class TestRieHandshake ( unittest . TestCase ) : <EOL> def test ( self ) : <EOL> cds = "<STR_LIT>" <EOL> hs = ComputeRemoteExportHandshake ( cds ) <EOL> self . assertEqual ( len ( hs ) , <NUM_LIT:3> ) <EOL> self . assertEqual ( hs [ <NUM_LIT:0> ] , constants . RIE_VERSION ) <EOL> self . assertEqual ( CheckRemoteExportHandshake ( cds , hs ) , None ) <EOL> def testCheckErrors ( self ) : <EOL> self . assert_ ( CheckRemoteExportHandshake ( None , None ) ) <EOL> self . assert_ ( CheckRemoteExportHandshake ( "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> self . assert_ ( CheckRemoteExportHandshake ( "<STR_LIT>" , ( "<STR_LIT>" , "<STR_LIT:foo>" ) ) ) <EOL> def testCheckWrongHash ( self ) : <EOL> cds = "<STR_LIT>" <EOL> self . assert_ ( CheckRemoteExportHandshake ( cds , ( <NUM_LIT:0> , "<STR_LIT>" , "<STR_LIT>" ) ) ) <EOL> def testCheckWrongVersion ( self ) : <EOL> version = <NUM_LIT> <EOL> self . assertNotEqual ( version , constants . RIE_VERSION ) <EOL> cds = "<STR_LIT>" <EOL> salt = "<STR_LIT>" <EOL> msg = "<STR_LIT>" % ( version , constants . RIE_HANDSHAKE ) <EOL> hs = ( version , utils . Sha1Hmac ( cds , msg , salt = salt ) , salt ) <EOL> self . assert_ ( CheckRemoteExportHandshake ( cds , hs ) ) <EOL> class TestRieDiskInfo ( unittest . TestCase ) : <EOL> def test ( self ) : <EOL> cds = "<STR_LIT>" <EOL> salt = "<STR_LIT>" <EOL> di = ComputeRemoteImportDiskInfo ( cds , salt , <NUM_LIT:0> , "<STR_LIT>" , <NUM_LIT> , "<STR_LIT>" ) <EOL> self . assertEqual ( CheckRemoteExportDiskInfo ( cds , <NUM_LIT:0> , di ) , <EOL> ( "<STR_LIT>" , <NUM_LIT> , "<STR_LIT>" ) ) <EOL> for i in range ( <NUM_LIT:1> , <NUM_LIT:100> ) : <EOL> self . assertRaises ( errors . GenericError , CheckRemoteExportDiskInfo , <EOL> cds , i , di ) <EOL> def testInvalidHostPort ( self ) : <EOL> cds = "<STR_LIT>" <EOL> salt = "<STR_LIT>" <EOL> for host in [ "<STR_LIT:U+002C>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:!>" , "<STR_LIT:#>" , "<STR_LIT:\\>" ] : <EOL> di = ComputeRemoteImportDiskInfo ( cds , salt , <NUM_LIT:0> , host , <NUM_LIT> , "<STR_LIT>" ) <EOL> self . assertRaises ( errors . OpPrereqError , <EOL> CheckRemoteExportDiskInfo , cds , <NUM_LIT:0> , di ) <EOL> for port in [ - <NUM_LIT:1> , <NUM_LIT> , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] : <EOL> di = ComputeRemoteImportDiskInfo ( cds , salt , <NUM_LIT:0> , "<STR_LIT:localhost>" , port , "<STR_LIT>" ) <EOL> self . assertRaises ( errors . OpPrereqError , <EOL> CheckRemoteExportDiskInfo , cds , <NUM_LIT:0> , di ) <EOL> def testCheckErrors ( self ) : <EOL> cds = "<STR_LIT>" <EOL> self . assertRaises ( errors . GenericError , CheckRemoteExportDiskInfo , <EOL> cds , <NUM_LIT:0> , "<STR_LIT>" ) <EOL> self . assertRaises ( errors . GenericError , CheckRemoteExportDiskInfo , <EOL> cds , <NUM_LIT:0> , ( ) ) <EOL> self . assertRaises ( errors . GenericError , CheckRemoteExportDiskInfo , <EOL> cds , <NUM_LIT:0> , ( "<STR_LIT>" , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> , <NUM_LIT:5> ) ) <EOL> self . assertRaises ( errors . GenericError , CheckRemoteExportDiskInfo , <EOL> cds , <NUM_LIT:0> , ( "<STR_LIT>" , <NUM_LIT> , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> self . assertRaises ( errors . GenericError , CheckRemoteExportDiskInfo , <EOL> cds , <NUM_LIT:0> , ( "<STR_LIT:host>" , <NUM_LIT:0> , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> self . assertRaises ( errors . GenericError , CheckRemoteExportDiskInfo , <EOL> cds , <NUM_LIT:0> , ( "<STR_LIT:host>" , <NUM_LIT> , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> self . assertRaises ( errors . GenericError , CheckRemoteExportDiskInfo , <EOL> cds , <NUM_LIT:0> , ( "<STR_LIT>" , <NUM_LIT> , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> class TestFormatProgress ( unittest . TestCase ) : <EOL> def test ( self ) : <EOL> FormatProgress ( ( <NUM_LIT:0> , <NUM_LIT:0> , None , None ) ) <EOL> FormatProgress ( ( <NUM_LIT:100> , <NUM_LIT> , <NUM_LIT:30> , None ) ) <EOL> FormatProgress ( ( <NUM_LIT:100> , <NUM_LIT> , <NUM_LIT:30> , <NUM_LIT> ) ) <EOL> self . assertEqual ( FormatProgress ( ( <NUM_LIT> , <NUM_LIT:12> , <NUM_LIT:30> , None ) ) , <EOL> "<STR_LIT>" ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> testutils . GanetiTestProgram ( ) </s>
