(self, env_vars, expected_env)
| 2063 | self.assertNotIn("ServerHeartbeatFailedEvent", log_output) |
| 2064 | |
| 2065 | async def _test_handshake(self, env_vars, expected_env): |
| 2066 | with patch.dict("os.environ", env_vars): |
| 2067 | metadata = copy.deepcopy(_METADATA) |
| 2068 | if has_c(): |
| 2069 | metadata["driver"]["name"] = "PyMongo|c|async" |
| 2070 | else: |
| 2071 | metadata["driver"]["name"] = "PyMongo|async" |
| 2072 | if expected_env is not None: |
| 2073 | metadata["env"] = expected_env |
| 2074 | |
| 2075 | if "AWS_REGION" not in env_vars: |
| 2076 | os.environ["AWS_REGION"] = "" |
| 2077 | client = await self.async_rs_or_single_client(serverSelectionTimeoutMS=10000) |
| 2078 | await client.admin.command("ping") |
| 2079 | options = client.options |
| 2080 | self.assertEqual(options.pool_options.metadata, metadata) |
| 2081 | |
| 2082 | async def test_handshake_01_aws(self): |
| 2083 | await self._test_handshake( |
no test coverage detected