SSL: Add software tests validating ssl=true vs. sslmode=require

Author: amotl

tests/connection_test.py

@@ -18,25 +18,34 @@
 # However, if you have executed another commercial license agreement
 # with Crate these terms will supersede the license and you may use the
 # software solely pursuant to the terms of the relevant commercial agreement.
-
+import contextlib
+import warnings
 from unittest import TestCase
 
+import pytest
 import sqlalchemy as sa
 from sqlalchemy.exc import NoSuchModuleError
 
+from sqlalchemy_cratedb import SA_1_4, SA_VERSION
+from tests.util import ExtraAssertions
+
 
-class SqlAlchemyConnectionTest(TestCase):
+class SqlAlchemyConnectionTest(TestCase, ExtraAssertions):
     def test_connection_server_uri_unknown_sa_plugin(self):
         with self.assertRaises(NoSuchModuleError):
             sa.create_engine("foobar://otherhost:19201")
 
-    def test_default_connection(self):
+    def test_connection_no_hostname_no_ssl(self):
         engine = sa.create_engine("crate://")
-        conn = engine.raw_connection()
-        self.assertEqual(
-            "<Connection <Client ['http://127.0.0.1:4200']>>", repr(conn.driver_connection)
-        )
-        conn.close()
+        servers = engine.raw_connection().driver_connection.client._active_servers
+        self.assertEqual(["http://127.0.0.1:4200"], servers)
+        engine.dispose()
+
+    @pytest.mark.skipif(SA_VERSION < SA_1_4, reason="Not supported by SQLAlchemy 1.3")
+    def test_connection_no_hostname_with_ssl(self):
+        engine = sa.create_engine("crate://?sslmode=require")
+        servers = engine.raw_connection().driver_connection.client._active_servers
+        self.assertEqual(["https://127.0.0.1:4200"], servers)
         engine.dispose()
 
     def test_connection_server_uri_http(self):
@@ -48,13 +57,51 @@ def test_connection_server_uri_http(self):
         conn.close()
         engine.dispose()
 
-    def test_connection_server_uri_https(self):
-        engine = sa.create_engine("crate://otherhost:19201/?ssl=true")
-        conn = engine.raw_connection()
-        self.assertEqual(
-            "<Connection <Client ['https://otherhost:19201']>>", repr(conn.driver_connection)
-        )
-        conn.close()
+    @contextlib.contextmanager
+    def verify_user_warning_about_ssl_deprecation(self):
+        """
+        The `ssl=true` option was flagged for deprecation. Verify that.
+        """
+        with warnings.catch_warnings(record=True) as w:
+            # Cause all warnings to always be triggered.
+            warnings.simplefilter("always")
+
+            # Run workhorse body.
+            yield
+
+            # Verify details of the deprecation warning.
+            self.assertEqual(len(w), 1)
+            self.assertIsSubclass(w[-1].category, DeprecationWarning)
+            self.assertIn(
+                "The `ssl=true` option will be deprecated, "
+                "please use `sslmode=require` going forward.",
+                str(w[-1].message),
+            )
+
+    def test_connection_server_uri_https_ssl_enabled(self):
+        with self.verify_user_warning_about_ssl_deprecation():
+            engine = sa.create_engine("crate://otherhost:19201/?ssl=true")
+            servers = engine.raw_connection().driver_connection.client._active_servers
+        self.assertEqual(["https://otherhost:19201"], servers)
+        engine.dispose()
+
+    def test_connection_server_uri_https_ssl_disabled(self):
+        with self.verify_user_warning_about_ssl_deprecation():
+            engine = sa.create_engine("crate://otherhost:19201/?ssl=false")
+            servers = engine.raw_connection().driver_connection.client._active_servers
+        self.assertEqual(["http://otherhost:19201"], servers)
+        engine.dispose()
+
+    def test_connection_server_uri_https_sslmode_enabled(self):
+        engine = sa.create_engine("crate://otherhost:19201/?sslmode=require")
+        servers = engine.raw_connection().driver_connection.client._active_servers
+        self.assertEqual(["https://otherhost:19201"], servers)
+        engine.dispose()
+
+    def test_connection_server_uri_https_sslmode_disabled(self):
+        engine = sa.create_engine("crate://otherhost:19201/?sslmode=disable")
+        servers = engine.raw_connection().driver_connection.client._active_servers
+        self.assertEqual(["http://otherhost:19201"], servers)
         engine.dispose()
 
     def test_connection_server_uri_invalid_port(self):