<s> import os <EOL> import sys <EOL> import unittest <EOL> import urllib2 <EOL> from flexmock import flexmock <EOL> sys . path . append ( os . path . join ( os . path . dirname ( __file__ ) , "<STR_LIT>" ) ) <EOL> from tq_config import TaskQueueConfig <EOL> sys . path . append ( os . path . join ( os . path . dirname ( __file__ ) , "<STR_LIT>" ) ) <EOL> import file_io <EOL> sys . path . append ( os . path . join ( os . path . dirname ( __file__ ) , "<STR_LIT>" ) ) <EOL> from google . appengine . api import api_base_pb <EOL> from google . appengine . api import datastore <EOL> sample_queue_yaml = """<STR_LIT>""" <EOL> sample_queue_yaml2 = """<STR_LIT>""" <EOL> sample_queue_xml = """<STR_LIT>""" <EOL> class FakeConnection ( ) : <EOL> def __init__ ( self ) : <EOL> pass <EOL> def add_header ( self , arg1 , arg2 ) : <EOL> pass <EOL> class FakeResponse ( ) : <EOL> def __init__ ( self ) : <EOL> pass <EOL> def read ( self ) : <EOL> return "<STR_LIT:{}>" <EOL> class TestTaskQueueConfig ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_constructor ( self ) : <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( sample_queue_yaml ) <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( None ) <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( None ) <EOL> tqc = TaskQueueConfig ( TaskQueueConfig . RABBITMQ , <EOL> '<STR_LIT>' ) <EOL> def test_load_queues_from_file ( self ) : <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( sample_queue_yaml ) <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( True ) <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( None ) <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( None ) <EOL> tqc = TaskQueueConfig ( TaskQueueConfig . RABBITMQ , <EOL> '<STR_LIT>' ) <EOL> queue_info = tqc . load_queues_from_file ( '<STR_LIT>' ) <EOL> self . assertEquals ( queue_info , { '<STR_LIT>' : [ { '<STR_LIT:name>' : '<STR_LIT:default>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT:name>' : '<STR_LIT:foo>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] } ) <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( '<STR_LIT>' ) . and_raise ( IOError ) <EOL> tqc = TaskQueueConfig ( TaskQueueConfig . RABBITMQ , <EOL> '<STR_LIT>' ) <EOL> queue_info = tqc . load_queues_from_file ( '<STR_LIT>' ) <EOL> self . assertEquals ( queue_info , { '<STR_LIT>' : [ { '<STR_LIT:name>' : '<STR_LIT:default>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] } ) <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( sample_queue_yaml2 ) <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( None ) <EOL> tqc = TaskQueueConfig ( TaskQueueConfig . RABBITMQ , <EOL> '<STR_LIT>' ) <EOL> queue_info = tqc . load_queues_from_file ( '<STR_LIT>' ) <EOL> self . assertEquals ( queue_info , { '<STR_LIT>' : [ { '<STR_LIT:name>' : '<STR_LIT:foo>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT:name>' : '<STR_LIT:default>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> ] } ) <EOL> def test_load_queues_from_xml_file ( self ) : <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( sample_queue_xml ) <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( False ) . and_return ( True ) <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( None ) <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( None ) <EOL> tqc = TaskQueueConfig ( TaskQueueConfig . RABBITMQ , <EOL> '<STR_LIT>' ) <EOL> queue_info = tqc . load_queues_from_file ( '<STR_LIT>' ) <EOL> self . assertEquals ( queue_info , { '<STR_LIT>' : [ { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT:100>' , '<STR_LIT:name>' : '<STR_LIT:default>' , '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } } , { '<STR_LIT>' : '<STR_LIT:100>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT:100>' , '<STR_LIT:name>' : '<STR_LIT>' , '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } } ] } ) <EOL> def test_load_queues_from_db ( self ) : <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( sample_queue_yaml2 ) <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( None ) <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( None ) <EOL> flexmock ( datastore ) . should_receive ( "<STR_LIT>" ) . and_return ( { TaskQueueConfig . QUEUE_INFO : '<STR_LIT>' } ) <EOL> tqc = TaskQueueConfig ( TaskQueueConfig . RABBITMQ , <EOL> '<STR_LIT>' ) <EOL> queue_info = tqc . load_queues_from_db ( ) <EOL> self . assertEquals ( queue_info , { '<STR_LIT>' : [ { '<STR_LIT:name>' : '<STR_LIT:foo>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> ] } ) <EOL> def test_save_queues_to_db ( self ) : <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( sample_queue_yaml2 ) <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( None ) <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( None ) <EOL> flexmock ( file_io ) . should_receive ( '<STR_LIT>' ) . and_return ( True ) <EOL> flexmock ( datastore ) . should_receive ( "<STR_LIT>" ) . and_return ( ) <EOL> tqc = TaskQueueConfig ( TaskQueueConfig . RABBITMQ , <EOL> '<STR_LIT>' ) <EOL> try : <EOL> queue_info = tqc . save_queues_to_db ( ) <EOL> raise <EOL> except ValueError : <EOL> pass <EOL> queue_info = tqc . load_queues_from_file ( '<STR_LIT>' ) <EOL> queue_info = tqc . save_queues_to_db ( ) <EOL> def test_load_queues ( self ) : <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( sample_queue_yaml2 ) <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( True ) <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( None ) <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( None ) <EOL> flexmock ( datastore ) . should_receive ( "<STR_LIT>" ) . and_return ( { TaskQueueConfig . QUEUE_INFO : '<STR_LIT>' } ) <EOL> tqc = TaskQueueConfig ( TaskQueueConfig . RABBITMQ , <EOL> '<STR_LIT>' ) <EOL> queue_info = tqc . load_queues_from_file ( '<STR_LIT>' ) <EOL> queue_info = tqc . load_queues_from_db ( ) <EOL> def test_create_celery_file ( self ) : <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( sample_queue_yaml2 ) <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( True ) <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( None ) <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( None ) <EOL> flexmock ( datastore ) . should_receive ( "<STR_LIT>" ) . and_return ( { TaskQueueConfig . QUEUE_INFO : '<STR_LIT>' } ) <EOL> tqc = TaskQueueConfig ( TaskQueueConfig . RABBITMQ , <EOL> '<STR_LIT>' ) <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( sample_queue_yaml2 ) <EOL> queue_info = tqc . load_queues_from_file ( '<STR_LIT>' ) <EOL> queue_info = tqc . load_queues_from_db ( ) <EOL> self . assertEquals ( tqc . create_celery_file ( TaskQueueConfig . QUEUE_INFO_DB ) , <EOL> TaskQueueConfig . CELERY_CONFIG_DIR + "<STR_LIT>" + "<STR_LIT>" ) <EOL> self . assertEquals ( tqc . create_celery_file ( TaskQueueConfig . QUEUE_INFO_FILE ) , <EOL> TaskQueueConfig . CELERY_CONFIG_DIR + "<STR_LIT>" + "<STR_LIT>" ) <EOL> def test_create_celery_worker_scripts ( self ) : <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( sample_queue_yaml2 ) <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( None ) <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( None ) <EOL> flexmock ( datastore ) . should_receive ( "<STR_LIT>" ) . and_return ( { TaskQueueConfig . QUEUE_INFO : '<STR_LIT>' } ) <EOL> tqc = TaskQueueConfig ( TaskQueueConfig . RABBITMQ , <EOL> '<STR_LIT>' ) <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( True ) <EOL> queue_info = tqc . load_queues_from_file ( '<STR_LIT>' ) <EOL> queue_info = tqc . load_queues_from_db ( ) <EOL> FILE1 = open ( os . path . dirname ( os . path . realpath ( __file__ ) ) + '<STR_LIT>' , '<STR_LIT:r>' ) <EOL> file1 = FILE1 . read ( ) <EOL> FILE1 . close ( ) <EOL> FILE2 = open ( os . path . dirname ( os . path . realpath ( __file__ ) ) + '<STR_LIT>' , '<STR_LIT:r>' ) <EOL> file2 = FILE2 . read ( ) <EOL> FILE2 . close ( ) <EOL> flexmock ( file_io ) . should_receive ( '<STR_LIT>' ) . and_return ( None ) <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( file1 ) . and_return ( file2 ) <EOL> self . assertEquals ( tqc . create_celery_worker_scripts ( TaskQueueConfig . QUEUE_INFO_DB ) , TaskQueueConfig . CELERY_WORKER_DIR + '<STR_LIT>' ) <EOL> self . assertEquals ( tqc . create_celery_worker_scripts ( TaskQueueConfig . QUEUE_INFO_FILE ) , TaskQueueConfig . CELERY_WORKER_DIR + '<STR_LIT>' ) <EOL> def test_validate_queue_name ( self ) : <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( sample_queue_yaml2 ) <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( None ) <EOL> flexmock ( file_io ) . should_receive ( "<STR_LIT>" ) . and_return ( None ) <EOL> flexmock ( datastore ) . should_receive ( "<STR_LIT>" ) . and_return ( { TaskQueueConfig . QUEUE_INFO : '<STR_LIT>' } ) <EOL> tqc = TaskQueueConfig ( TaskQueueConfig . RABBITMQ , <EOL> '<STR_LIT>' ) <EOL> tqc . validate_queue_name ( "<STR_LIT:hello>" ) <EOL> tqc . validate_queue_name ( "<STR_LIT>" ) <EOL> try : <EOL> tqc . validate_queue_name ( "<STR_LIT>" ) <EOL> raise <EOL> except NameError : <EOL> pass <EOL> try : <EOL> tqc . validate_queue_name ( "<STR_LIT>" ) <EOL> raise <EOL> except NameError : <EOL> pass <EOL> try : <EOL> tqc . validate_queue_name ( "<STR_LIT>" ) <EOL> raise <EOL> except NameError : <EOL> pass <EOL> try : <EOL> tqc . validate_queue_name ( "<STR_LIT>" ) <EOL> raise <EOL> except NameError : <EOL> pass <EOL> try : <EOL> tqc . validate_queue_name ( "<STR_LIT>" ) <EOL> raise <EOL> except NameError : <EOL> pass <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> unittest . main ( ) </s>
