<s> from __future__ import absolute_import <EOL> import unittest <EOL> from pyes . tests import ESTestCase <EOL> from pyes . rivers import CouchDBRiver , RabbitMQRiver , TwitterRiver <EOL> class RiversTestCase ( ESTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( RiversTestCase , self ) . setUp ( ) <EOL> def testCreateCouchDBRiver ( self ) : <EOL> """<STR_LIT>""" <EOL> test_river = CouchDBRiver ( index_name = '<STR_LIT>' , index_type = '<STR_LIT>' ) <EOL> result = self . conn . create_river ( test_river , river_name = '<STR_LIT>' ) <EOL> self . assertResultContains ( result , { '<STR_LIT>' : True } ) <EOL> def testDeleteCouchDBRiver ( self ) : <EOL> """<STR_LIT>""" <EOL> test_river = CouchDBRiver ( index_name = '<STR_LIT>' , index_type = '<STR_LIT>' ) <EOL> result = self . conn . delete_river ( test_river , river_name = '<STR_LIT>' ) <EOL> self . assertResultContains ( result , { '<STR_LIT>' : True } ) <EOL> def testCreateRabbitMQRiver ( self ) : <EOL> """<STR_LIT>""" <EOL> test_river = RabbitMQRiver ( index_name = '<STR_LIT>' , index_type = '<STR_LIT>' ) <EOL> result = self . conn . create_river ( test_river , river_name = '<STR_LIT>' ) <EOL> self . assertResultContains ( result , { '<STR_LIT>' : True } ) <EOL> def testDeleteRabbitMQRiver ( self ) : <EOL> """<STR_LIT>""" <EOL> test_river = RabbitMQRiver ( index_name = '<STR_LIT>' , index_type = '<STR_LIT>' ) <EOL> result = self . conn . create_river ( test_river , river_name = '<STR_LIT>' ) <EOL> result = self . conn . delete_river ( test_river , river_name = '<STR_LIT>' ) <EOL> self . assertResultContains ( result , { '<STR_LIT>' : True } ) <EOL> def testCreateTwitterRiver ( self ) : <EOL> """<STR_LIT>""" <EOL> test_river = TwitterRiver ( '<STR_LIT:test>' , '<STR_LIT:test>' , index_name = '<STR_LIT>' , index_type = '<STR_LIT:status>' ) <EOL> result = self . conn . create_river ( test_river , river_name = '<STR_LIT>' ) <EOL> self . assertResultContains ( result , { '<STR_LIT>' : True } ) <EOL> def testDeleteTwitterRiver ( self ) : <EOL> """<STR_LIT>""" <EOL> test_river = TwitterRiver ( '<STR_LIT:test>' , '<STR_LIT:test>' , index_name = '<STR_LIT>' , index_type = '<STR_LIT:status>' ) <EOL> result = self . conn . create_river ( test_river , river_name = '<STR_LIT>' ) <EOL> result = self . conn . delete_river ( test_river , river_name = '<STR_LIT>' ) <EOL> self . assertResultContains ( result , { '<STR_LIT>' : True } ) <EOL> def testCreateTwitterRiverOAuth ( self ) : <EOL> test_river = TwitterRiver ( '<STR_LIT:test>' , '<STR_LIT:test>' , index_name = '<STR_LIT>' , index_type = '<STR_LIT>' , <EOL> consumer_key = "<STR_LIT>" , <EOL> consumer_secret = "<STR_LIT>" , <EOL> access_token = "<STR_LIT>" , <EOL> access_token_secret = "<STR_LIT>" , <EOL> ) <EOL> result = self . conn . create_river ( test_river , river_name = '<STR_LIT>' ) <EOL> self . assertResultContains ( result , { '<STR_LIT>' : True } ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> unittest . main ( ) </s>
