#NEWLINE# def test_storage_celery_save( self ) : #NEWLINE# #INDENT# '\n        Make sure it actually works when using Celery as a task queue\n        ' #NEWLINE# storage = QueuedStorage ( local = 'django.core.files.storage.FileSystemStorage' , remote = 'django.core.files.storage.FileSystemStorage' , local_options = dict ( location = self . local_dir ) , remote_options = dict ( location = self . remote_dir ) ) #NEWLINE# field = models . TestModel . _meta . get_field ( 'testfile' ) #NEWLINE# field . storage = storage #NEWLINE# obj = models . TestModel ( testfile = File ( self . test_file ) ) #NEWLINE# obj . save ( ) #NEWLINE# self . assertTrue ( obj . testfile . storage . result . get ( ) ) #NEWLINE# self . assertTrue ( path . isfile ( path . join ( self . local_dir , obj . testfile . name ) ) ) #NEWLINE# self . assertTrue ( path . isfile ( path . join ( self . remote_dir , obj . testfile . name ) ) , 'Remote file is not available.' ) #NEWLINE# self . assertFalse ( storage . using_local ( obj . testfile . name ) ) #NEWLINE# self . assertTrue ( storage . using_remote ( obj . testfile . name ) ) #NEWLINE# self . assertEqual ( self . test_file_name , storage . get_valid_name ( self . test_file_name ) ) #NEWLINE# self . assertEqual ( self . test_file_name , storage . get_available_name ( self . test_file_name ) ) #NEWLINE# subdir_path = os . path . join ( 'test' , self . test_file_name ) #NEWLINE# self . assertTrue ( storage . exists ( subdir_path ) ) #NEWLINE# self . assertEqual ( storage . path ( self . test_file_name ) , path . join ( self . local_dir , self . test_file_name ) ) #NEWLINE# self . assertEqual ( storage . listdir ( 'test' ) [ 1 ] , [ self . test_file_name ] ) #NEWLINE# self . assertEqual ( storage . size ( subdir_path ) , os . stat ( self . test_file_path ) . st_size ) #NEWLINE# self . assertEqual ( storage . url ( self . test_file_name ) , self . test_file_name ) #NEWLINE# self . assertIsInstance ( storage . accessed_time ( subdir_path ) , datetime ) #NEWLINE# self . assertIsInstance ( storage . created_time ( subdir_path ) , datetime ) #NEWLINE# self . assertIsInstance ( storage . modified_time ( subdir_path ) , datetime ) #NEWLINE# subdir_name = 'queued_storage_2.txt' #NEWLINE# testfile = storage . open ( subdir_name , 'w' ) #NEWLINE# try : #NEWLINE# #INDENT# testfile . write ( 'test' ) #NEWLINE# #UNINDENT# finally : #NEWLINE# #INDENT# testfile . close ( ) #NEWLINE# #UNINDENT# self . assertTrue ( storage . exists ( subdir_name ) ) #NEWLINE# storage . delete ( subdir_name ) #NEWLINE# self . assertFalse ( storage . exists ( subdir_name ) )