test: add coverage for dynamic currency detection feature

Author: richardfogaca

tests/integration_tests/viz_tests.py

@@ -179,6 +179,70 @@ def test_cache_timeout(self):
             data_cache_timeout
         )
 
+    def test_detect_currency_returns_none_when_currency_format_not_dict(self):
+        """Test _detect_currency returns None when currency_format is not a dict."""
+        datasource = self.get_datasource_mock()
+        datasource.currency_code_column = "currency_code"
+        form_data = {"currency_format": "invalid"}
+        test_viz = viz.BaseViz(datasource, form_data)
+
+        result = test_viz._detect_currency()
+
+        assert result is None
+
+    def test_detect_currency_returns_none_when_symbol_not_auto(self):
+        """Test _detect_currency returns None when symbol is not AUTO."""
+        datasource = self.get_datasource_mock()
+        datasource.currency_code_column = "currency_code"
+        form_data = {"currency_format": {"symbol": "USD"}}
+        test_viz = viz.BaseViz(datasource, form_data)
+
+        result = test_viz._detect_currency()
+
+        assert result is None
+
+    def test_detect_currency_returns_none_when_no_currency_column(self):
+        """Test _detect_currency returns None when datasource has no currency column."""
+        datasource = self.get_datasource_mock()
+        datasource.currency_code_column = None
+        form_data = {"currency_format": {"symbol": "AUTO"}}
+        test_viz = viz.BaseViz(datasource, form_data)
+
+        result = test_viz._detect_currency()
+
+        assert result is None
+
+    @patch("superset.viz.detect_currency_from_df")
+    def test_detect_currency_uses_dataframe_when_column_present(
+        self, mock_detect_from_df
+    ):
+        """Test _detect_currency uses df when currency column is present."""
+        datasource = self.get_datasource_mock()
+        datasource.currency_code_column = "currency_code"
+        form_data = {"currency_format": {"symbol": "AUTO"}}
+        test_viz = viz.BaseViz(datasource, form_data)
+        df = pd.DataFrame({"currency_code": ["USD", "USD"]})
+        mock_detect_from_df.return_value = "USD"
+
+        result = test_viz._detect_currency(df=df)
+
+        assert result == "USD"
+        mock_detect_from_df.assert_called_once_with(df, "currency_code")
+
+    @patch("superset.viz.detect_currency")
+    def test_detect_currency_queries_datasource_when_no_df(self, mock_detect):
+        """Test _detect_currency queries datasource when df is None."""
+        datasource = self.get_datasource_mock()
+        datasource.currency_code_column = "currency_code"
+        form_data = {"currency_format": {"symbol": "AUTO"}}
+        test_viz = viz.BaseViz(datasource, form_data)
+        mock_detect.return_value = "EUR"
+
+        result = test_viz._detect_currency()
+
+        assert result == "EUR"
+        mock_detect.assert_called_once()
+
 
 class TestPairedTTest(SupersetTestCase):
     def test_get_data_transforms_dataframe(self):

tests/unit_tests/common/test_query_actions_currency.py

