<commit_before># Copyright 2009-2011 10gen, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Test the replica_set_connection module."""

import datetime
import os
import signal
import sys
import socket
import time
import thread
import unittest
sys.path[0:0] = [""]

from nose.plugins.skip import SkipTest

from bson.son import SON
from bson.tz_util import utc
from pymongo import ReadPreference
from pymongo.connection import Connection
from pymongo.replica_set_connection import ReplicaSetConnection
from pymongo.replica_set_connection import _partition_node
from pymongo.database import Database
from pymongo.errors import (AutoReconnect,
                            ConfigurationError,
                            ConnectionFailure,
                            InvalidName,
                            OperationFailure)
from test import version

host = os.environ.get("DB_IP", socket.gethostname())
port = int(os.environ.get("DB_PORT", 27017))
pair = '%s:%d' % (host, port)

class TestReplicaSetConnectionAgainstStandalone(unittest.TestCase):
    """This is a funny beast -- we want to run tests for ReplicaSetConnection
    but only if the database at DB_IP and DB_PORT is a standalone.
    """
    def setUp(self):
        conn = Connection(pair)
        response = conn.admin.command('ismaster')
        if 'setName' in response:
            raise SkipTest()

    def test_connect(self):
        self.assertRaises(ConfigurationError, ReplicaSetConnection,
                          pair, replicaSet='anything',
                          connectTimeoutMS=600)


class TestConnectionReplicaSetBase(unittest.TestCase):
    def setUp(self):
        conn = Connection(pair)
        response = conn.admin.command('ismaster')
        if 'setName' in response:
            self.name = str(response['setName'])
            self.w = len(response['hosts'])
            self.hosts = set([_partition_node(h)
                              for h in response["hosts"]])
            self.arbiters = set([_partition_node(h)
                                 for h in response.get("arbiters", [])])

            repl_set_status = conn.admin.command('replSetGetStatus')
            primary_info = [
                m for m in repl_set_status['members']
                if m['stateStr'] == 'PRIMARY'
            ][0]

            self.primary = _partition_node(primary_info['name'])
        else:
            raise SkipTest()

    def _get_connection(self, **kwargs):
        return ReplicaSetConnection(pair,
            replicaSet=self.name,
            **kwargs)

class TestConnection(TestConnectionReplicaSetBase):
    def test_connect(self):
        self.assertRaises(ConnectionFailure, ReplicaSetConnection,
                          "somedomainthatdoesntexist.org:27017",
                          replicaSet=self.name,
                          connectTimeoutMS=600)
        self.assertRaises(ConfigurationError, ReplicaSetConnection,
                          pair, replicaSet='fdlksjfdslkjfd')
        self.assert_(ReplicaSetConnection(pair, replicaSet=self.name))

    def test_repr(self):
        connection = self._get_connection()
        self.assertEqual(repr(connection),
                         "ReplicaSetConnection(%r)" % (["%s:%d" % n
                                                         for n in 
                                                         self.hosts],))

    def test_properties(self):
        c = ReplicaSetConnection(pair, replicaSet=self.name)
        c.admin.command('ping')
        self.assertEqual(c.primary, self.primary)
        self.assertEqual(c.hosts, self.hosts)
        self.assertEqual(c.arbiters, self.arbiters)
        self.assertEqual(c.read_preference, ReadPreference.PRIMARY)
        self.assertEqual(c.max_pool_size, 10)
        self.assertEqual(c.document_class, dict)
        self.assertEqual(c.tz_aware, False)
        self.assertEqual(c.slave_okay, False)
        self.assertEqual(c.safe, False)
        c.close()

        c = ReplicaSetConnection(pair, replicaSet=self.name, max_pool_size=25,
                                 document_class=SON, tz_aware=True,
                                 slaveOk=False, safe=True,
                                 read_preference=ReadPreference.SECONDARY)
        c.admin.command('ping')
        self.assertEqual(c.primary, self.primary)
        self.assertEqual(c.hosts, self.hosts)
        self.assertEqual(c.arbiters, self.arbiters)
        self.assertEqual(c.read_preference, ReadPreference.SECONDARY)
        self.assertEqual(c.max_pool_size, 25)
        self.assertEqual(c.document_class, SON)
        self.assertEqual(c.tz_aware, True)
        self.assertEqual(c.slave_okay, False)
        self.assertEqual(c.safe, True)

        if version.at_least(c, (1, 7, 4)):
            self.assertEqual(c.max_bson_size, 16777216)
        else:
            self.assertEqual(c.max_bson_size, 4194304)
        c.close()

    def test_get_db(self):
        connection = self._get_connection()

        def make_db(base, name):
            return base[name]

        self.assertRaises(InvalidName, make_db, connection, "")
        self.assertRaises(InvalidName, make_db, connection, "te$t")
        self.assertRaises(InvalidName, make_db, connection, "te.t")
        self.assertRaises(InvalidName, make_db, connection, "te\\t")
        self.assertRaises(InvalidName, make_db, connection, "te/t")
        self.assertRaises(InvalidName, make_db, connection, "te st")

        self.assert_(isinstance(connection.test, Database))
        self.assertEqual(connection.test, connection["test"])
        self.assertEqual(connection.test, Database(connection, "test"))
        connection.close()

    def test_operations(self):
        c = self._get_connection()

        db = c.pymongo_test
        db.test.remove({})
        self.assertEqual(0, db.test.count())
        db.test.insert({'foo': 'x'}, safe=True, w=self.w)
        self.assertEqual(1, db.test.count())

        cursor = db.test.find()
        self.assertEqual('x', cursor.next()['foo'])
        # Ensure we read from the primary
        self.assertEqual(c.primary, cursor._Cursor__connection_id)

        cursor = db.test.find(read_preference=ReadPreference.SECONDARY)
        self.assertEqual('x', cursor.next()['foo'])
        # Ensure we didn't read from the primary
        self.assertTrue(cursor._Cursor__connection_id in c.secondaries)

        self.assertEqual(1, db.test.count())
        db.test.remove({})
        self.assertEqual(0, db.test.count())
        c.drop_database(db)<commit_msg>Make sure we don't hang in test_operations when connected to a replica set in recovery.<commit_after># Copyright 2009-2011 10gen, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Test the replica_set_connection module."""

