(self)
| 680 | super().tearDown() |
| 681 | |
| 682 | def test_uri_options(self): |
| 683 | # Test default to admin |
| 684 | host, port = client_context.host, client_context.port |
| 685 | client = self.rs_or_single_client_noauth("mongodb://admin:pass@%s:%d" % (host, port)) |
| 686 | self.assertTrue(client.admin.command("dbstats")) |
| 687 | |
| 688 | if client_context.is_rs: |
| 689 | uri = "mongodb://admin:pass@%s:%d/?replicaSet=%s" % ( |
| 690 | host, |
| 691 | port, |
| 692 | client_context.replica_set_name, |
| 693 | ) |
| 694 | client = self.single_client_noauth(uri) |
| 695 | self.assertTrue(client.admin.command("dbstats")) |
| 696 | db = client.get_database("admin", read_preference=ReadPreference.SECONDARY) |
| 697 | self.assertTrue(db.command("dbstats")) |
| 698 | |
| 699 | # Test explicit database |
| 700 | uri = "mongodb://user:pass@%s:%d/pymongo_test" % (host, port) |
| 701 | client = self.rs_or_single_client_noauth(uri) |
| 702 | with self.assertRaises(OperationFailure): |
| 703 | client.admin.command("dbstats") |
| 704 | self.assertTrue(client.pymongo_test.command("dbstats")) |
| 705 | |
| 706 | if client_context.is_rs: |
| 707 | uri = "mongodb://user:pass@%s:%d/pymongo_test?replicaSet=%s" % ( |
| 708 | host, |
| 709 | port, |
| 710 | client_context.replica_set_name, |
| 711 | ) |
| 712 | client = self.single_client_noauth(uri) |
| 713 | with self.assertRaises(OperationFailure): |
| 714 | client.admin.command("dbstats") |
| 715 | self.assertTrue(client.pymongo_test.command("dbstats")) |
| 716 | db = client.get_database("pymongo_test", read_preference=ReadPreference.SECONDARY) |
| 717 | self.assertTrue(db.command("dbstats")) |
| 718 | |
| 719 | # Test authSource |
| 720 | uri = "mongodb://user:pass@%s:%d/pymongo_test2?authSource=pymongo_test" % (host, port) |
| 721 | client = self.rs_or_single_client_noauth(uri) |
| 722 | with self.assertRaises(OperationFailure): |
| 723 | client.pymongo_test2.command("dbstats") |
| 724 | self.assertTrue(client.pymongo_test.command("dbstats")) |
| 725 | |
| 726 | if client_context.is_rs: |
| 727 | uri = ( |
| 728 | "mongodb://user:pass@%s:%d/pymongo_test2?replicaSet=" |
| 729 | "%s;authSource=pymongo_test" % (host, port, client_context.replica_set_name) |
| 730 | ) |
| 731 | client = self.single_client_noauth(uri) |
| 732 | with self.assertRaises(OperationFailure): |
| 733 | client.pymongo_test2.command("dbstats") |
| 734 | self.assertTrue(client.pymongo_test.command("dbstats")) |
| 735 | db = client.get_database("pymongo_test", read_preference=ReadPreference.SECONDARY) |
| 736 | self.assertTrue(db.command("dbstats")) |
| 737 | |
| 738 | |
| 739 | if __name__ == "__main__": |
nothing calls this directly
no test coverage detected