<s> import os <EOL> from . test_utils import files_server <EOL> from pulsar . client . action_mapper import RemoteTransferAction <EOL> def test_write_to_file ( ) : <EOL> with files_server ( ) as ( server , directory ) : <EOL> from_path = os . path . join ( directory , "<STR_LIT>" ) <EOL> open ( from_path , "<STR_LIT:wb>" ) . write ( b"<STR_LIT>" ) <EOL> to_path = os . path . join ( directory , "<STR_LIT>" ) <EOL> url = server . application_url + "<STR_LIT>" % from_path <EOL> RemoteTransferAction ( to_path , url = url ) . write_to_path ( to_path ) <EOL> assert open ( to_path , "<STR_LIT:rb>" ) . read ( ) == b"<STR_LIT>" <EOL> def test_write_from_file ( ) : <EOL> with files_server ( ) as ( server , directory ) : <EOL> from_path = os . path . join ( directory , "<STR_LIT>" ) <EOL> open ( from_path , "<STR_LIT:wb>" ) . write ( b"<STR_LIT>" ) <EOL> to_path = os . path . join ( directory , "<STR_LIT>" ) <EOL> url = server . application_url + "<STR_LIT>" % to_path <EOL> RemoteTransferAction ( to_path , url = url ) . write_from_path ( from_path ) <EOL> posted_contents = open ( to_path , "<STR_LIT:rb>" ) . read ( ) <EOL> assert posted_contents == b"<STR_LIT>" , posted_contents </s>