import datetime
import os
import signal
import sys
import socket
import time
import thread
import unittest
sys.path[0:0] = [""]

from nose.plugins.skip import SkipTest

from bson.son import SON
from bson.tz_util import utc
from pymongo import ReadPreference
from pymongo.connection import Connection
from pymongo.replica_set_connection import ReplicaSetConnection
from pymongo.replica_set_connection import _partition_node
from pymongo.database import Database
from pymongo.errors import (AutoReconnect,
                            ConfigurationError,
                            ConnectionFailure,
                            InvalidName,
                            OperationFailure)
from test import version

host = os.environ.get("DB_IP", socket.gethostname())
port = int(os.environ.get("DB_PORT", 27017))
pair = '%s:%d' % (host, port)

class TestReplicaSetConnectionAgainstStandalone(unittest.TestCase):
    """This is a funny beast -- we want to run tests for ReplicaSetConnection
    but only if the database at DB_IP and DB_PORT is a standalone.
    """
    def setUp(self):
        conn = Connection(pair)
        response = conn.admin.command('ismaster')
        if 'setName' in response:
            raise SkipTest()

    def test_connect(self):
        self.assertRaises(ConfigurationError, ReplicaSetConnection,
                          pair, replicaSet='anything',
                          connectTimeoutMS=600)


class TestConnectionReplicaSetBase(unittest.TestCase):
    def setUp(self):
        conn = Connection(pair)
        response = conn.admin.command('ismaster')
        if 'setName' in response:
            self.name = str(response['setName'])
            self.w = len(response['hosts'])
            self.hosts = set([_partition_node(h)
                              for h in response["hosts"]])
            self.arbiters = set([_partition_node(h)
                                 for h in response.get("arbiters", [])])

            repl_set_status = conn.admin.command('replSetGetStatus')
            primary_info = [
                m for m in repl_set_status['members']
                if m['stateStr'] == 'PRIMARY'
            ][0]

            self.primary = _partition_node(primary_info['name'])
        else:
            raise SkipTest()

    def _get_connection(self, **kwargs):
        return ReplicaSetConnection(pair,
            replicaSet=self.name,
            **kwargs)

