(uri, monitor_class=DummyMonitor)
| 87 | |
| 88 | |
| 89 | def create_mock_topology(uri, monitor_class=DummyMonitor): |
| 90 | parsed_uri = parse_uri(uri) |
| 91 | replica_set_name = None |
| 92 | direct_connection = None |
| 93 | load_balanced = None |
| 94 | if "replicaSet" in parsed_uri["options"]: |
| 95 | replica_set_name = parsed_uri["options"]["replicaSet"] |
| 96 | if "directConnection" in parsed_uri["options"]: |
| 97 | direct_connection = parsed_uri["options"]["directConnection"] |
| 98 | if "loadBalanced" in parsed_uri["options"]: |
| 99 | load_balanced = parsed_uri["options"]["loadBalanced"] |
| 100 | |
| 101 | topology_settings = TopologySettings( |
| 102 | parsed_uri["nodelist"], |
| 103 | replica_set_name=replica_set_name, |
| 104 | monitor_class=monitor_class, |
| 105 | direct_connection=direct_connection, |
| 106 | load_balanced=load_balanced, |
| 107 | ) |
| 108 | |
| 109 | c = Topology(topology_settings) |
| 110 | c.open() |
| 111 | return c |
| 112 | |
| 113 | |
| 114 | def got_hello(topology, server_address, hello_response): |
no test coverage detected