Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
89 changes: 54 additions & 35 deletions tests/integration_tests/model_tests.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down
Loading