<s> from unittest import TestCase <EOL> import mock <EOL> from json import loads <EOL> from redis import Redis <EOL> from preggy import expect <EOL> from thumbor . detectors . queued_detector import QueuedDetector <EOL> class QueuedDetectorTestCase ( TestCase ) : <EOL> def setUp ( self , * args , ** kw ) : <EOL> super ( QueuedDetectorTestCase , self ) . setUp ( * args , ** kw ) <EOL> self . redis = Redis ( <EOL> host = '<STR_LIT>' , <EOL> port = <NUM_LIT> , <EOL> db = <NUM_LIT:0> , <EOL> password = '<STR_LIT>' , <EOL> ) <EOL> self . redis . delete ( '<STR_LIT>' ) <EOL> self . redis . delete ( '<STR_LIT>' ) <EOL> QueuedDetector . queue = None <EOL> def test_detector_sends_to_queue ( self ) : <EOL> ctx = mock . Mock ( <EOL> config = mock . Mock ( <EOL> REDIS_QUEUE_SERVER_HOST = '<STR_LIT>' , <EOL> REDIS_QUEUE_SERVER_PORT = <NUM_LIT> , <EOL> REDIS_QUEUE_SERVER_DB = <NUM_LIT:0> , <EOL> REDIS_QUEUE_SERVER_PASSWORD = '<STR_LIT>' , <EOL> ) , <EOL> request = mock . Mock ( <EOL> image_url = '<STR_LIT>' , <EOL> detection_error = False , <EOL> ) , <EOL> ) <EOL> detector = QueuedDetector ( ctx , <NUM_LIT:1> , [ ] ) <EOL> expect ( detector ) . not_to_be_null ( ) <EOL> def validate ( data ) : <EOL> expect ( data ) . to_be_empty ( ) <EOL> detector . detect ( validate ) <EOL> expect ( ctx . request . detection_error ) . to_be_false ( ) <EOL> result = self . redis . get ( '<STR_LIT>' ) <EOL> expect ( result ) . to_equal ( '<STR_LIT:1>' ) <EOL> expected_payload = { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:args>" : [ "<STR_LIT:all>" , "<STR_LIT>" , "<STR_LIT>" ] , <EOL> "<STR_LIT:class>" : "<STR_LIT>" , <EOL> "<STR_LIT:key>" : "<STR_LIT>" <EOL> } <EOL> result = self . redis . lpop ( '<STR_LIT>' ) <EOL> expect ( loads ( result ) ) . to_be_like ( expected_payload ) <EOL> def test_detector_fails_properly ( self ) : <EOL> ctx = mock . Mock ( <EOL> config = mock . Mock ( <EOL> REDIS_QUEUE_SERVER_HOST = '<STR_LIT>' , <EOL> REDIS_QUEUE_SERVER_PORT = <NUM_LIT> , <EOL> REDIS_QUEUE_SERVER_DB = <NUM_LIT:0> , <EOL> REDIS_QUEUE_SERVER_PASSWORD = '<STR_LIT>' , <EOL> ) , <EOL> request = mock . Mock ( <EOL> image_url = '<STR_LIT>' , <EOL> detection_error = False , <EOL> ) , <EOL> ) <EOL> detector = QueuedDetector ( ctx , <NUM_LIT:1> , [ ] ) <EOL> expect ( detector ) . not_to_be_null ( ) <EOL> def validate ( data ) : <EOL> expect ( data ) . to_be_empty ( ) <EOL> detector . detect ( validate ) <EOL> expect ( ctx . request . detection_error ) . to_be_true ( ) <EOL> expect ( detector . queue ) . to_be_null ( ) <EOL> def test_detector_can_detect_twice ( self ) : <EOL> ctx = mock . Mock ( <EOL> config = mock . Mock ( <EOL> REDIS_QUEUE_SERVER_HOST = '<STR_LIT>' , <EOL> REDIS_QUEUE_SERVER_PORT = <NUM_LIT> , <EOL> REDIS_QUEUE_SERVER_DB = <NUM_LIT:0> , <EOL> REDIS_QUEUE_SERVER_PASSWORD = '<STR_LIT>' , <EOL> ) , <EOL> request = mock . Mock ( <EOL> image_url = '<STR_LIT>' , <EOL> detection_error = False , <EOL> ) , <EOL> ) <EOL> detector = QueuedDetector ( ctx , <NUM_LIT:1> , [ ] ) <EOL> expect ( detector ) . not_to_be_null ( ) <EOL> def validate ( data ) : <EOL> expect ( data ) . to_be_empty ( ) <EOL> detector . detect ( validate ) <EOL> expect ( ctx . request . detection_error ) . to_be_false ( ) <EOL> expect ( detector . queue ) . not_to_be_null ( ) <EOL> detector . detect ( validate ) <EOL> expect ( detector . queue ) . not_to_be_null ( ) </s>
