(
seeds=None,
replica_set_name=None,
monitor_class=DummyMonitor,
direct_connection=False,
)
| 51 | |
| 52 | |
| 53 | def create_mock_topology( |
| 54 | seeds=None, |
| 55 | replica_set_name=None, |
| 56 | monitor_class=DummyMonitor, |
| 57 | direct_connection=False, |
| 58 | ): |
| 59 | partitioned_seeds = list(map(common.partition_node, seeds or ["a"])) |
| 60 | topology_settings = TopologySettings( |
| 61 | partitioned_seeds, |
| 62 | replica_set_name=replica_set_name, |
| 63 | pool_class=MockPool, # type: ignore[arg-type] |
| 64 | monitor_class=monitor_class, |
| 65 | direct_connection=direct_connection, |
| 66 | ) |
| 67 | |
| 68 | t = Topology(topology_settings) |
| 69 | t.open() |
| 70 | return t |
| 71 | |
| 72 | |
| 73 | def got_hello(topology, server_address, hello_response): |
no test coverage detected