@@ -0,0 +1,159 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from unittest.mock import MagicMock, patch
+
+import pandas as pd
+import pytest
+
+from superset.common.query_actions import _detect_currency
+
+
+@pytest.fixture
+def mock_query_context() -> MagicMock:
+    """Create a mock QueryContext with AUTO currency format."""
+    context = MagicMock()
+    context.form_data = {"currency_format": {"symbol": "AUTO"}}
+    return context
+
+
+@pytest.fixture
+def mock_query_obj() -> MagicMock:
+    """Create a mock QueryObject with filter attributes."""
+    obj = MagicMock()
+    obj.filter = []
+    obj.granularity = None
+    obj.from_dttm = None
+    obj.to_dttm = None
+    obj.extras = {}
+    return obj
+
+
+@pytest.fixture
+def mock_datasource() -> MagicMock:
+    """Create a mock datasource with currency column."""
+    ds = MagicMock()
+    ds.currency_code_column = "currency_code"
+    return ds
+
+
+def test_detect_currency_returns_none_when_form_data_is_none(
+    mock_query_obj: MagicMock,
+    mock_datasource: MagicMock,
+) -> None:
+    """Returns None when query context has no form_data."""
+    context = MagicMock()
+    context.form_data = None
+
+    result = _detect_currency(context, mock_query_obj, mock_datasource)
+
+    assert result is None
+
+
+def test_detect_currency_returns_none_when_currency_format_not_dict(
+    mock_query_obj: MagicMock,
+    mock_datasource: MagicMock,
+) -> None:
+    """Returns None when currency_format is not a dict."""
+    context = MagicMock()
+    context.form_data = {"currency_format": "invalid"}
+
+    result = _detect_currency(context, mock_query_obj, mock_datasource)
+
+    assert result is None
+
+
+def test_detect_currency_returns_none_when_symbol_not_auto(
+    mock_query_obj: MagicMock,
+    mock_datasource: MagicMock,
+) -> None:
+    """Returns None when currency_format.symbol is not AUTO."""
+    context = MagicMock()
+    context.form_data = {"currency_format": {"symbol": "USD"}}
+
+    result = _detect_currency(context, mock_query_obj, mock_datasource)
+
+    assert result is None
+
+
+def test_detect_currency_returns_none_when_no_currency_column(
+    mock_query_context: MagicMock,
+    mock_query_obj: MagicMock,
+) -> None:
+    """Returns None when datasource has no currency_code_column."""
+    datasource = MagicMock()
+    datasource.currency_code_column = None
+
+    result = _detect_currency(mock_query_context, mock_query_obj, datasource)
+
+    assert result is None
+
+
+@patch("superset.common.query_actions.detect_currency_from_df")
+def test_detect_currency_uses_dataframe_when_column_present(
+    mock_detect_from_df: MagicMock,
+    mock_query_context: MagicMock,
+    mock_query_obj: MagicMock,
+    mock_datasource: MagicMock,
+) -> None:
+    """Uses detect_currency_from_df when df contains currency column."""
+    df = pd.DataFrame({"currency_code": ["USD", "USD"]})
+    mock_detect_from_df.return_value = "USD"
+
+    result = _detect_currency(mock_query_context, mock_query_obj, mock_datasource, df)
+
+    assert result == "USD"
+    mock_detect_from_df.assert_called_once_with(df, "currency_code")
+
+
+@patch("superset.common.query_actions.detect_currency")
+def test_detect_currency_queries_datasource_when_no_df(
+    mock_detect: MagicMock,
+    mock_query_context: MagicMock,
+    mock_query_obj: MagicMock,
+    mock_datasource: MagicMock,
+) -> None:
+    """Queries datasource when df is None."""
+    mock_detect.return_value = "EUR"
+
+    result = _detect_currency(mock_query_context, mock_query_obj, mock_datasource)
+
+    assert result == "EUR"
+    mock_detect.assert_called_once_with(
+        datasource=mock_datasource,
+        filters=mock_query_obj.filter,
+        granularity=mock_query_obj.granularity,
+        from_dttm=mock_query_obj.from_dttm,
+        to_dttm=mock_query_obj.to_dttm,
+        extras=mock_query_obj.extras,
+    )
+
+
+@patch("superset.common.query_actions.detect_currency")
+def test_detect_currency_queries_datasource_when_column_not_in_df(
+    mock_detect: MagicMock,
+    mock_query_context: MagicMock,
+    mock_query_obj: MagicMock,
+    mock_datasource: MagicMock,
+) -> None:
+    """Falls back to query when df doesn't have currency column."""
+    df = pd.DataFrame({"other_column": ["value"]})
+    mock_detect.return_value = "GBP"
+
+    result = _detect_currency(mock_query_context, mock_query_obj, mock_datasource, df)
+
+    assert result == "GBP"
+    mock_detect.assert_called_once()

tests/unit_tests/common/test_query_context_factory.py

@@ -436,3 +436,97 @@ def test_apply_filters_no_temporal_filters(self):
         self.factory._apply_filters(query_object)
 
         assert query_object.filter[0]["val"] == "value"
