<s> from os import makedirs <EOL> from os . path import join , dirname , exists <EOL> from string import Template <EOL> from galaxy . util . bunch import Bunch <EOL> from galaxy . objectstore import build_object_store_from_config <EOL> from . test_utils import TempDirectoryTestCase <EOL> from . test_utils import skip <EOL> class MockDataset ( object ) : <EOL> def __init__ ( self , id ) : <EOL> self . id = id <EOL> self . object_store_id = None <EOL> class PulsarObjectStoreTest ( TempDirectoryTestCase ) : <EOL> def __write ( self , contents , name ) : <EOL> path = join ( self . temp_directory , name ) <EOL> directory = dirname ( path ) <EOL> if not exists ( directory ) : <EOL> makedirs ( directory ) <EOL> open ( path , "<STR_LIT:wb>" ) . write ( contents ) <EOL> return path <EOL> @ skip <EOL> def test_pulsar_objectstore ( self ) : <EOL> object_store_config_file = join ( self . temp_directory , "<STR_LIT>" ) <EOL> with open ( object_store_config_file , "<STR_LIT:w>" ) as configf : <EOL> config_template = Template ( """<STR_LIT>""" ) <EOL> config_contents = config_template . safe_substitute ( temp_directory = self . temp_directory ) <EOL> configf . write ( config_contents ) <EOL> app_conf = dict ( <EOL> object_store_config_file = object_store_config_file , <EOL> private_token = "<STR_LIT>" , <EOL> ) <EOL> from . test_utils import test_pulsar_server <EOL> with test_pulsar_server ( app_conf = app_conf ) as server : <EOL> url = server . application_url <EOL> proxy_object_store_config_file = join ( self . temp_directory , "<STR_LIT>" ) <EOL> with open ( proxy_object_store_config_file , "<STR_LIT:w>" ) as configf : <EOL> config_template = Template ( """<STR_LIT>""" ) <EOL> contents = config_template . safe_substitute ( url = url ) <EOL> configf . write ( contents ) <EOL> config = Bunch ( object_store_config_file = proxy_object_store_config_file ) <EOL> object_store = build_object_store_from_config ( config = config ) <EOL> absent_dataset = MockDataset ( <NUM_LIT:1> ) <EOL> assert not object_store . exists ( absent_dataset ) <EOL> empty_dataset = MockDataset ( <NUM_LIT:2> ) <EOL> self . __write ( b"<STR_LIT>" , "<STR_LIT>" ) <EOL> assert object_store . exists ( empty_dataset ) <EOL> assert object_store . empty ( empty_dataset ) <EOL> hello_world_dataset = MockDataset ( <NUM_LIT:3> ) <EOL> self . __write ( b"<STR_LIT>" , "<STR_LIT>" ) <EOL> assert object_store . exists ( hello_world_dataset ) <EOL> assert not object_store . empty ( hello_world_dataset ) <EOL> data = object_store . get_data ( hello_world_dataset ) <EOL> assert data == "<STR_LIT>" <EOL> data = object_store . get_data ( hello_world_dataset , start = <NUM_LIT:1> , count = <NUM_LIT:6> ) <EOL> assert data == "<STR_LIT>" <EOL> assert object_store . size ( absent_dataset ) == <NUM_LIT:0> <EOL> assert object_store . size ( empty_dataset ) == <NUM_LIT:0> <EOL> assert object_store . size ( hello_world_dataset ) > <NUM_LIT:0> <EOL> percent_store_used = object_store . get_store_usage_percent ( ) <EOL> assert percent_store_used > <NUM_LIT:0.0> <EOL> assert percent_store_used < <NUM_LIT> <EOL> output_dataset = MockDataset ( <NUM_LIT:4> ) <EOL> output_real_path = join ( self . temp_directory , "<STR_LIT>" , "<STR_LIT>" ) <EOL> assert not exists ( output_real_path ) <EOL> output_working_path = self . __write ( b"<STR_LIT>" , "<STR_LIT>" ) <EOL> object_store . update_from_file ( output_dataset , file_name = output_working_path , create = True ) <EOL> assert exists ( output_real_path ) <EOL> to_delete_dataset = MockDataset ( <NUM_LIT:5> ) <EOL> to_delete_real_path = self . __write ( b"<STR_LIT>" , "<STR_LIT>" ) <EOL> assert object_store . exists ( to_delete_dataset ) <EOL> assert object_store . delete ( to_delete_dataset ) <EOL> assert not object_store . exists ( to_delete_dataset ) <EOL> assert not exists ( to_delete_real_path ) <EOL> complex_contents_dataset = MockDataset ( <NUM_LIT:6> ) <EOL> complex_content = b'<STR_LIT>' <EOL> self . __write ( complex_content , "<STR_LIT>" ) <EOL> assert object_store . exists ( complex_contents_dataset ) <EOL> data = object_store . get_data ( complex_contents_dataset ) == complex_content </s>