class TestConnection(TestConnectionReplicaSetBase):
    def test_connect(self):
        self.assertRaises(ConnectionFailure, ReplicaSetConnection,
                          "somedomainthatdoesntexist.org:27017",
                          replicaSet=self.name,
                          connectTimeoutMS=600)
        self.assertRaises(ConfigurationError, ReplicaSetConnection,
                          pair, replicaSet='fdlksjfdslkjfd')
        self.assert_(ReplicaSetConnection(pair, replicaSet=self.name))

    def test_repr(self):
        connection = self._get_connection()
        self.assertEqual(repr(connection),
                         "ReplicaSetConnection(%r)" % (["%s:%d" % n
                                                         for n in 
                                                         self.hosts],))

    def test_properties(self):
        c = ReplicaSetConnection(pair, replicaSet=self.name)
        c.admin.command('ping')
        self.assertEqual(c.primary, self.primary)
        self.assertEqual(c.hosts, self.hosts)
        self.assertEqual(c.arbiters, self.arbiters)
        self.assertEqual(c.read_preference, ReadPreference.PRIMARY)
        self.assertEqual(c.max_pool_size, 10)
        self.assertEqual(c.document_class, dict)
        self.assertEqual(c.tz_aware, False)
        self.assertEqual(c.slave_okay, False)
        self.assertEqual(c.safe, False)
        c.close()

        c = ReplicaSetConnection(pair, replicaSet=self.name, max_pool_size=25,
                                 document_class=SON, tz_aware=True,
                                 slaveOk=False, safe=True,
                                 read_preference=ReadPreference.SECONDARY)
        c.admin.command('ping')
        self.assertEqual(c.primary, self.primary)
        self.assertEqual(c.hosts, self.hosts)
        self.assertEqual(c.arbiters, self.arbiters)
        self.assertEqual(c.read_preference, ReadPreference.SECONDARY)
        self.assertEqual(c.max_pool_size, 25)
        self.assertEqual(c.document_class, SON)
        self.assertEqual(c.tz_aware, True)
        self.assertEqual(c.slave_okay, False)
        self.assertEqual(c.safe, True)

        if version.at_least(c, (1, 7, 4)):
            self.assertEqual(c.max_bson_size, 16777216)
        else:
            self.assertEqual(c.max_bson_size, 4194304)
        c.close()

    def test_get_db(self):
        connection = self._get_connection()

        def make_db(base, name):
            return base[name]

        self.assertRaises(InvalidName, make_db, connection, "")
        self.assertRaises(InvalidName, make_db, connection, "te$t")
        self.assertRaises(InvalidName, make_db, connection, "te.t")
        self.assertRaises(InvalidName, make_db, connection, "te\\t")
        self.assertRaises(InvalidName, make_db, connection, "te/t")
        self.assertRaises(InvalidName, make_db, connection, "te st")

        self.assert_(isinstance(connection.test, Database))
        self.assertEqual(connection.test, connection["test"])
        self.assertEqual(connection.test, Database(connection, "test"))
        connection.close()

    def test_operations(self):
        c = self._get_connection()

        # Check explicitly for a case we've commonly hit in tests:
        # a replica set is started with a tiny oplog, a previous
        # test does a big insert that leaves the secondaries
        # permanently "RECOVERING", and our insert(w=self.w) hangs
        # forever.
        rs_status = c.admin.command('replSetGetStatus')
        members = rs_status['members']
        self.assertFalse(
            [m for m in members if m['stateStr'] == 'RECOVERING'],
            "Replica set is recovering, try a larger oplogSize next time"
        )

        db = c.pymongo_test
        db.test.remove({})
        self.assertEqual(0, db.test.count())
        db.test.insert({'foo': 'x'}, safe=True, w=self.w)
        self.assertEqual(1, db.test.count())

        cursor = db.test.find()
        self.assertEqual('x', cursor.next()['foo'])
        # Ensure we read from the primary
        self.assertEqual(c.primary, cursor._Cursor__connection_id)

        cursor = db.test.find(read_preference=ReadPreference.SECONDARY)
        self.assertEqual('x', cursor.next()['foo'])
        # Ensure we didn't read from the primary
        self.assertTrue(cursor._Cursor__connection_id in c.secondaries)

        self.assertEqual(1, db.test.count())
        db.test.remove({})
        self.assertEqual(0, db.test.count())
        c.drop_database(db)
{'fraction_non_alphanumeric': 0.10031284004352557, 'fraction_numerical': 0.008297062023939063, 'mean_word_length': 2.403147419578801, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 20, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7132248', 'n_tokens_mistral': 4194, 'n_tokens_neox': 4062, 'n_words': 941}