diff --git a/src/MySQLdb/_mysql.c b/src/MySQLdb/_mysql.c index 97914820..8d832bfd 100644 --- a/src/MySQLdb/_mysql.c +++ b/src/MySQLdb/_mysql.c @@ -2968,9 +2968,8 @@ _mysql_ConnectionObject_getattro( _mysql_ConnectionObject *self, PyObject *name) { - const char *cname; - cname = PyUnicode_AsUTF8(name); - if (strcmp(cname, "closed") == 0) + if (PyUnicode_Check(name) && + PyUnicode_CompareWithASCIIString(name, "closed") == 0) return PyLong_FromLong((long)!(self->open)); return PyObject_GenericGetAttr((PyObject *)self, name); diff --git a/tests/test__mysql.py b/tests/test__mysql.py index aea9db71..0ef1fb94 100644 --- a/tests/test__mysql.py +++ b/tests/test__mysql.py @@ -1,8 +1,36 @@ import pytest from MySQLdb import _mysql +from MySQLdb.connections import Connection def test_result_type(): with pytest.raises(TypeError): _mysql.result(b"xyz") + + +@pytest.mark.parametrize("connection_type", [_mysql.connection, Connection]) +@pytest.mark.parametrize("name", [5, None, b"closed", [], {}]) +def test_connection_getattribute_non_string(connection_type, name): + conn = connection_type.__new__(connection_type) + with pytest.raises(TypeError): + conn.__getattribute__(name) + + +@pytest.mark.parametrize("connection_type", [_mysql.connection, Connection]) +@pytest.mark.parametrize("name", ["missing", "closed\x00suffix", "\ud800"]) +def test_connection_getattribute_missing(connection_type, name): + conn = connection_type.__new__(connection_type) + with pytest.raises(AttributeError): + conn.__getattribute__(name) + + +@pytest.mark.parametrize("connection_type", [_mysql.connection, Connection]) +def test_connection_getattribute(connection_type): + class AttributeName(str): + pass + + conn = connection_type.__new__(connection_type) + assert conn.closed == 1 + assert conn.__getattribute__(AttributeName("closed")) == 1 + assert conn.__class__ is connection_type