<s> import rpyc <EOL> import time <EOL> import threading <EOL> import sys <EOL> import os <EOL> import unittest <EOL> from rpyc . utils . server import ThreadedServer <EOL> from rpyc import SlaveService <EOL> from plumbum import SshMachine <EOL> class Test_Ssh ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> if sys . platform == "<STR_LIT:win32>" : <EOL> self . server = None <EOL> os . environ [ "<STR_LIT>" ] = os . path . expanduser ( "<STR_LIT>" ) <EOL> self . remote_machine = SshMachine ( "<STR_LIT:localhost>" ) <EOL> else : <EOL> self . server = ThreadedServer ( SlaveService , hostname = "<STR_LIT:localhost>" , <EOL> ipv6 = False , port = <NUM_LIT> , auto_register = False ) <EOL> t = threading . Thread ( target = self . server . start ) <EOL> t . setDaemon ( True ) <EOL> t . start ( ) <EOL> time . sleep ( <NUM_LIT:0.5> ) <EOL> self . remote_machine = SshMachine ( "<STR_LIT:localhost>" ) <EOL> def tearDown ( self ) : <EOL> if self . server : <EOL> self . server . close ( ) <EOL> def test_simple ( self ) : <EOL> conn = rpyc . classic . ssh_connect ( self . remote_machine , <NUM_LIT> ) <EOL> print ( "<STR_LIT>" , conn . modules . os . getpid ( ) ) <EOL> conn . modules . sys . stdout . write ( "<STR_LIT>" ) <EOL> conn . modules . sys . stdout . flush ( ) <EOL> def test_connect ( self ) : <EOL> conn2 = rpyc . ssh_connect ( self . remote_machine , <NUM_LIT> , service = SlaveService ) <EOL> conn2 . modules . sys . stdout . write ( "<STR_LIT>" ) <EOL> conn2 . modules . sys . stdout . flush ( ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> unittest . main ( ) </s>
