<s> """<STR_LIT>""" <EOL> import os <EOL> import sys <EOL> import threading <EOL> sys . path [ <NUM_LIT:0> : <NUM_LIT:0> ] = [ "<STR_LIT>" ] <EOL> from bson import json_util <EOL> from pymongo import common <EOL> from pymongo . topology import Topology <EOL> from pymongo . topology_description import TOPOLOGY_TYPE <EOL> from pymongo . ismaster import IsMaster <EOL> from pymongo . server_description import ServerDescription , SERVER_TYPE <EOL> from pymongo . settings import TopologySettings <EOL> from pymongo . uri_parser import parse_uri <EOL> from test import unittest <EOL> _TEST_PATH = os . path . join ( <EOL> os . path . dirname ( os . path . realpath ( __file__ ) ) , '<STR_LIT>' ) <EOL> class MockSocketInfo ( object ) : <EOL> def close ( self ) : <EOL> pass <EOL> def __enter__ ( self ) : <EOL> return self <EOL> def __exit__ ( self , exc_type , exc_val , exc_tb ) : <EOL> pass <EOL> class MockPool ( object ) : <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> self . pool_id = <NUM_LIT:0> <EOL> self . _lock = threading . Lock ( ) <EOL> def reset ( self ) : <EOL> with self . _lock : <EOL> self . pool_id += <NUM_LIT:1> <EOL> class MockMonitor ( object ) : <EOL> def __init__ ( self , server_description , topology , pool , topology_settings ) : <EOL> self . _server_description = server_description <EOL> self . _topology = topology <EOL> def open ( self ) : <EOL> pass <EOL> def request_check ( self ) : <EOL> pass <EOL> def close ( self ) : <EOL> pass <EOL> def create_mock_topology ( uri , monitor_class = MockMonitor ) : <EOL> parsed_uri = parse_uri ( uri . replace ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> replica_set_name = None <EOL> if '<STR_LIT>' in parsed_uri [ '<STR_LIT>' ] : <EOL> replica_set_name = parsed_uri [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> topology_settings = TopologySettings ( <EOL> parsed_uri [ '<STR_LIT>' ] , <EOL> replica_set_name = replica_set_name , <EOL> pool_class = MockPool , <EOL> monitor_class = monitor_class ) <EOL> c = Topology ( topology_settings ) <EOL> c . open ( ) <EOL> return c <EOL> def got_ismaster ( topology , server_address , ismaster_response ) : <EOL> server_description = ServerDescription ( <EOL> server_address , IsMaster ( ismaster_response ) , <NUM_LIT:0> ) <EOL> topology . on_change ( server_description ) <EOL> def get_type ( topology , hostname ) : <EOL> description = topology . get_server_by_address ( ( hostname , <NUM_LIT> ) ) . description <EOL> return description . server_type <EOL> class TestAllScenarios ( unittest . TestCase ) : <EOL> pass <EOL> def topology_type_name ( topology_type ) : <EOL> return TOPOLOGY_TYPE . _fields [ topology_type ] <EOL> def server_type_name ( server_type ) : <EOL> return SERVER_TYPE . _fields [ server_type ] <EOL> def check_outcome ( self , topology , outcome ) : <EOL> expected_servers = outcome [ '<STR_LIT>' ] <EOL> self . assertEqual ( <EOL> len ( topology . description . server_descriptions ( ) ) , <EOL> len ( expected_servers ) ) <EOL> for expected_server_address , expected_server in expected_servers . items ( ) : <EOL> node = common . partition_node ( expected_server_address ) <EOL> self . assertTrue ( topology . has_server ( node ) ) <EOL> actual_server = topology . get_server_by_address ( node ) <EOL> actual_server_description = actual_server . description <EOL> if expected_server [ '<STR_LIT:type>' ] == '<STR_LIT>' : <EOL> expected_server_type = SERVER_TYPE . Unknown <EOL> else : <EOL> expected_server_type = getattr ( <EOL> SERVER_TYPE , expected_server [ '<STR_LIT:type>' ] ) <EOL> self . assertEqual ( <EOL> server_type_name ( expected_server_type ) , <EOL> server_type_name ( actual_server_description . server_type ) ) <EOL> self . assertEqual ( <EOL> expected_server [ '<STR_LIT>' ] , <EOL> actual_server_description . replica_set_name ) <EOL> self . assertEqual ( <EOL> expected_server . get ( '<STR_LIT>' ) , <EOL> actual_server_description . set_version ) <EOL> self . assertEqual ( <EOL> expected_server . get ( '<STR_LIT>' ) , <EOL> actual_server_description . election_id ) <EOL> self . assertEqual ( outcome [ '<STR_LIT>' ] , topology . description . replica_set_name ) <EOL> expected_topology_type = getattr ( TOPOLOGY_TYPE , outcome [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( topology_type_name ( expected_topology_type ) , <EOL> topology_type_name ( topology . description . topology_type ) ) <EOL> def create_test ( scenario_def ) : <EOL> def run_scenario ( self ) : <EOL> c = create_mock_topology ( scenario_def [ '<STR_LIT>' ] ) <EOL> for phase in scenario_def [ '<STR_LIT>' ] : <EOL> for response in phase [ '<STR_LIT>' ] : <EOL> got_ismaster ( c , <EOL> common . partition_node ( response [ <NUM_LIT:0> ] ) , <EOL> response [ <NUM_LIT:1> ] ) <EOL> check_outcome ( self , c , phase [ '<STR_LIT>' ] ) <EOL> return run_scenario <EOL> def create_tests ( ) : <EOL> for dirpath , _ , filenames in os . walk ( _TEST_PATH ) : <EOL> dirname = os . path . split ( dirpath ) [ - <NUM_LIT:1> ] <EOL> for filename in filenames : <EOL> with open ( os . path . join ( dirpath , filename ) ) as scenario_stream : <EOL> scenario_def = json_util . loads ( scenario_stream . read ( ) ) <EOL> new_test = create_test ( scenario_def ) <EOL> test_name = '<STR_LIT>' % ( <EOL> dirname , os . path . splitext ( filename ) [ <NUM_LIT:0> ] ) <EOL> new_test . __name__ = test_name <EOL> setattr ( TestAllScenarios , new_test . __name__ , new_test ) <EOL> create_tests ( ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> unittest . main ( ) </s>
