<s> from __future__ import absolute_import , unicode_literals <EOL> import mock <EOL> from django . test import TestCase <EOL> from drivers import DriverFactory <EOL> from physical . tests import factory as factory_physical <EOL> from logical . tests import factory as factory_logical <EOL> from logical . models import Database <EOL> from . . mongodb import MongoDB <EOL> class AbstractTestDriverMongo ( TestCase ) : <EOL> def setUp ( self ) : <EOL> self . databaseinfra = factory_physical . DatabaseInfraFactory ( ) <EOL> self . instance = factory_physical . InstanceFactory ( <EOL> databaseinfra = self . databaseinfra ) <EOL> self . driver = MongoDB ( databaseinfra = self . databaseinfra ) <EOL> self . _mongo_client = None <EOL> def tearDown ( self ) : <EOL> if not Database . objects . filter ( databaseinfra_id = self . databaseinfra . id ) : <EOL> self . databaseinfra . delete ( ) <EOL> if self . _mongo_client : <EOL> self . _mongo_client . close ( ) <EOL> self . driver = self . databaseinfra = self . _mongo_client = None <EOL> @ property <EOL> def mongo_client ( self ) : <EOL> if self . _mongo_client is None : <EOL> self . _mongo_client = self . driver . __mongo_client__ ( self . instance ) <EOL> return self . _mongo_client <EOL> class MongoDBEngineTestCase ( AbstractTestDriverMongo ) : <EOL> """<STR_LIT>""" <EOL> def test_mongodb_app_installed ( self ) : <EOL> self . assertTrue ( DriverFactory . is_driver_available ( "<STR_LIT>" ) ) <EOL> def test_instantiate_mongodb_using_engine_factory ( self ) : <EOL> self . assertEqual ( MongoDB , type ( self . driver ) ) <EOL> self . assertEqual ( self . databaseinfra , self . driver . databaseinfra ) <EOL> def test_connection_string ( self ) : <EOL> self . assertEqual ( <EOL> "<STR_LIT>" , self . driver . get_connection ( ) ) <EOL> def test_get_user ( self ) : <EOL> self . assertEqual ( self . databaseinfra . user , self . driver . get_user ( ) ) <EOL> def test_get_password ( self ) : <EOL> self . assertEqual ( <EOL> self . databaseinfra . password , self . driver . get_password ( ) ) <EOL> def test_get_default_port ( self ) : <EOL> self . assertEqual ( <NUM_LIT> , self . driver . default_port ) <EOL> @ mock . patch . object ( MongoDB , '<STR_LIT>' ) <EOL> def test_connection_string_when_in_replica_set ( self , get_replica_name ) : <EOL> self . instance = factory_physical . InstanceFactory ( <EOL> databaseinfra = self . databaseinfra , address = '<STR_LIT>' , port = <NUM_LIT> ) <EOL> get_replica_name . return_value = '<STR_LIT>' <EOL> self . assertEqual ( <EOL> "<STR_LIT>" , self . driver . get_connection ( ) ) <EOL> def test_connection_with_database ( self ) : <EOL> self . database = factory_logical . DatabaseFactory ( <EOL> name = "<STR_LIT>" , databaseinfra = self . databaseinfra ) <EOL> self . assertEqual ( "<STR_LIT>" , <EOL> self . driver . get_connection ( database = self . database ) ) <EOL> @ mock . patch . object ( MongoDB , '<STR_LIT>' ) <EOL> def test_connection_with_database_and_replica ( self , get_replica_name ) : <EOL> self . instance = factory_physical . InstanceFactory ( <EOL> databaseinfra = self . databaseinfra , address = '<STR_LIT>' , port = <NUM_LIT> ) <EOL> get_replica_name . return_value = '<STR_LIT>' <EOL> self . database = factory_logical . DatabaseFactory ( <EOL> name = "<STR_LIT>" , databaseinfra = self . databaseinfra ) <EOL> self . assertEqual ( "<STR_LIT>" , <EOL> self . driver . get_connection ( database = self . database ) ) <EOL> class ManageDatabaseMongoDBTestCase ( AbstractTestDriverMongo ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> super ( ManageDatabaseMongoDBTestCase , self ) . setUp ( ) <EOL> self . database = factory_logical . DatabaseFactory ( <EOL> databaseinfra = self . databaseinfra ) <EOL> self . mongo_client . drop_database ( self . database . name ) <EOL> def tearDown ( self ) : <EOL> if not Database . objects . filter ( databaseinfra_id = self . databaseinfra . id ) : <EOL> self . database . delete ( ) <EOL> super ( ManageDatabaseMongoDBTestCase , self ) . tearDown ( ) <EOL> def test_mongodb_create_database ( self ) : <EOL> self . assertFalse ( <EOL> self . database . name in self . mongo_client . database_names ( ) ) <EOL> self . driver . create_database ( self . database ) <EOL> self . assertTrue ( <EOL> self . database . name in self . mongo_client . database_names ( ) ) <EOL> def test_mongodb_remove_database ( self ) : <EOL> self . driver . create_database ( self . database ) <EOL> self . assertTrue ( <EOL> self . database . name in self . mongo_client . database_names ( ) ) <EOL> self . driver . remove_database ( self . database ) <EOL> self . assertFalse ( <EOL> self . database . name in self . mongo_client . database_names ( ) ) <EOL> class ManageCredentialsMongoDBTestCase ( AbstractTestDriverMongo ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> super ( ManageCredentialsMongoDBTestCase , self ) . setUp ( ) <EOL> self . database = factory_logical . DatabaseFactory ( <EOL> databaseinfra = self . databaseinfra ) <EOL> self . credential = factory_logical . CredentialFactory ( <EOL> database = self . database ) <EOL> self . driver . create_database ( self . database ) <EOL> def tearDown ( self ) : <EOL> self . driver . remove_database ( self . database ) <EOL> self . credential . delete ( ) <EOL> self . database . delete ( ) <EOL> super ( ManageCredentialsMongoDBTestCase , self ) . tearDown ( ) <EOL> def __find_user__ ( self , credential ) : <EOL> v = self . mongo_client . server_info ( ) [ '<STR_LIT:version>' ] <EOL> if v < '<STR_LIT>' : <EOL> return getattr ( self . mongo_client , credential . database . name ) . system . users . find_one ( { "<STR_LIT:user>" : credential . user } ) <EOL> else : <EOL> return getattr ( self . mongo_client , "<STR_LIT>" ) . system . users . find_one ( { "<STR_LIT:user>" : credential . user , "<STR_LIT>" : credential . database . name } ) <EOL> def test_mongodb_create_credential ( self ) : <EOL> self . assertIsNone ( self . __find_user__ ( <EOL> self . credential ) , "<STR_LIT>" % self . credential ) <EOL> self . driver . create_user ( self . credential ) <EOL> user = self . __find_user__ ( self . credential ) <EOL> self . assertIsNotNone ( user ) <EOL> self . assertEquals ( self . credential . user , user [ '<STR_LIT:user>' ] ) <EOL> self . driver . remove_user ( self . credential ) <EOL> def test_mongodb_remove_credential ( self ) : <EOL> self . driver . create_user ( self . credential ) <EOL> self . assertIsNotNone ( self . __find_user__ ( <EOL> self . credential ) , "<STR_LIT>" % self . credential ) <EOL> self . driver . remove_user ( self . credential ) <EOL> self . assertIsNone ( self . __find_user__ ( self . credential ) ) </s>
