(self)
| 88 | |
| 89 | def create_test(test_case): |
| 90 | async def run_test(self): |
| 91 | uri = test_case["uri"] |
| 92 | seeds = test_case.get("seeds") |
| 93 | num_seeds = test_case.get("numSeeds", len(seeds or [])) |
| 94 | hosts = test_case.get("hosts") |
| 95 | num_hosts = test_case.get("numHosts", len(hosts or [])) |
| 96 | |
| 97 | options = test_case.get("options", {}) |
| 98 | if "ssl" in options: |
| 99 | options["tls"] = options.pop("ssl") |
| 100 | parsed_options = test_case.get("parsed_options") |
| 101 | # See DRIVERS-1324, unless tls is explicitly set to False we need TLS. |
| 102 | needs_tls = not (options and (options.get("ssl") is False or options.get("tls") is False)) |
| 103 | if needs_tls and not async_client_context.tls: |
| 104 | self.skipTest("this test requires a TLS cluster") |
| 105 | if not needs_tls and async_client_context.tls: |
| 106 | self.skipTest("this test requires a non-TLS cluster") |
| 107 | |
| 108 | if seeds: |
| 109 | seeds = split_hosts(",".join(seeds)) |
| 110 | if hosts: |
| 111 | hosts = frozenset(split_hosts(",".join(hosts))) |
| 112 | |
| 113 | if seeds or num_seeds: |
| 114 | result = await parse_uri(uri, validate=True) |
| 115 | if seeds is not None: |
| 116 | self.assertEqual(sorted(result["nodelist"]), sorted(seeds)) |
| 117 | if num_seeds is not None: |
| 118 | self.assertEqual(len(result["nodelist"]), num_seeds) |
| 119 | if options: |
| 120 | opts = result["options"] |
| 121 | if "readpreferencetags" in opts: |
| 122 | rpts = validate_read_preference_tags( |
| 123 | "readPreferenceTags", opts.pop("readpreferencetags") |
| 124 | ) |
| 125 | opts["readPreferenceTags"] = rpts |
| 126 | self.assertEqual(result["options"], options) |
| 127 | if parsed_options: |
| 128 | for opt, expected in parsed_options.items(): |
| 129 | if opt == "user": |
| 130 | self.assertEqual(result["username"], expected) |
| 131 | elif opt == "password": |
| 132 | self.assertEqual(result["password"], expected) |
| 133 | elif opt == "auth_database" or opt == "db": |
| 134 | self.assertEqual(result["database"], expected) |
| 135 | |
| 136 | hostname = next(iter(async_client_context.client.nodes))[0] |
| 137 | # The replica set members must be configured as 'localhost'. |
| 138 | if hostname == "localhost": |
| 139 | copts = async_client_context.default_client_options.copy() |
| 140 | # Remove tls since SRV parsing should add it automatically. |
| 141 | copts.pop("tls", None) |
| 142 | if async_client_context.tls: |
| 143 | # Our test certs don't support the SRV hosts used in these |
| 144 | # tests. |
| 145 | copts["tlsAllowInvalidHostnames"] = True |
| 146 | |
| 147 | client = self.simple_client(uri, **copts) |
no test coverage detected