| 48 | |
| 49 | |
| 50 | def create_test(test_case): |
| 51 | def run_test(self): |
| 52 | uri = test_case["uri"] |
| 53 | valid = test_case["valid"] |
| 54 | credential = test_case.get("credential") |
| 55 | |
| 56 | if not valid: |
| 57 | with warnings.catch_warnings(): |
| 58 | warnings.simplefilter("default") |
| 59 | self.assertRaises(Exception, MongoClient, uri, connect=False) |
| 60 | else: |
| 61 | client = self.simple_client(uri, connect=False) |
| 62 | credentials = client.options.pool_options._credentials |
| 63 | if credential is None: |
| 64 | self.assertIsNone(credentials) |
| 65 | else: |
| 66 | self.assertIsNotNone(credentials) |
| 67 | self.assertEqual(credentials.username, credential["username"]) |
| 68 | self.assertEqual(credentials.password, credential["password"]) |
| 69 | self.assertEqual(credentials.source, credential["source"]) |
| 70 | if credential["mechanism"] is not None: |
| 71 | self.assertEqual(credentials.mechanism, credential["mechanism"]) |
| 72 | else: |
| 73 | self.assertEqual(credentials.mechanism, "DEFAULT") |
| 74 | expected = credential["mechanism_properties"] |
| 75 | if expected is not None: |
| 76 | actual = credentials.mechanism_properties |
| 77 | for key, value in expected.items(): |
| 78 | self.assertEqual(getattr(actual, key.lower()), value) |
| 79 | else: |
| 80 | if credential["mechanism"] == "MONGODB-AWS": |
| 81 | self.assertIsNone(credentials.mechanism_properties.aws_session_token) |
| 82 | else: |
| 83 | self.assertIsNone(credentials.mechanism_properties) |
| 84 | |
| 85 | return run_test |
| 86 | |
| 87 | |
| 88 | def create_tests(): |