# fixed in mysql-connector as of 2.0.1,
# see http://bugs.mysql.com/bug.php?id=73266
@testing.provide_metadata
- def test_precision_float_roundtrip(self):
+ def test_precision_float_roundtrip(self, connection):
t = Table(
"t",
self.metadata,
mysql.DOUBLE(decimal_return_scale=12, asdecimal=True),
),
)
- with testing.db.connect() as conn:
- t.create(conn)
- conn.execute(
- t.insert(),
- scale_value=45.768392065789,
- unscale_value=45.768392065789,
- )
- result = conn.scalar(select(t.c.scale_value))
- eq_(result, decimal.Decimal("45.768392065789"))
+ t.create(connection)
+ connection.execute(
+ t.insert(),
+ dict(scale_value=45.768392065789, unscale_value=45.768392065789,),
+ )
+ result = connection.scalar(select(t.c.scale_value))
+ eq_(result, decimal.Decimal("45.768392065789"))
- result = conn.scalar(select(t.c.unscale_value))
- eq_(result, decimal.Decimal("45.768392065789"))
+ result = connection.scalar(select(t.c.unscale_value))
+ eq_(result, decimal.Decimal("45.768392065789"))
@testing.only_if("mysql")
- @testing.fails_on("mysql+mysqlconnector", "different unicode behavior")
- @testing.exclude("mysql", "<", (5, 0, 5), "a 5.0+ feature")
@testing.provide_metadata
- def test_charset_collate_table(self):
+ def test_charset_collate_table(self, connection):
t = Table(
"foo",
self.metadata,
mysql_default_charset="utf8",
mysql_collate="utf8_bin",
)
- t.create()
- m2 = MetaData(testing.db)
- t2 = Table("foo", m2, autoload=True)
+ t.create(connection)
+ t2 = Table("foo", MetaData(), autoload_with=connection)
eq_(t2.kwargs["mysql_collate"], "utf8_bin")
eq_(t2.kwargs["mysql_default charset"], "utf8")
# in order to test the condition here, need to use
# MySQLdb 1.2.3 and also need to pass either use_unicode=1
# or charset=utf8 to the URL.
- t.insert().execute(id=1, data=u("some text"))
- assert isinstance(testing.db.scalar(select(t.c.data)), util.text_type)
+ connection.execute(t.insert(), dict(id=1, data=u("some text")))
+ assert isinstance(connection.scalar(select(t.c.data)), util.text_type)
@testing.metadata_fixture(ddl="class")
def bit_table(self, metadata):
([0, 0, 0, 0, 0, 0, 0, l], None),
argnames="store, expected",
)
- def test_bit_50_roundtrip(self, bit_table, store, expected):
+ def test_bit_50_roundtrip(self, connection, bit_table, store, expected):
- meta2 = MetaData(testing.db)
- reflected = Table("mysql_bits", meta2, autoload=True)
+ reflected = Table("mysql_bits", MetaData(), autoload_with=connection)
- with testing.db.connect() as conn:
- expected = expected or store
- conn.execute(reflected.insert(store))
- row = conn.execute(reflected.select()).first()
- eq_(list(row), expected)
- conn.execute(reflected.delete())
+ expected = expected or store
+ connection.execute(reflected.insert().values(store))
+ row = connection.execute(reflected.select()).first()
+ eq_(list(row), expected)
@testing.combinations(
(([0] * 8), None),
([0, 0, 0, 0, 0, 0, 0, l], None),
argnames="store, expected",
)
- def test_bit_50_roundtrip_reflected(self, bit_table, store, expected):
- meta2 = MetaData()
- bit_table = Table("mysql_bits", meta2, autoload_with=testing.db)
+ def test_bit_50_roundtrip_reflected(
+ self, connection, bit_table, store, expected
+ ):
+ bit_table = Table("mysql_bits", MetaData(), autoload_with=connection)
- with testing.db.connect() as conn:
- expected = expected or store
- conn.execute(bit_table.insert(store))
- row = conn.execute(bit_table.select()).first()
- eq_(list(row), expected)
- conn.execute(bit_table.delete())
+ expected = expected or store
+ connection.execute(bit_table.insert().values(store))
+ row = connection.execute(bit_table.select()).first()
+ eq_(list(row), expected)
@testing.metadata_fixture(ddl="class")
def boolean_table(self, metadata):
([False, False, 0, 0, 0], [False, False, 0, 0, 0]),
argnames="store, expected",
)
- def test_boolean_roundtrip(self, boolean_table, store, expected):
+ def test_boolean_roundtrip(
+ self, connection, boolean_table, store, expected
+ ):
table = boolean_table
- with testing.db.connect() as conn:
- expected = expected or store
- conn.execute(table.insert(store))
- row = conn.execute(table.select()).first()
- eq_(list(row), expected)
- for i, val in enumerate(expected):
- if isinstance(val, bool):
- self.assert_(val is row[i])
- conn.execute(table.delete())
+ expected = expected or store
+ connection.execute(table.insert().values(store))
+ row = connection.execute(table.select()).first()
+ eq_(list(row), expected)
+ for i, val in enumerate(expected):
+ if isinstance(val, bool):
+ self.assert_(val is row[i])
@testing.combinations(
([None, None, None, None, None], None),
([False, False, 0, 0, 0], [False, False, False, False, 0]),
argnames="store, expected",
)
- def test_boolean_roundtrip_reflected(self, boolean_table, store, expected):
- meta2 = MetaData(testing.db)
- table = Table("mysql_bool", meta2, autoload=True)
+ def test_boolean_roundtrip_reflected(
+ self, connection, boolean_table, store, expected
+ ):
+ table = Table("mysql_bool", MetaData(), autoload_with=connection)
eq_(colspec(table.c.b3), "b3 TINYINT(1)")
eq_regex(colspec(table.c.b4), r"b4 TINYINT(?:\(1\))? UNSIGNED")
- meta2 = MetaData(testing.db)
table = Table(
"mysql_bool",
- meta2,
+ MetaData(),
Column("b1", BOOLEAN),
Column("b2", Boolean),
Column("b3", BOOLEAN),
Column("b4", BOOLEAN),
- autoload=True,
+ autoload_with=connection,
)
eq_(colspec(table.c.b3), "b3 BOOL")
eq_(colspec(table.c.b4), "b4 BOOL")
- with testing.db.connect() as conn:
- expected = expected or store
- conn.execute(table.insert(store))
- row = conn.execute(table.select()).first()
- eq_(list(row), expected)
- for i, val in enumerate(expected):
- if isinstance(val, bool):
- self.assert_(val is row[i])
- conn.execute(table.delete())
+ expected = expected or store
+ connection.execute(table.insert().values(store))
+ row = connection.execute(table.select()).first()
+ eq_(list(row), expected)
+ for i, val in enumerate(expected):
+ if isinstance(val, bool):
+ self.assert_(val is row[i])
class MyTime(TypeDecorator):
impl = TIMESTAMP
[(now, now), (None, now), (None, now)],
)
- @testing.fails_on("mysql+oursql", "TODO: probable OurSQL bug")
@testing.provide_metadata
- def test_time_roundtrip(self):
+ def test_time_roundtrip(self, connection):
t = Table("mysql_time", self.metadata, Column("t1", mysql.TIME()))
- with testing.db.connect() as conn:
- t.create(conn)
+ t.create(connection)
- conn.execute(t.insert().values(t1=datetime.time(8, 37, 35)))
- eq_(
- conn.execute(select(t.c.t1)).scalar(),
- datetime.time(8, 37, 35),
- )
+ connection.execute(t.insert().values(t1=datetime.time(8, 37, 35)))
+ eq_(
+ connection.execute(select(t.c.t1)).scalar(),
+ datetime.time(8, 37, 35),
+ )
@testing.provide_metadata
- def test_year(self):
+ def test_year(self, connection):
"""Exercise YEAR."""
year_table = Table(
for col in year_table.c:
self.assert_(repr(col))
- year_table.create()
- reflected = Table("mysql_year", MetaData(testing.db), autoload=True)
+ year_table.create(connection)
+ reflected = Table("mysql_year", MetaData(), autoload_with=connection)
for table in year_table, reflected:
- with testing.db.connect() as conn:
- conn.execute(table.insert(["1950", "50", None, 1950]))
- row = conn.execute(table.select()).first()
- eq_(list(row), [1950, 2050, None, 1950])
- conn.execute(table.delete())
- self.assert_(colspec(table.c.y1).startswith("y1 YEAR"))
- eq_regex(colspec(table.c.y5), r"y5 YEAR(?:\(4\))?")
+ connection.execute(
+ table.insert().values(["1950", "50", None, 1950])
+ )
+ row = connection.execute(table.select()).first()
+ eq_(list(row), [1950, 2050, None, 1950])
+ self.assert_(colspec(table.c.y1).startswith("y1 YEAR"))
+ eq_regex(colspec(table.c.y5), r"y5 YEAR(?:\(4\))?")
class JSONTest(fixtures.TestBase):
@testing.provide_metadata
@testing.requires.reflects_json_type
- def test_reflection(self):
+ def test_reflection(self, connection):
Table("mysql_json", self.metadata, Column("foo", mysql.JSON))
- self.metadata.create_all()
+ self.metadata.create_all(connection)
- reflected = Table("mysql_json", MetaData(), autoload_with=testing.db)
+ reflected = Table("mysql_json", MetaData(), autoload_with=connection)
is_(reflected.c.foo.type._type_affinity, sqltypes.JSON)
assert isinstance(reflected.c.foo.type, mysql.JSON)
@testing.provide_metadata
- def test_rudimental_round_trip(self):
+ def test_rudimental_round_trip(self, connection):
# note that test_suite has many more JSON round trip tests
# using the backend-agnostic JSON type
mysql_json = Table(
"mysql_json", self.metadata, Column("foo", mysql.JSON)
)
- self.metadata.create_all()
+ self.metadata.create_all(connection)
value = {"json": {"foo": "bar"}, "recs": ["one", "two"]}
- with testing.db.connect() as conn:
- conn.execute(mysql_json.insert(), foo=value)
+ connection.execute(mysql_json.insert(), dict(foo=value))
- eq_(conn.scalar(select(mysql_json.c.foo)), value)
+ eq_(connection.scalar(select(mysql_json.c.foo)), value)
class EnumSetTest(
return [str(v.value) for v in some_enum.__members__.values()]
@testing.provide_metadata
- def test_enum(self):
+ def test_enum(self, connection):
"""Exercise the ENUM type."""
e1 = mysql.ENUM("a", "b")
colspec(enum_table.c.e8),
"e8 ENUM('one','two','three','AMember','BMember')",
)
- enum_table.create()
-
- assert_raises(
- exc.DBAPIError,
- enum_table.insert().execute,
- e1=None,
- e2=None,
- e3=None,
- e4=None,
- )
+ enum_table.create(connection)
assert enum_table.c.e2generic.type.validate_strings
+
assert_raises(
exc.StatementError,
- enum_table.insert().execute,
- e1="c",
- e2="c",
- e2generic="c",
- e3="c",
- e4="c",
- e5="c",
- e5generic="c",
- e6="c",
- e7="c",
- e8="c",
+ connection.execute,
+ enum_table.insert(),
+ dict(
+ e1="c",
+ e2="c",
+ e2generic="c",
+ e3="c",
+ e4="c",
+ e5="c",
+ e5generic="c",
+ e6="c",
+ e7="c",
+ e8="c",
+ ),
)
- enum_table.insert().execute()
- enum_table.insert().execute(
- e1="a",
- e2="a",
- e2generic="a",
- e3="a",
- e4="a",
- e5="a",
- e5generic="a",
- e6="'a'",
- e7="a",
- e8="AMember",
+ connection.execute(enum_table.insert())
+ connection.execute(
+ enum_table.insert(),
+ dict(
+ e1="a",
+ e2="a",
+ e2generic="a",
+ e3="a",
+ e4="a",
+ e5="a",
+ e5generic="a",
+ e6="'a'",
+ e7="a",
+ e8="AMember",
+ ),
)
- enum_table.insert().execute(
- e1="b",
- e2="b",
- e2generic="b",
- e3="b",
- e4="b",
- e5="b",
- e5generic="b",
- e6="b",
- e7="b",
- e8="BMember",
+ connection.execute(
+ enum_table.insert(),
+ dict(
+ e1="b",
+ e2="b",
+ e2generic="b",
+ e3="b",
+ e4="b",
+ e5="b",
+ e5generic="b",
+ e6="b",
+ e7="b",
+ e8="BMember",
+ ),
)
- res = enum_table.select().execute().fetchall()
+ res = connection.execute(enum_table.select()).fetchall()
expected = [
(None, "a", "a", None, "a", None, None, None, None, None),
eq_(colspec(set_table.c.e5), "e5 SET('a','b')")
@testing.provide_metadata
- def test_no_null(self):
+ def test_no_null(self, connection):
set_table = self._set_fixture_one()
- set_table.create()
+ set_table.create(connection)
assert_raises(
exc.DBAPIError,
- set_table.insert().execute,
- e1=None,
- e2=None,
- e3=None,
- e4=None,
- )
-
- @testing.only_on("+oursql")
- @testing.provide_metadata
- def test_oursql_error_one(self):
- set_table = self._set_fixture_one()
- set_table.create()
- assert_raises(
- exc.StatementError,
- set_table.insert().execute,
- e1="c",
- e2="c",
- e3="c",
- e4="c",
+ connection.execute,
+ set_table.insert(),
+ dict(e1=None, e2=None, e3=None, e4=None),
)
@testing.requires.mysql_non_strict
@testing.provide_metadata
- def test_empty_set_no_empty_string(self):
+ def test_empty_set_no_empty_string(self, connection):
t = Table(
"t",
self.metadata,
Column("id", Integer),
Column("data", mysql.SET("a", "b")),
)
- t.create()
- with testing.db.begin() as conn:
- conn.execute(
- t.insert(),
+ t.create(connection)
+ connection.execute(
+ t.insert(),
+ [
{"id": 1, "data": set()},
{"id": 2, "data": set([""])},
{"id": 3, "data": set(["a", ""])},
{"id": 4, "data": set(["b"])},
- )
- eq_(
- conn.execute(t.select().order_by(t.c.id)).fetchall(),
- [(1, set()), (2, set()), (3, set(["a"])), (4, set(["b"]))],
- )
+ ],
+ )
+ eq_(
+ connection.execute(t.select().order_by(t.c.id)).fetchall(),
+ [(1, set()), (2, set()), (3, set(["a"])), (4, set(["b"]))],
+ )
def test_bitwise_required_for_empty(self):
assert_raises_message(
)
@testing.provide_metadata
- def test_empty_set_empty_string(self):
+ def test_empty_set_empty_string(self, connection):
t = Table(
"t",
self.metadata,
Column("id", Integer),
Column("data", mysql.SET("a", "b", "", retrieve_as_bitwise=True)),
)
- t.create()
- with testing.db.begin() as conn:
- conn.execute(
- t.insert(),
+ t.create(connection)
+ connection.execute(
+ t.insert(),
+ [
{"id": 1, "data": set()},
{"id": 2, "data": set([""])},
{"id": 3, "data": set(["a", ""])},
{"id": 4, "data": set(["b"])},
- )
- eq_(
- conn.execute(t.select().order_by(t.c.id)).fetchall(),
- [
- (1, set()),
- (2, set([""])),
- (3, set(["a", ""])),
- (4, set(["b"])),
- ],
- )
+ ],
+ )
+ eq_(
+ connection.execute(t.select().order_by(t.c.id)).fetchall(),
+ [
+ (1, set()),
+ (2, set([""])),
+ (3, set(["a", ""])),
+ (4, set(["b"])),
+ ],
+ )
@testing.provide_metadata
- def test_string_roundtrip(self):
+ def test_string_roundtrip(self, connection):
set_table = self._set_fixture_one()
- set_table.create()
- with testing.db.begin() as conn:
- conn.execute(
- set_table.insert(),
- dict(e1="a", e2="a", e3="a", e4="'a'", e5="a,b"),
- )
- conn.execute(
- set_table.insert(),
- dict(e1="b", e2="b", e3="b", e4="b", e5="a,b"),
- )
+ set_table.create(connection)
+ connection.execute(
+ set_table.insert(),
+ dict(e1="a", e2="a", e3="a", e4="'a'", e5="a,b"),
+ )
+ connection.execute(
+ set_table.insert(), dict(e1="b", e2="b", e3="b", e4="b", e5="a,b"),
+ )
- expected = [
- (
- set(["a"]),
- set(["a"]),
- set(["a"]),
- set(["'a'"]),
- set(["a", "b"]),
- ),
- (
- set(["b"]),
- set(["b"]),
- set(["b"]),
- set(["b"]),
- set(["a", "b"]),
- ),
- ]
- res = conn.execute(set_table.select()).fetchall()
+ expected = [
+ (
+ set(["a"]),
+ set(["a"]),
+ set(["a"]),
+ set(["'a'"]),
+ set(["a", "b"]),
+ ),
+ (set(["b"]), set(["b"]), set(["b"]), set(["b"]), set(["a", "b"]),),
+ ]
+ res = connection.execute(set_table.select()).fetchall()
- eq_(res, expected)
+ eq_(res, expected)
@testing.provide_metadata
- def test_unicode_roundtrip(self):
+ def test_unicode_roundtrip(self, connection):
set_table = Table(
"t",
self.metadata,
Column("data", mysql.SET(u("réveillé"), u("drôle"), u("S’il"))),
)
- set_table.create()
- with testing.db.begin() as conn:
- conn.execute(
- set_table.insert(), {"data": set([u("réveillé"), u("drôle")])}
- )
+ set_table.create(connection)
+ connection.execute(
+ set_table.insert(), {"data": set([u("réveillé"), u("drôle")])}
+ )
- row = conn.execute(set_table.select()).first()
+ row = connection.execute(set_table.select()).first()
- eq_(row, (1, set([u("réveillé"), u("drôle")])))
+ eq_(row, (1, set([u("réveillé"), u("drôle")])))
@testing.provide_metadata
- def test_int_roundtrip(self):
+ def test_int_roundtrip(self, connection):
set_table = self._set_fixture_one()
- set_table.create()
- with testing.db.begin() as conn:
- conn.execute(
- set_table.insert(), dict(e1=1, e2=2, e3=3, e4=3, e5=0)
- )
- res = conn.execute(set_table.select()).first()
- eq_(
- res,
- (
- set(["a"]),
- set(["b"]),
- set(["a", "b"]),
- set(["'a'", "b"]),
- set([]),
- ),
- )
+ set_table.create(connection)
+ connection.execute(
+ set_table.insert(), dict(e1=1, e2=2, e3=3, e4=3, e5=0)
+ )
+ res = connection.execute(set_table.select()).first()
+ eq_(
+ res,
+ (
+ set(["a"]),
+ set(["b"]),
+ set(["a", "b"]),
+ set(["'a'", "b"]),
+ set([]),
+ ),
+ )
@testing.provide_metadata
- def test_set_roundtrip_plus_reflection(self):
+ def test_set_roundtrip_plus_reflection(self, connection):
set_table = Table(
"mysql_set",
self.metadata,
eq_(colspec(set_table.c.s1), "s1 SET('dq','sq')")
eq_(colspec(set_table.c.s2), "s2 SET('a')")
eq_(colspec(set_table.c.s3), "s3 SET('5','7','9')")
- set_table.create()
- reflected = Table("mysql_set", MetaData(testing.db), autoload=True)
+ set_table.create(connection)
+ reflected = Table("mysql_set", MetaData(), autoload_with=connection)
for table in set_table, reflected:
def roundtrip(store, expected=None):
expected = expected or store
- table.insert(store).execute()
- row = table.select().execute().first()
+ connection.execute(table.insert().values(store))
+ row = connection.execute(table.select()).first()
eq_(row, tuple(expected))
- table.delete().execute()
+ connection.execute(table.delete())
roundtrip([None, None, None], [None] * 3)
roundtrip(["", "", ""], [set([])] * 3)
roundtrip(["dq", "a", "5"], [set(["dq"]), set(["a"]), set(["5"])])
roundtrip([1, 1, 1], [set(["dq"]), set(["a"]), set(["5"])])
roundtrip([set(["dq", "sq"]), None, set(["9", "5", "7"])])
- set_table.insert().execute(
- {"s3": set(["5"])},
- {"s3": set(["5", "7"])},
- {"s3": set(["5", "7", "9"])},
- {"s3": set(["7", "9"])},
+ connection.execute(
+ set_table.insert(),
+ [
+ {"s3": set(["5"])},
+ {"s3": set(["5", "7"])},
+ {"s3": set(["5", "7", "9"])},
+ {"s3": set(["7", "9"])},
+ ],
)
- rows = (
- select(
- [set_table.c.s3], set_table.c.s3.in_([set(["5"]), ["5", "7"]])
+ rows = connection.execute(
+ select(set_table.c.s3).where(
+ set_table.c.s3.in_([set(["5"]), ["5", "7"]])
)
- .execute()
- .fetchall()
- )
- found = set([frozenset(row[0]) for row in rows])
- eq_(found, set([frozenset(["5"]), frozenset(["5", "7"])]))
+ ).fetchall()
+
+ eq_(list(rows), [({"5"},), ({"7", "5"},)])
@testing.provide_metadata
- def test_unicode_enum(self):
+ def test_unicode_enum(self, connection):
metadata = self.metadata
t1 = Table(
"table",
Column("value", Enum(u("réveillé"), u("drôle"), u("S’il"))),
Column("value2", mysql.ENUM(u("réveillé"), u("drôle"), u("S’il"))),
)
- metadata.create_all()
- t1.insert().execute(value=u("drôle"), value2=u("drôle"))
- t1.insert().execute(value=u("réveillé"), value2=u("réveillé"))
- t1.insert().execute(value=u("S’il"), value2=u("S’il"))
+ metadata.create_all(connection)
+
+ connection.execute(
+ t1.insert(),
+ [
+ dict(value=u("drôle"), value2=u("drôle")),
+ dict(value=u("réveillé"), value2=u("réveillé")),
+ dict(value=u("S’il"), value2=u("S’il")),
+ ],
+ )
eq_(
- t1.select().order_by(t1.c.id).execute().fetchall(),
+ connection.execute(t1.select().order_by(t1.c.id)).fetchall(),
[
(1, u("drôle"), u("drôle")),
(2, u("réveillé"), u("réveillé")),
# test reflection of the enum labels
- m2 = MetaData(testing.db)
- t2 = Table("table", m2, autoload=True)
+ t2 = Table("table", MetaData(), autoload_with=connection)
# TODO: what's wrong with the last element ? is there
# latin-1 stuff forcing its way in ?
)
@testing.provide_metadata
- @testing.exclude("mysql", "<", (4,), "3.23 can't handle an ENUM of ''")
- def test_enum_parse(self):
+ def test_enum_parse(self, connection):
enum_table = Table(
"mysql_enum",
for col in enum_table.c:
self.assert_(repr(col))
- enum_table.create()
- reflected = Table("mysql_enum", MetaData(testing.db), autoload=True)
+ enum_table.create(connection)
+ reflected = Table("mysql_enum", MetaData(), autoload_with=connection)
for t in enum_table, reflected:
eq_(t.c.e1.type.enums, ["a"])
eq_(t.c.e2.type.enums, [""])
eq_(t.c.e7.type.enums, ["", "'a'", "b'b", "'"])
@testing.provide_metadata
- @testing.exclude("mysql", "<", (5,))
- def test_set_parse(self):
+ def test_set_parse(self, connection):
set_table = Table(
"mysql_set",
self.metadata,
for col in set_table.c:
self.assert_(repr(col))
- set_table.create()
+ set_table.create(connection)
# don't want any warnings on reflection
- reflected = Table("mysql_set", MetaData(testing.db), autoload=True)
+ reflected = Table("mysql_set", MetaData(), autoload_with=connection)
for t in set_table, reflected:
eq_(t.c.e1.type.values, ("a",))
eq_(t.c.e2.type.values, ("",))
@testing.requires.mysql_non_strict
@testing.provide_metadata
- def test_broken_enum_returns_blanks(self):
+ def test_broken_enum_returns_blanks(self, connection):
t = Table(
"enum_missing",
self.metadata,
Column("e1", sqltypes.Enum("one", "two", "three")),
Column("e2", mysql.ENUM("one", "two", "three")),
)
- t.create()
+ t.create(connection)
- with testing.db.connect() as conn:
- conn.execute(
- t.insert(), {"e1": "nonexistent", "e2": "nonexistent"}
- )
- conn.execute(t.insert(), {"e1": "", "e2": ""})
- conn.execute(t.insert(), {"e1": "two", "e2": "two"})
- conn.execute(t.insert(), {"e1": None, "e2": None})
+ connection.execute(
+ t.insert(),
+ [
+ {"e1": "nonexistent", "e2": "nonexistent"},
+ {"e1": "", "e2": ""},
+ {"e1": "two", "e2": "two"},
+ {"e1": None, "e2": None},
+ ],
+ )
- eq_(
- conn.execute(
- select(t.c.e1, t.c.e2).order_by(t.c.id)
- ).fetchall(),
- [("", ""), ("", ""), ("two", "two"), (None, None)],
- )
+ eq_(
+ connection.execute(
+ select(t.c.e1, t.c.e2).order_by(t.c.id)
+ ).fetchall(),
+ [("", ""), ("", ""), ("two", "two"), (None, None)],
+ )
def colspec(c):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
