diff --git a/tests/integration_tests/model_tests.py b/tests/integration_tests/model_tests.py index 1da5e3d02102..f956286c315a 100644 --- a/tests/integration_tests/model_tests.py +++ b/tests/integration_tests/model_tests.py @@ -395,47 +395,66 @@ class TestSqlaTableModel(SupersetTestCase): def test_get_timestamp_expression(self): tbl = self.get_table(name="birth_names") ds_col = tbl.get_column("ds") - sqla_literal = ds_col.get_timestamp_expression(None) - assert str(sqla_literal.compile()) == "ds" - - sqla_literal = ds_col.get_timestamp_expression("P1D") - compiled = f"{sqla_literal.compile()}" - if tbl.database.backend == "mysql": - assert compiled == "DATE(ds)" - - prev_ds_expr = ds_col.expression - ds_col.expression = "DATE_ADD(ds, 1)" - sqla_literal = ds_col.get_timestamp_expression("P1D") - compiled = f"{sqla_literal.compile()}" - if tbl.database.backend == "mysql": - assert compiled == "DATE(DATE_ADD(ds, 1))" - ds_col.expression = prev_ds_expr + try: + sqla_literal = ds_col.get_timestamp_expression(None) + assert str(sqla_literal.compile()) == "ds" + + sqla_literal = ds_col.get_timestamp_expression("P1D") + compiled = f"{sqla_literal.compile()}" + if tbl.database.backend == "mysql": + assert compiled == "DATE(ds)" + + prev_ds_expr = ds_col.expression + ds_col.expression = "DATE_ADD(ds, 1)" + sqla_literal = ds_col.get_timestamp_expression("P1D") + compiled = f"{sqla_literal.compile()}" + if tbl.database.backend == "mysql": + assert compiled == "DATE(DATE_ADD(ds, 1))" + ds_col.expression = prev_ds_expr + finally: + # Discard the in-memory attribute history so the next session + # autoflush doesn't see this row as dirty. The test only + # exercises the in-memory compile path; any persisted write + # would be accidental. ``rollback`` rather than ``expire`` — + # the latter doesn't reliably clear SA's per-attribute history + # tracking for already-loaded objects. + metadata_db.session.rollback() @pytest.mark.usefixtures("load_birth_names_dashboard_with_slices") def test_get_timestamp_expression_epoch(self): tbl = self.get_table(name="birth_names") ds_col = tbl.get_column("ds") - ds_col.expression = None - ds_col.python_date_format = "epoch_s" - sqla_literal = ds_col.get_timestamp_expression(None) - compiled = f"{sqla_literal.compile()}" - if tbl.database.backend == "mysql": - assert compiled == "from_unixtime(ds)" - - ds_col.python_date_format = "epoch_s" - sqla_literal = ds_col.get_timestamp_expression("P1D") - compiled = f"{sqla_literal.compile()}" - if tbl.database.backend == "mysql": - assert compiled == "DATE(from_unixtime(ds))" - - prev_ds_expr = ds_col.expression - ds_col.expression = "DATE_ADD(ds, 1)" - sqla_literal = ds_col.get_timestamp_expression("P1D") - compiled = f"{sqla_literal.compile()}" - if tbl.database.backend == "mysql": - assert compiled == "DATE(from_unixtime(DATE_ADD(ds, 1)))" - ds_col.expression = prev_ds_expr + try: + ds_col.expression = None + ds_col.python_date_format = "epoch_s" + sqla_literal = ds_col.get_timestamp_expression(None) + compiled = f"{sqla_literal.compile()}" + if tbl.database.backend == "mysql": + assert compiled == "from_unixtime(ds)" + + ds_col.python_date_format = "epoch_s" + sqla_literal = ds_col.get_timestamp_expression("P1D") + compiled = f"{sqla_literal.compile()}" + if tbl.database.backend == "mysql": + assert compiled == "DATE(from_unixtime(ds))" + + prev_ds_expr = ds_col.expression + ds_col.expression = "DATE_ADD(ds, 1)" + sqla_literal = ds_col.get_timestamp_expression("P1D") + compiled = f"{sqla_literal.compile()}" + if tbl.database.backend == "mysql": + assert compiled == "DATE(from_unixtime(DATE_ADD(ds, 1)))" + ds_col.expression = prev_ds_expr + finally: + # Discard the in-memory attribute history so the next session + # autoflush doesn't see this row as dirty — + # ``python_date_format`` isn't restored above and the test + # never commits, so the mutation would otherwise leak. + # ``rollback`` rather than ``expire`` — the latter doesn't + # reliably clear SA's per-attribute history tracking for + # already-loaded objects. + metadata_db.session.rollback() def query_with_expr_helper(self, is_timeseries, inner_join=True): tbl = self.get_table(name="birth_names")