Fix bugs

Author: passren

pydynamodb/__init__.py

@@ -6,7 +6,7 @@
 if TYPE_CHECKING:
     from .connection import Connection
 
-__version__: str = "0.8.1"
+__version__: str = "0.8.2"
 
 # Globals https://www.python.org/dev/peps/pep-0249/#globals
 apilevel: str = "2.0"

pydynamodb/sqlalchemy_dynamodb/pydynamodb.py

@@ -712,7 +712,22 @@ def _create_connect_args(self, url):
         return opts
 
     def do_ping(self, dbapi_connection):
-        return dbapi_connection.test_connection()
+        connection = dbapi_connection
+
+        for attr_name in ("driver_connection", "dbapi_connection", "connection"):
+            nested_connection = getattr(connection, attr_name, None)
+            if nested_connection is not None:
+                connection = nested_connection
+                break
+
+        test_connection = getattr(connection, "test_connection", None)
+        if not callable(test_connection):
+            return False
+
+        try:
+            return bool(test_connection())
+        except Exception:
+            return False
 
     def get_schema_names(self, connection, **kw):
         # DynamoDB does not have the concept of a schema

tests/test_sqlalchemy_dynamodb.py

@@ -40,6 +40,35 @@ def test_ping(self, engine):
         conn = engine.raw_connection().dbapi_connection
         assert engine.dialect.do_ping(conn)
 
+    def test_ping_returns_false_without_test_connection(self, engine):
+        engine, _ = engine
+
+        class ConnectionWithoutPing:
+            pass
+
+        assert not engine.dialect.do_ping(ConnectionWithoutPing())
+
+    def test_ping_uses_wrapped_driver_connection(self, engine):
+        engine, _ = engine
+
+        class WrappedConnection:
+            def __init__(self):
+                self.driver_connection = self
+
+            def test_connection(self):
+                return True
+
+        assert engine.dialect.do_ping(WrappedConnection())
+
+    def test_ping_returns_false_when_test_connection_raises(self, engine):
+        engine, _ = engine
+
+        class FailingConnection:
+            def test_connection(self):
+                raise RuntimeError("ping failed")
+
+        assert not engine.dialect.do_ping(FailingConnection())
+
     def test_basic_insert(self, engine):
         engine, conn = engine