(self)
| 1727 | self.assertIn("heartbeatFrequencyMS", str(context.exception)) |
| 1728 | |
| 1729 | async def test_compression(self): |
| 1730 | def compression_settings(client): |
| 1731 | pool_options = client.options.pool_options |
| 1732 | return pool_options._compression_settings |
| 1733 | |
| 1734 | uri = "mongodb://localhost:27017/?compressors=zlib" |
| 1735 | client = self.simple_client(uri, connect=False) |
| 1736 | opts = compression_settings(client) |
| 1737 | self.assertEqual(opts.compressors, ["zlib"]) |
| 1738 | uri = "mongodb://localhost:27017/?compressors=zlib&zlibCompressionLevel=4" |
| 1739 | client = self.simple_client(uri, connect=False) |
| 1740 | opts = compression_settings(client) |
| 1741 | self.assertEqual(opts.compressors, ["zlib"]) |
| 1742 | self.assertEqual(opts.zlib_compression_level, 4) |
| 1743 | uri = "mongodb://localhost:27017/?compressors=zlib&zlibCompressionLevel=-1" |
| 1744 | client = self.simple_client(uri, connect=False) |
| 1745 | opts = compression_settings(client) |
| 1746 | self.assertEqual(opts.compressors, ["zlib"]) |
| 1747 | self.assertEqual(opts.zlib_compression_level, -1) |
| 1748 | uri = "mongodb://localhost:27017" |
| 1749 | client = self.simple_client(uri, connect=False) |
| 1750 | opts = compression_settings(client) |
| 1751 | self.assertEqual(opts.compressors, []) |
| 1752 | self.assertEqual(opts.zlib_compression_level, -1) |
| 1753 | uri = "mongodb://localhost:27017/?compressors=foobar" |
| 1754 | client = self.simple_client(uri, connect=False) |
| 1755 | opts = compression_settings(client) |
| 1756 | self.assertEqual(opts.compressors, []) |
| 1757 | self.assertEqual(opts.zlib_compression_level, -1) |
| 1758 | uri = "mongodb://localhost:27017/?compressors=foobar,zlib" |
| 1759 | client = self.simple_client(uri, connect=False) |
| 1760 | opts = compression_settings(client) |
| 1761 | self.assertEqual(opts.compressors, ["zlib"]) |
| 1762 | self.assertEqual(opts.zlib_compression_level, -1) |
| 1763 | |
| 1764 | # According to the connection string spec, unsupported values |
| 1765 | # just raise a warning and are ignored. |
| 1766 | uri = "mongodb://localhost:27017/?compressors=zlib&zlibCompressionLevel=10" |
| 1767 | client = self.simple_client(uri, connect=False) |
| 1768 | opts = compression_settings(client) |
| 1769 | self.assertEqual(opts.compressors, ["zlib"]) |
| 1770 | self.assertEqual(opts.zlib_compression_level, -1) |
| 1771 | uri = "mongodb://localhost:27017/?compressors=zlib&zlibCompressionLevel=-2" |
| 1772 | client = self.simple_client(uri, connect=False) |
| 1773 | opts = compression_settings(client) |
| 1774 | self.assertEqual(opts.compressors, ["zlib"]) |
| 1775 | self.assertEqual(opts.zlib_compression_level, -1) |
| 1776 | |
| 1777 | if not _have_snappy(): |
| 1778 | uri = "mongodb://localhost:27017/?compressors=snappy" |
| 1779 | client = self.simple_client(uri, connect=False) |
| 1780 | opts = compression_settings(client) |
| 1781 | self.assertEqual(opts.compressors, []) |
| 1782 | else: |
| 1783 | uri = "mongodb://localhost:27017/?compressors=snappy" |
| 1784 | client = self.simple_client(uri, connect=False) |
| 1785 | opts = compression_settings(client) |
| 1786 | self.assertEqual(opts.compressors, ["snappy"]) |
nothing calls this directly
no test coverage detected