(self)
| 91 | |
| 92 | def create_test(scenario_def): |
| 93 | def run_scenario(self): |
| 94 | _, hosts = get_addresses(scenario_def["topology_description"]["servers"]) |
| 95 | # "Eligible servers" is defined in the server selection spec as |
| 96 | # the set of servers matching both the ReadPreference's mode |
| 97 | # and tag sets. |
| 98 | top_suitable = create_topology(scenario_def, local_threshold_ms=1000000) |
| 99 | |
| 100 | # "In latency window" is defined in the server selection |
| 101 | # spec as the subset of suitable_servers that falls within the |
| 102 | # allowable latency window. |
| 103 | top_latency = create_topology(scenario_def) |
| 104 | |
| 105 | top_suitable_deprioritized_servers = [ |
| 106 | top_suitable.get_server_by_address(clean_node(server["address"])) |
| 107 | for server in scenario_def.get("deprioritized_servers", []) |
| 108 | ] |
| 109 | top_latency_deprioritized_servers = [ |
| 110 | top_latency.get_server_by_address(clean_node(server["address"])) |
| 111 | for server in scenario_def.get("deprioritized_servers", []) |
| 112 | ] |
| 113 | |
| 114 | # Create server selector. |
| 115 | if scenario_def.get("operation") == "write": |
| 116 | pref = writable_server_selector |
| 117 | else: |
| 118 | # Make first letter lowercase to match read_pref's modes. |
| 119 | pref_def = scenario_def["read_preference"] |
| 120 | if scenario_def.get("error"): |
| 121 | with self.assertRaises((ConfigurationError, ValueError)): |
| 122 | # Error can be raised when making Read Pref or selecting. |
| 123 | pref = parse_read_preference(pref_def) |
| 124 | top_latency.select_server(pref, _Op.TEST) |
| 125 | return |
| 126 | |
| 127 | pref = parse_read_preference(pref_def) |
| 128 | |
| 129 | # Select servers. |
| 130 | if not scenario_def.get("suitable_servers"): |
| 131 | with self.assertRaises(AutoReconnect): |
| 132 | top_suitable.select_server( |
| 133 | pref, |
| 134 | _Op.TEST, |
| 135 | server_selection_timeout=0, |
| 136 | deprioritized_servers=top_suitable_deprioritized_servers, |
| 137 | ) |
| 138 | |
| 139 | return |
| 140 | |
| 141 | if not scenario_def["in_latency_window"]: |
| 142 | with self.assertRaises(AutoReconnect): |
| 143 | top_latency.select_server( |
| 144 | pref, |
| 145 | _Op.TEST, |
| 146 | server_selection_timeout=0, |
| 147 | deprioritized_servers=top_latency_deprioritized_servers, |
| 148 | ) |
| 149 | |
| 150 | return |
nothing calls this directly
no test coverage detected