mirror of
https://github.com/bigchaindb/bigchaindb.git
synced 2024-10-13 13:34:05 +00:00
2d1f670eec
Solution: Remove un-necessary code. Deployment of MongoDB with or without replicaSet should be the responsibility of MongoDB admin which can and cannot be a BigchainDB node operator. As far as BigchainDB is concerned replicaset, if provided in bigchaindb configs, should be used to establish connection with MongoDB.
111 lines
3.4 KiB
Python
111 lines
3.4 KiB
Python
# Copyright BigchainDB GmbH and BigchainDB contributors
|
|
# SPDX-License-Identifier: (Apache-2.0 AND CC-BY-4.0)
|
|
# Code is Apache-2.0 and docs are CC-BY-4.0
|
|
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
import pymongo
|
|
from pymongo import MongoClient
|
|
|
|
|
|
pytestmark = [pytest.mark.bdb, pytest.mark.tendermint]
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_cmd_line_opts():
|
|
return {'argv': ['mongod', '--dbpath=/data'],
|
|
'ok': 1.0,
|
|
'parsed': {'replication': {'replSet': None},
|
|
'storage': {'dbPath': '/data'}}}
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_config_opts():
|
|
return {'argv': ['mongod', '--dbpath=/data'],
|
|
'ok': 1.0,
|
|
'parsed': {'replication': {'replSetName': None},
|
|
'storage': {'dbPath': '/data'}}}
|
|
|
|
|
|
@pytest.fixture
|
|
def mongodb_connection():
|
|
import bigchaindb
|
|
return MongoClient(host=bigchaindb.config['database']['host'],
|
|
port=bigchaindb.config['database']['port'])
|
|
|
|
|
|
def test_get_connection_returns_the_correct_instance(db_host, db_port):
|
|
from bigchaindb.backend import connect
|
|
from bigchaindb.backend.connection import Connection
|
|
from bigchaindb.backend.localmongodb.connection import LocalMongoDBConnection
|
|
|
|
config = {
|
|
'backend': 'localmongodb',
|
|
'host': db_host,
|
|
'port': db_port,
|
|
'name': 'test',
|
|
'replicaset': None,
|
|
}
|
|
|
|
conn = connect(**config)
|
|
assert isinstance(conn, Connection)
|
|
assert isinstance(conn, LocalMongoDBConnection)
|
|
assert conn.conn._topology_settings.replica_set_name == config['replicaset']
|
|
|
|
|
|
@mock.patch('pymongo.MongoClient.__init__')
|
|
def test_connection_error(mock_client):
|
|
from bigchaindb.backend import connect
|
|
from bigchaindb.backend.exceptions import ConnectionError
|
|
|
|
# force the driver to throw ConnectionFailure
|
|
# the mock on time.sleep is to prevent the actual sleep when running
|
|
# the tests
|
|
mock_client.side_effect = pymongo.errors.ConnectionFailure()
|
|
|
|
with pytest.raises(ConnectionError):
|
|
conn = connect()
|
|
conn.db
|
|
|
|
assert mock_client.call_count == 3
|
|
|
|
|
|
def test_connection_run_errors():
|
|
from bigchaindb.backend import connect
|
|
from bigchaindb.backend.exceptions import (DuplicateKeyError,
|
|
OperationError,
|
|
ConnectionError)
|
|
|
|
conn = connect()
|
|
|
|
query = mock.Mock()
|
|
query.run.side_effect = pymongo.errors.AutoReconnect('foo')
|
|
with pytest.raises(ConnectionError):
|
|
conn.run(query)
|
|
assert query.run.call_count == 2
|
|
|
|
query = mock.Mock()
|
|
query.run.side_effect = pymongo.errors.DuplicateKeyError('foo')
|
|
with pytest.raises(DuplicateKeyError):
|
|
conn.run(query)
|
|
assert query.run.call_count == 1
|
|
|
|
query = mock.Mock()
|
|
query.run.side_effect = pymongo.errors.OperationFailure('foo')
|
|
with pytest.raises(OperationError):
|
|
conn.run(query)
|
|
assert query.run.call_count == 1
|
|
|
|
|
|
@mock.patch('pymongo.database.Database.authenticate')
|
|
def test_connection_with_credentials(mock_authenticate):
|
|
import bigchaindb
|
|
from bigchaindb.backend.localmongodb.connection import LocalMongoDBConnection
|
|
conn = LocalMongoDBConnection(host=bigchaindb.config['database']['host'],
|
|
port=bigchaindb.config['database']['port'],
|
|
login='theplague',
|
|
password='secret')
|
|
conn.connect()
|
|
assert mock_authenticate.call_count == 1
|