diff --git a/lib/dl_connector_mysql/dl_connector_mysql_tests/db/core/test_sa_dialect.py b/lib/dl_connector_mysql/dl_connector_mysql_tests/db/core/test_sa_dialect.py new file mode 100644 index 000000000..045611c95 --- /dev/null +++ b/lib/dl_connector_mysql/dl_connector_mysql_tests/db/core/test_sa_dialect.py @@ -0,0 +1,40 @@ +import datetime + +import pytest +import sqlalchemy as sa +import sqlalchemy.dialects.mysql +import sqlalchemy.sql.sqltypes + +from dl_connector_mysql_tests.db.core.base import BaseMySQLTestClass + + +class TestMySQLSaDialect(BaseMySQLTestClass): + @pytest.mark.parametrize( + ("value", "type_", "expected"), + ( + pytest.param( + datetime.date(2022, 1, 2), sqlalchemy.sql.sqltypes.Date(), datetime.date(2022, 1, 2), id="date-as-date" + ), + pytest.param("2022-01-02", sqlalchemy.dialects.mysql.DATE(), datetime.date(2022, 1, 2), id="date-as-string"), + pytest.param( + datetime.datetime(2022, 1, 2, 12, 59, 59), + sqlalchemy.sql.sqltypes.DateTime(), + datetime.datetime(2022, 1, 2, 12, 59, 59), + id="datetime-as-datetime", + ), + pytest.param( + "2022-01-02T12:59:59", + sqlalchemy.dialects.mysql.DATETIME(fsp=6), + datetime.datetime(2022, 1, 2, 12, 59, 59), + id="datetime-as-string", + ), + ), + ) + def test_mysql_literal_bind_datetimes(self, value, type_, expected, db): + execute = db.execute + dialect = db._engine_wrapper.dialect + + query = sa.select([sa.literal(value, type_=type_)]) + compiled = str(query.compile(dialect=dialect, compile_kwargs={"literal_binds": True})) + res_literal = list(execute(compiled)) + assert res_literal[0][0] == expected diff --git a/lib/dl_connector_postgresql/dl_connector_postgresql_tests/db/core/test_sa_dialect.py b/lib/dl_connector_postgresql/dl_connector_postgresql_tests/db/core/test_sa_dialect.py new file mode 100644 index 000000000..96313a9fc --- /dev/null +++ b/lib/dl_connector_postgresql/dl_connector_postgresql_tests/db/core/test_sa_dialect.py @@ -0,0 +1,40 @@ +import datetime + +import pytest +import pytz +import sqlalchemy as sa + +from dl_connector_postgresql_tests.db.core.base import BasePostgreSQLTestClass + + +TEST_VALUES = [datetime.date(2020, 1, 1)] + [ + datetime.datetime(2020, idx1 + 1, idx2 + 1, 3, 4, 5, us).replace(tzinfo=tzinfo) + for idx1, us in enumerate((0, 123356)) + for idx2, tzinfo in enumerate( + ( + None, + datetime.timezone.utc, + pytz.timezone("America/New_York"), + ) + ) +] + + +class TestPostgresqlSaDialect(BasePostgreSQLTestClass): + @pytest.mark.parametrize("value", TEST_VALUES, ids=[val.isoformat() for val in TEST_VALUES]) + def test_pg_literal_bind_datetimes(self, value, db): + """ + Test that query results for literal_binds matches the query results without, + for the custom dialect code. + + This test should be in the bi_postgresql dialect itself, but it doesn't have + a postgres-using test at the moment. + """ + execute = db.execute + dialect = db._engine_wrapper.dialect + + query = sa.select([sa.literal(value)]) + compiled = str(query.compile(dialect=dialect, compile_kwargs={"literal_binds": True})) + res_direct = list(execute(query)) + res_literal = list(execute(compiled)) + assert res_direct == res_literal, dict(literal_query=compiled)