i

test/native/auth/test_scram-sha-256.py

@@ -1,32 +1,38 @@
 import pytest
 
-from pg8000.native import Connection, DatabaseError
+from pg8000.native import Connection, InterfaceError
 
 # This requires a line in pg_hba.conf that requires scram-sha-256 for the
 # database scram_sha_256
 
+DB = "pg8000_scram_sha_256"
 
-def test_scram_sha_256(db_kwargs):
-    db_kwargs["database"] = "pg8000_scram_sha_256"
 
-    # Should only raise an exception saying db doesn't exist
-    with pytest.raises(DatabaseError, match="3D000"):
-        Connection(**db_kwargs)
+@pytest.fixture
+def setup(con):
+    con.run("ALTER SYSTEM SET ssl = off")
+    con.run("SELECT pg_reload_conf()")
+    yield
+    con.run("ALTER SYSTEM SET ssl = on")
+    con.run("SELECT pg_reload_conf()")
+
+
+def test_scram_sha_256(setup, db_kwargs):
+    db_kwargs["database"] = DB
 
+    Connection(**db_kwargs)
 
-def test_scram_sha_256_ssl_False(db_kwargs):
-    db_kwargs["database"] = "pg8000_scram_sha_256"
+
+def test_scram_sha_256_ssl_False(setup, db_kwargs):
+    db_kwargs["database"] = DB
     db_kwargs["ssl_context"] = False
 
-    # Should only raise an exception saying db doesn't exist
-    with pytest.raises(DatabaseError, match="3D000"):
-        Connection(**db_kwargs)
+    Connection(**db_kwargs)
 
 
-def test_scram_sha_256_ssl_True(db_kwargs):
-    db_kwargs["database"] = "pg8000_scram_sha_256"
+def test_scram_sha_256_ssl_True(setup, db_kwargs):
+    db_kwargs["database"] = DB
     db_kwargs["ssl_context"] = True
 
-    # Should only raise an exception saying db doesn't exist
-    with pytest.raises(DatabaseError, match="3D000"):
+    with pytest.raises(InterfaceError, match="Server refuses SSL"):
         Connection(**db_kwargs)

test/native/auth/test_scram-sha-256_ssl.py

@@ -1,21 +1,53 @@
-import ssl
+from ssl import CERT_NONE, SSLSocket, create_default_context
 
 import pytest
 
 from pg8000.native import Connection, DatabaseError
 
 # This requires a line in pg_hba.conf that requires scram-sha-256 for the
-# database scram_sha_256
+# database pg8000_scram_sha_256
+
+DB = "pg8000_scram_sha_256"
+
+
+@pytest.fixture
+def setup(con):
+    try:
+        con.run("CREATE DATABASE {DB}")
+    except DatabaseError:
+        pass
 
 
 def test_scram_sha_256_plus(db_kwargs):
-    context = ssl.create_default_context()
+    db_kwargs["database"] = DB
+
+    con = Connection(**db_kwargs)
+    assert isinstance(con._usock, SSLSocket)
+
+
+def test_scram_sha_256_plus_ssl_True(db_kwargs):
+    db_kwargs["ssl_context"] = True
+    db_kwargs["database"] = DB
+
+    con = Connection(**db_kwargs)
+    assert isinstance(con._usock, SSLSocket)
+
+
+def test_scram_sha_256_plus_ssl_custom(db_kwargs):
+    context = create_default_context()
     context.check_hostname = False
-    context.verify_mode = ssl.CERT_NONE
+    context.verify_mode = CERT_NONE
 
     db_kwargs["ssl_context"] = context
-    db_kwargs["database"] = "pg8000_scram_sha_256"
+    db_kwargs["database"] = DB
+
+    con = Connection(**db_kwargs)
+    assert isinstance(con._usock, SSLSocket)
+
+
+def test_scram_sha_256_plus_ssl_False(db_kwargs):
+    db_kwargs["ssl_context"] = False
+    db_kwargs["database"] = DB
 
-    # Should only raise an exception saying db doesn't exist
-    with pytest.raises(DatabaseError, match="3D000"):
-        Connection(**db_kwargs)
+    con = Connection(**db_kwargs)
+    assert not isinstance(con._usock, SSLSocket)