(self)
| 1684 | self.assertIn("heartbeatFrequencyMS", str(context.exception)) |
| 1685 | |
| 1686 | def test_compression(self): |
| 1687 | def compression_settings(client): |
| 1688 | pool_options = client.options.pool_options |
| 1689 | return pool_options._compression_settings |
| 1690 | |
| 1691 | uri = "mongodb://localhost:27017/?compressors=zlib" |
| 1692 | client = self.simple_client(uri, connect=False) |
| 1693 | opts = compression_settings(client) |
| 1694 | self.assertEqual(opts.compressors, ["zlib"]) |
| 1695 | uri = "mongodb://localhost:27017/?compressors=zlib&zlibCompressionLevel=4" |
| 1696 | client = self.simple_client(uri, connect=False) |
| 1697 | opts = compression_settings(client) |
| 1698 | self.assertEqual(opts.compressors, ["zlib"]) |
| 1699 | self.assertEqual(opts.zlib_compression_level, 4) |
| 1700 | uri = "mongodb://localhost:27017/?compressors=zlib&zlibCompressionLevel=-1" |
| 1701 | client = self.simple_client(uri, connect=False) |
| 1702 | opts = compression_settings(client) |
| 1703 | self.assertEqual(opts.compressors, ["zlib"]) |
| 1704 | self.assertEqual(opts.zlib_compression_level, -1) |
| 1705 | uri = "mongodb://localhost:27017" |
| 1706 | client = self.simple_client(uri, connect=False) |
| 1707 | opts = compression_settings(client) |
| 1708 | self.assertEqual(opts.compressors, []) |
| 1709 | self.assertEqual(opts.zlib_compression_level, -1) |
| 1710 | uri = "mongodb://localhost:27017/?compressors=foobar" |
| 1711 | client = self.simple_client(uri, connect=False) |
| 1712 | opts = compression_settings(client) |
| 1713 | self.assertEqual(opts.compressors, []) |
| 1714 | self.assertEqual(opts.zlib_compression_level, -1) |
| 1715 | uri = "mongodb://localhost:27017/?compressors=foobar,zlib" |
| 1716 | client = self.simple_client(uri, connect=False) |
| 1717 | opts = compression_settings(client) |
| 1718 | self.assertEqual(opts.compressors, ["zlib"]) |
| 1719 | self.assertEqual(opts.zlib_compression_level, -1) |
| 1720 | |
| 1721 | # According to the connection string spec, unsupported values |
| 1722 | # just raise a warning and are ignored. |
| 1723 | uri = "mongodb://localhost:27017/?compressors=zlib&zlibCompressionLevel=10" |
| 1724 | client = self.simple_client(uri, connect=False) |
| 1725 | opts = compression_settings(client) |
| 1726 | self.assertEqual(opts.compressors, ["zlib"]) |
| 1727 | self.assertEqual(opts.zlib_compression_level, -1) |
| 1728 | uri = "mongodb://localhost:27017/?compressors=zlib&zlibCompressionLevel=-2" |
| 1729 | client = self.simple_client(uri, connect=False) |
| 1730 | opts = compression_settings(client) |
| 1731 | self.assertEqual(opts.compressors, ["zlib"]) |
| 1732 | self.assertEqual(opts.zlib_compression_level, -1) |
| 1733 | |
| 1734 | if not _have_snappy(): |
| 1735 | uri = "mongodb://localhost:27017/?compressors=snappy" |
| 1736 | client = self.simple_client(uri, connect=False) |
| 1737 | opts = compression_settings(client) |
| 1738 | self.assertEqual(opts.compressors, []) |
| 1739 | else: |
| 1740 | uri = "mongodb://localhost:27017/?compressors=snappy" |
| 1741 | client = self.simple_client(uri, connect=False) |
| 1742 | opts = compression_settings(client) |
| 1743 | self.assertEqual(opts.compressors, ["snappy"]) |
nothing calls this directly
no test coverage detected