Method test_tls_version

tests/test_connect.py:2005–2054

(self)

                await con.close()

    async def test_tls_version(self):
        if self.cluster.get_pg_version() < (12, 0):
            self.skipTest("PostgreSQL < 12 cannot set ssl protocol version")

        # XXX: uvloop artifact
        old_handler = self.loop.get_exception_handler()

        with warnings.catch_warnings():
            warnings.filterwarnings(
                "ignore",
                message="ssl.TLSVersion.TLSv1_1 is deprecated",
                category=DeprecationWarning
            )
            try:
                self.loop.set_exception_handler(lambda *args: None)
                with self.assertRaisesRegex(
                    ssl.SSLError,
                    '(protocol version)|(handshake failure)',
                ):
                    await self.connect(
                        dsn='postgresql://ssl_user@localhost/postgres'
                            '?sslmode=require&ssl_min_protocol_version=TLSv1.3'
                    )
                with self.assertRaises((ssl.SSLError, ConnectionResetError)):
                    await self.connect(
                        dsn='postgresql://ssl_user@localhost/postgres'
                            '?sslmode=require'
                            '&ssl_min_protocol_version=TLSv1.1'
                            '&ssl_max_protocol_version=TLSv1.1'
                    )
                if not ssl.OPENSSL_VERSION.startswith('LibreSSL'):
                    with self.assertRaisesRegex(ssl.SSLError, 'no protocols'):
                        await self.connect(
                            dsn='postgresql://ssl_user@localhost/postgres'
                                '?sslmode=require'
                                '&ssl_min_protocol_version=TLSv1.2'
                                '&ssl_max_protocol_version=TLSv1.1'
                        )
                con = await self.connect(
                    dsn='postgresql://ssl_user@localhost/postgres'
                        '?sslmode=require'
                        '&ssl_min_protocol_version=TLSv1.2'
                        '&ssl_max_protocol_version=TLSv1.2'
                )
                try:
                    self.assertEqual(await con.fetchval('SELECT 42'), 42)
                finally:
                    await con.close()
            finally:
                self.loop.set_exception_handler(old_handler)


@unittest.skipIf(os.environ.get('PGHOST'), 'unmanaged cluster')