<s> from __future__ import absolute_import , division <EOL> from mock import patch <EOL> from time import time <EOL> from twisted . trial import unittest <EOL> from twisted . internet import defer <EOL> from txmongo import connection <EOL> from txmongo . utils import check_deadline <EOL> from txmongo . errors import TimeExceeded <EOL> mongo_host = "<STR_LIT:127.0.0.1>" <EOL> mongo_port = <NUM_LIT> <EOL> class TestMongoConnection ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . named_conn = connection . ConnectionPool ( "<STR_LIT>" ) <EOL> self . unnamed_conn = connection . ConnectionPool ( "<STR_LIT:127.0.0.1>" ) <EOL> @ defer . inlineCallbacks <EOL> def tearDown ( self ) : <EOL> yield self . named_conn . disconnect ( ) <EOL> yield self . unnamed_conn . disconnect ( ) <EOL> def test_GetDefaultDatabase ( self ) : <EOL> self . assertEqual ( self . named_conn . get_default_database ( ) . name , <EOL> self . named_conn [ "<STR_LIT>" ] . name ) <EOL> self . assertEqual ( self . unnamed_conn . get_default_database ( ) , None ) <EOL> def test_Misc ( self ) : <EOL> result = self . named_conn . getprotocols ( ) <EOL> result [ <NUM_LIT:0> ] . uri [ '<STR_LIT>' ] . pop ( ) <EOL> self . assertTrue ( len ( result [ <NUM_LIT:0> ] . uri [ '<STR_LIT>' ] ) == <NUM_LIT:0> ) <EOL> self . assertEqual ( "<STR_LIT>" , repr ( self . named_conn ) ) <EOL> @ defer . inlineCallbacks <EOL> def test_uri_input ( self ) : <EOL> test = connection . ConnectionPool ( ) <EOL> yield test . disconnect ( ) <EOL> test = connection . ConnectionPool ( "<STR_LIT>" ) <EOL> yield test . disconnect ( ) <EOL> test = connection . ConnectionPool ( u"<STR_LIT>" ) <EOL> yield test . disconnect ( ) <EOL> self . assertRaises ( AssertionError , connection . ConnectionPool , object ) <EOL> self . assertRaises ( AssertionError , connection . ConnectionPool , <NUM_LIT:1> ) <EOL> @ defer . inlineCallbacks <EOL> def test_Timeout_and_Deadline ( self ) : <EOL> yield self . named_conn . db . coll . insert ( { '<STR_LIT:x>' : <NUM_LIT> } , safe = True , timeout = <NUM_LIT:10> ) <EOL> yield self . named_conn . db . coll . insert ( { '<STR_LIT:x>' : <NUM_LIT> } , safe = True , deadline = time ( ) + <NUM_LIT:10> ) <EOL> d_insert = self . named_conn . db . coll . insert ( { '<STR_LIT:x>' : <NUM_LIT> } , safe = True , deadline = time ( ) - <NUM_LIT:10> ) <EOL> yield self . assertFailure ( d_insert , TimeExceeded ) <EOL> d_insert = self . named_conn . db . coll . insert ( { '<STR_LIT:x>' : <NUM_LIT> } , safe = True , timeout = - <NUM_LIT:10> ) <EOL> yield self . assertFailure ( d_insert , TimeExceeded ) <EOL> def patch_deadline ( _ ) : <EOL> check_deadline ( time ( ) - <NUM_LIT:2> ) <EOL> with patch ( '<STR_LIT>' , side_effect = patch_deadline ) : <EOL> d_insert = self . named_conn . db . coll . find_one ( <EOL> { '<STR_LIT:x>' : <NUM_LIT> } , deadline = time ( ) + <NUM_LIT:2> ) <EOL> yield self . assertFailure ( d_insert , TimeExceeded ) </s>