+
+    def test_add_currency_column_no_form_data(self):
+        """Test _add_currency_column when form_data is None."""
+        query_object = Mock(spec=QueryObject)
+        query_object.columns = ["col1"]
+        datasource = Mock()
+
+        self.factory._add_currency_column(query_object, None, datasource)
+
+        assert query_object.columns == ["col1"]
+
+    def test_add_currency_column_no_columns(self):
+        """Test _add_currency_column when query_object has no columns."""
+        query_object = Mock(spec=QueryObject)
+        query_object.columns = []
+        form_data = {
+            "viz_type": "pivot_table_v2",
+            "currency_format": {"symbol": "AUTO"},
+        }
+        datasource = Mock()
+        datasource.currency_code_column = "currency_code"
+
+        self.factory._add_currency_column(query_object, form_data, datasource)
+
+        assert query_object.columns == []
+
+    def test_add_currency_column_unsupported_viz_type(self):
+        """Test _add_currency_column with unsupported viz type."""
+        query_object = Mock(spec=QueryObject)
+        query_object.columns = ["col1"]
+        form_data = {"viz_type": "pie", "currency_format": {"symbol": "AUTO"}}
+        datasource = Mock()
+        datasource.currency_code_column = "currency_code"
+
+        self.factory._add_currency_column(query_object, form_data, datasource)
+
+        assert query_object.columns == ["col1"]
+
+    def test_add_currency_column_symbol_not_auto(self):
+        """Test _add_currency_column when symbol is not AUTO."""
+        query_object = Mock(spec=QueryObject)
+        query_object.columns = ["col1"]
+        form_data = {"viz_type": "pivot_table_v2", "currency_format": {"symbol": "USD"}}
+        datasource = Mock()
+        datasource.currency_code_column = "currency_code"
+
+        self.factory._add_currency_column(query_object, form_data, datasource)
+
+        assert query_object.columns == ["col1"]
+
+    def test_add_currency_column_no_currency_column_on_datasource(self):
+        """Test _add_currency_column when datasource has no currency column."""
+        query_object = Mock(spec=QueryObject)
+        query_object.columns = ["col1"]
+        form_data = {
+            "viz_type": "pivot_table_v2",
+            "currency_format": {"symbol": "AUTO"},
+        }
+        datasource = Mock()
+        datasource.currency_code_column = None
+
+        self.factory._add_currency_column(query_object, form_data, datasource)
+
+        assert query_object.columns == ["col1"]
+
+    def test_add_currency_column_already_in_query(self):
+        """Test _add_currency_column when currency column already exists."""
+        query_object = Mock(spec=QueryObject)
+        query_object.columns = ["col1", "currency_code"]
+        form_data = {
+            "viz_type": "pivot_table_v2",
+            "currency_format": {"symbol": "AUTO"},
+        }
+        datasource = Mock()
+        datasource.currency_code_column = "currency_code"
+
+        self.factory._add_currency_column(query_object, form_data, datasource)
+
+        assert query_object.columns == ["col1", "currency_code"]
+
+    def test_add_currency_column_adds_column_for_supported_viz_types(self):
+        """Test _add_currency_column adds column for supported viz types"""
+        for viz_type in ["pivot_table_v2", "table"]:
+            query_object = Mock(spec=QueryObject)
+            query_object.columns = ["col1"]
+            form_data = {"viz_type": viz_type, "currency_format": {"symbol": "AUTO"}}
+            datasource = Mock()
+            datasource.currency_code_column = "currency_code"
+
+            self.factory._add_currency_column(query_object, form_data, datasource)
+
+            assert query_object.columns == ["col1", "currency_code"], (
+                f"Failed for {viz_type}"
+            )

tests/unit_tests/connectors/sqla/models_test.py

@@ -855,3 +855,54 @@ def test_sqla_table_currency_code_column_property() -> None:
         currency_code_column="currency",
     )
     assert table.currency_code_column == "currency"
+
+
+def test_sqla_table_data_includes_currency_code_column(mocker: MockerFixture) -> None:
+    """
+    Test that data property includes currency_code_column.
+    """
+    database = mocker.MagicMock()
+    database.get_sqla_engine.return_value.__enter__ = mocker.MagicMock()
+    database.get_sqla_engine.return_value.__exit__ = mocker.MagicMock()
+
+    table = SqlaTable(
+        table_name="sales",
+        database=database,
+        currency_code_column="currency_code",
+        main_dttm_col="ds",
+    )
+    table.columns = []
+    table.metrics = []
+
+    # Mock the columns property to return empty list
+    mocker.patch.object(SqlaTable, "columns", [])
+    mocker.patch.object(SqlaTable, "metrics", [])
+
+    data = table.data
+    assert data["currency_code_column"] == "currency_code"
+    assert data["main_dttm_col"] == "ds"
+
+
+def test_sqla_table_link_escapes_url(mocker: MockerFixture) -> None:
+    """
+    Test that link property properly escapes URL to prevent XSS.
+    """
+    database = Database(database_name="my_db")
+    table = SqlaTable(
+        table_name='test<script>alert("xss")</script>',
+        database=database,
+        id=1,
+    )
+
+    # Mock explore_url to return a URL with special characters
+    mocker.patch.object(
+        SqlaTable,
+        "explore_url",
+        new_callable=mocker.PropertyMock,
+        return_value='/explore/?datasource_type=table&datasource_id=1&name=<script>alert("xss")</script>',
+    )
+
+    link = table.link
+    # Verify that special characters are escaped in both name and URL
+    assert "&lt;script&gt;" in str(link)
+    assert "<script>" not in str(link)

tests/unit_tests/utils/currency_test.py

@@ -216,3 +216,14 @@ def test_detect_currency_from_df_ignores_null_values() -> None:
     result = detect_currency_from_df(df, "currency_code")
 
     assert result == "USD"
+
+
+def test_detect_currency_returns_none_when_query_not_callable() -> None:
+    """Returns None when datasource query attribute is not callable."""
+    datasource = MagicMock()
+    datasource.currency_code_column = "currency_code"
+    datasource.query = "not_a_callable"  # Set to a string instead of a method
+
+    result = detect_currency(datasource)
+
+    assert result is None