(scenario_def, **kwargs)
| 54 | |
| 55 | |
| 56 | async def create_topology(scenario_def, **kwargs): |
| 57 | # Initialize topologies. |
| 58 | if "heartbeatFrequencyMS" in scenario_def: |
| 59 | frequency = int(scenario_def["heartbeatFrequencyMS"]) / 1000.0 |
| 60 | else: |
| 61 | frequency = HEARTBEAT_FREQUENCY |
| 62 | |
| 63 | seeds, hosts = get_addresses(scenario_def["topology_description"]["servers"]) |
| 64 | |
| 65 | topology_type = get_topology_type_name(scenario_def) |
| 66 | if topology_type == "LoadBalanced": |
| 67 | kwargs.setdefault("load_balanced", True) |
| 68 | # Force topology description to ReplicaSet |
| 69 | elif topology_type in ["ReplicaSetNoPrimary", "ReplicaSetWithPrimary"]: |
| 70 | kwargs.setdefault("replica_set_name", "rs") |
| 71 | settings = get_topology_settings_dict(heartbeat_frequency=frequency, seeds=seeds, **kwargs) |
| 72 | |
| 73 | # "Eligible servers" is defined in the server selection spec as |
| 74 | # the set of servers matching both the ReadPreference's mode |
| 75 | # and tag sets. |
| 76 | topology = Topology(TopologySettings(**settings)) |
| 77 | await topology.open() |
| 78 | |
| 79 | # Update topologies with server descriptions. |
| 80 | for server in scenario_def["topology_description"]["servers"]: |
| 81 | server_description = make_server_description(server, hosts) |
| 82 | await topology.on_change(server_description) |
| 83 | |
| 84 | # Assert that descriptions match |
| 85 | assert ( |
| 86 | scenario_def["topology_description"]["type"] == topology.description.topology_type_name |
| 87 | ), topology.description.topology_type_name |
| 88 | |
| 89 | return topology |
| 90 | |
| 91 | |
| 92 | def create_test(scenario_def): |
no test coverage detected