Make a ServerDescription from server info in a JSON test.
(server, hosts)
| 49 | |
| 50 | |
| 51 | def make_server_description(server, hosts): |
| 52 | """Make a ServerDescription from server info in a JSON test.""" |
| 53 | server_type = server["type"] |
| 54 | if server_type in ("Unknown", "PossiblePrimary"): |
| 55 | return ServerDescription(clean_node(server["address"]), Hello({})) |
| 56 | |
| 57 | hello_response = {"ok": True, "hosts": hosts} |
| 58 | if server_type not in ("Standalone", "Mongos", "RSGhost"): |
| 59 | hello_response["setName"] = "rs" |
| 60 | |
| 61 | if server_type == "RSPrimary": |
| 62 | hello_response[HelloCompat.LEGACY_CMD] = True |
| 63 | elif server_type == "RSSecondary": |
| 64 | hello_response["secondary"] = True |
| 65 | elif server_type == "Mongos": |
| 66 | hello_response["msg"] = "isdbgrid" |
| 67 | elif server_type == "RSGhost": |
| 68 | hello_response["isreplicaset"] = True |
| 69 | elif server_type == "RSArbiter": |
| 70 | hello_response["arbiterOnly"] = True |
| 71 | |
| 72 | hello_response["lastWrite"] = {"lastWriteDate": make_last_write_date(server)} |
| 73 | |
| 74 | for field in "maxWireVersion", "tags", "idleWritePeriodMillis": |
| 75 | if field in server: |
| 76 | hello_response[field] = server[field] |
| 77 | |
| 78 | hello_response.setdefault("maxWireVersion", MIN_SUPPORTED_WIRE_VERSION) |
| 79 | |
| 80 | # Sets _last_update_time to now. |
| 81 | sd = ServerDescription( |
| 82 | clean_node(server["address"]), |
| 83 | Hello(hello_response), |
| 84 | round_trip_time=server["avg_rtt_ms"] / 1000.0, |
| 85 | ) |
| 86 | |
| 87 | if "lastUpdateTime" in server: |
| 88 | sd._last_update_time = server["lastUpdateTime"] / 1000.0 # ms to sec. |
| 89 | |
| 90 | return sd |
| 91 | |
| 92 | |
| 93 | def get_topology_type_name(scenario_def): |
no test coverage detected