superset2/tests/unit_tests/semantic_layers/models_test.py

# 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.

"""Tests for semantic layer models."""

from __future__ import annotations

import uuid
from typing import Any
from unittest.mock import MagicMock, patch

import pyarrow as pa
import pytest
from superset_core.semantic_layers.types import (
    Dimension,
    Grains,
    Metric,
)

from superset.semantic_layers.models import (
    ColumnMetadata,
    get_column_type,
    MetricMetadata,
    SemanticLayer,
    SemanticView,
)
from superset.utils.core import GenericDataType

# =============================================================================
# get_column_type tests
# =============================================================================


def test_get_column_type_temporal_date() -> None:
    """Test that date types map to TEMPORAL."""
    assert get_column_type(pa.date32()) == GenericDataType.TEMPORAL
    assert get_column_type(pa.date64()) == GenericDataType.TEMPORAL


def test_get_column_type_temporal_timestamp() -> None:
    """Test that timestamp types map to TEMPORAL."""
    assert get_column_type(pa.timestamp("us")) == GenericDataType.TEMPORAL


def test_get_column_type_temporal_time() -> None:
    """Test that time types map to TEMPORAL."""
    assert get_column_type(pa.time64("us")) == GenericDataType.TEMPORAL
    assert get_column_type(pa.time32("ms")) == GenericDataType.TEMPORAL


def test_get_column_type_numeric_integer() -> None:
    """Test that integer types map to NUMERIC."""
    assert get_column_type(pa.int64()) == GenericDataType.NUMERIC
    assert get_column_type(pa.int32()) == GenericDataType.NUMERIC


def test_get_column_type_numeric_float() -> None:
    """Test that float types map to NUMERIC."""
    assert get_column_type(pa.float64()) == GenericDataType.NUMERIC


def test_get_column_type_numeric_decimal() -> None:
    """Test that decimal types map to NUMERIC."""
    assert get_column_type(pa.decimal128(38, 10)) == GenericDataType.NUMERIC


def test_get_column_type_numeric_duration() -> None:
    """Test that duration types map to NUMERIC."""
    assert get_column_type(pa.duration("us")) == GenericDataType.NUMERIC


def test_get_column_type_boolean() -> None:
    """Test that boolean types map to BOOLEAN."""
    assert get_column_type(pa.bool_()) == GenericDataType.BOOLEAN


def test_get_column_type_string() -> None:
    """Test that string types map to STRING."""
    assert get_column_type(pa.utf8()) == GenericDataType.STRING
    assert get_column_type(pa.large_utf8()) == GenericDataType.STRING


def test_get_column_type_binary() -> None:
    """Test that binary types map to STRING."""
    assert get_column_type(pa.binary()) == GenericDataType.STRING


def test_get_column_type_unknown() -> None:
    """Test that unknown types default to STRING."""
    assert get_column_type(pa.null()) == GenericDataType.STRING


# =============================================================================
# MetricMetadata tests
# =============================================================================


def test_metric_metadata_required_fields() -> None:
    """Test MetricMetadata with required fields only."""
    metadata = MetricMetadata(
        metric_name="revenue",
        expression="SUM(amount)",
    )
    assert metadata.metric_name == "revenue"
    assert metadata.expression == "SUM(amount)"
    assert metadata.verbose_name is None
    assert metadata.description is None
    assert metadata.d3format is None
    assert metadata.currency is None
    assert metadata.warning_text is None
    assert metadata.certified_by is None
    assert metadata.certification_details is None


def test_metric_metadata_all_fields() -> None:
    """Test MetricMetadata with all fields."""
    metadata = MetricMetadata(
        metric_name="revenue",
        expression="SUM(amount)",
        verbose_name="Total Revenue",
        description="Sum of all revenue",
        d3format="$,.2f",
        currency={"symbol": "$", "symbolPosition": "prefix"},
        warning_text="Data may be incomplete",
        certified_by="Data Team",
        certification_details="Verified Q1 2024",
    )
    assert metadata.metric_name == "revenue"
    assert metadata.expression == "SUM(amount)"
    assert metadata.verbose_name == "Total Revenue"
    assert metadata.description == "Sum of all revenue"
    assert metadata.d3format == "$,.2f"
    assert metadata.currency == {"symbol": "$", "symbolPosition": "prefix"}
    assert metadata.warning_text == "Data may be incomplete"
    assert metadata.certified_by == "Data Team"
    assert metadata.certification_details == "Verified Q1 2024"


# =============================================================================
# ColumnMetadata tests
# =============================================================================


def test_column_metadata_required_fields() -> None:
    """Test ColumnMetadata with required fields only."""
    metadata = ColumnMetadata(
        column_name="order_date",
        type="DATE",
        is_dttm=True,
    )
    assert metadata.column_name == "order_date"
    assert metadata.type == "DATE"
    assert metadata.is_dttm is True
    assert metadata.verbose_name is None
    assert metadata.description is None
    assert metadata.groupby is True
    assert metadata.filterable is True
    assert metadata.expression is None
    assert metadata.python_date_format is None
    assert metadata.advanced_data_type is None
    assert metadata.extra is None


def test_column_metadata_all_fields() -> None:
    """Test ColumnMetadata with all fields."""
    metadata = ColumnMetadata(
        column_name="order_date",
        type="DATE",
        is_dttm=True,
        verbose_name="Order Date",
        description="Date of the order",
        groupby=True,
        filterable=True,
        expression="DATE(order_timestamp)",
        python_date_format="%Y-%m-%d",
        advanced_data_type="date",
        extra='{"grain": "day"}',
    )
    assert metadata.column_name == "order_date"
    assert metadata.type == "DATE"
    assert metadata.is_dttm is True
    assert metadata.verbose_name == "Order Date"
    assert metadata.description == "Date of the order"
    assert metadata.groupby is True
    assert metadata.filterable is True
    assert metadata.expression == "DATE(order_timestamp)"
    assert metadata.python_date_format == "%Y-%m-%d"
    assert metadata.advanced_data_type == "date"
    assert metadata.extra == '{"grain": "day"}'


# =============================================================================
# SemanticLayer tests
# =============================================================================


def test_semantic_layer_repr_with_name() -> None:
    """Test SemanticLayer __repr__ with name."""
    layer = SemanticLayer()
    layer.name = "My Semantic Layer"
    layer.uuid = uuid.uuid4()
    assert repr(layer) == "My Semantic Layer"


def test_semantic_layer_repr_without_name() -> None:
    """Test SemanticLayer __repr__ without name (uses uuid)."""
    layer = SemanticLayer()
    layer.name = None
    test_uuid = uuid.uuid4()
    layer.uuid = test_uuid
    assert repr(layer) == str(test_uuid)


def test_semantic_layer_implementation_not_implemented() -> None:
    """Test that implementation raises KeyError for unregistered type."""
    layer = SemanticLayer()
    with pytest.raises(KeyError):
        _ = layer.implementation


def test_semantic_layer_implementation() -> None:
    """Test that implementation returns a configured semantic layer."""
    layer = SemanticLayer()
    layer.type = "test_type"
    layer.configuration = '{"key": "value"}'

    mock_class = MagicMock()
    mock_impl = MagicMock()
    mock_class.from_configuration.return_value = mock_impl

    with patch.dict(
        "superset.semantic_layers.models.registry",
        {"test_type": mock_class},
    ):
        # Clear cached property if it exists
        if "implementation" in layer.__dict__:
            del layer.__dict__["implementation"]

        result = layer.implementation

    mock_class.from_configuration.assert_called_once_with({"key": "value"})
    assert result == mock_impl


# =============================================================================
# SemanticView tests
# =============================================================================


@pytest.fixture
def mock_dimensions() -> list[Dimension]:
    """Create mock dimensions for testing."""
    return [
        Dimension(
            id="orders.order_date",
            name="order_date",
            type=pa.date32(),
            definition="orders.order_date",
            description="Date of the order",
            grain=Grains.DAY,
        ),
        Dimension(
            id="products.category",
            name="category",
            type=pa.utf8(),
            definition="products.category",
            description="Product category",
            grain=None,
        ),
    ]


@pytest.fixture
def mock_metrics() -> list[Metric]:
    """Create mock metrics for testing."""
    return [
        Metric(
            id="orders.revenue",
            name="revenue",
            type=pa.float64(),
            definition="SUM(orders.amount)",
            description="Total revenue",
        ),
        Metric(
            id="orders.count",
            name="order_count",
            type=pa.int64(),
            definition="COUNT(*)",
            description="Number of orders",
        ),
    ]


@pytest.fixture
def mock_implementation(
    mock_dimensions: list[Dimension],
    mock_metrics: list[Metric],
) -> MagicMock:
    """Create a mock implementation."""
    impl = MagicMock()
    impl.get_dimensions.return_value = mock_dimensions
    impl.get_metrics.return_value = mock_metrics
    impl.uid.return_value = "semantic_view_uid_123"
    return impl


@pytest.fixture
def semantic_view(mock_implementation: MagicMock) -> SemanticView:
    """Create a SemanticView with mocked implementation."""
    layer = SemanticLayer()
    layer.name = "Test Layer"
    layer.uuid = uuid.UUID("87654321-4321-8765-4321-876543218765")
    layer.perm = "[Test Layer](id:87654321432187654321876543218765)"

    view = SemanticView()
    view.name = "Orders View"
    view.description = "View of order data"
    view.id = 1
    view.uuid = uuid.UUID("12345678-1234-5678-1234-567812345678")
    view.semantic_layer_uuid = uuid.UUID("87654321-4321-8765-4321-876543218765")
    view.semantic_layer = layer
    view.cache_timeout = 3600
    view.configuration = "{}"
    view.perm = "[Test Layer].[Orders View](id:1)"

    # Persist mocked implementation on this instance
    view.__dict__["implementation"] = mock_implementation

    return view


def test_semantic_view_repr_with_name() -> None:
    """Test SemanticView __repr__ with name."""
    view = SemanticView()
    view.name = "My View"
    view.uuid = uuid.uuid4()
    assert repr(view) == "My View"


def test_semantic_view_repr_without_name() -> None:
    """Test SemanticView __repr__ without name (uses uuid)."""
    view = SemanticView()
    view.name = None
    test_uuid = uuid.uuid4()
    view.uuid = test_uuid
    assert repr(view) == str(test_uuid)


def test_semantic_view_type() -> None:
    """Test SemanticView type property."""
    view = SemanticView()
    assert view.type == "semantic_view"


def test_semantic_view_table_name() -> None:
    """Test SemanticView table_name property."""
    view = SemanticView()
    view.name = "Orders View"
    assert view.table_name == "Orders View"


def test_semantic_view_kind() -> None:
    """Test SemanticView kind property."""
    view = SemanticView()
    assert view.kind == "semantic_view"


def test_semantic_view_offset() -> None:
    """Test SemanticView offset property."""
    view = SemanticView()
    assert view.offset == 0


def test_semantic_view_is_rls_supported() -> None:
    """Test SemanticView is_rls_supported property."""
    view = SemanticView()
    assert view.is_rls_supported is False


def test_semantic_view_query_language() -> None:
    """Test SemanticView query_language property."""
    view = SemanticView()
    assert view.query_language is None


def test_semantic_view_get_query_str() -> None:
    """Test SemanticView get_query_str method."""
    view = SemanticView()
    result = view.get_query_str({})
    assert result == "Not implemented for semantic layers"


def test_semantic_view_get_extra_cache_keys() -> None:
    """Test SemanticView get_extra_cache_keys method."""
    view = SemanticView()
    result = view.get_extra_cache_keys({})
    assert result == []


def test_semantic_view_perm() -> None:
    """Test SemanticView perm stores the view-level permission string."""
    view = SemanticView()
    view.perm = "[My Layer].[My View](id:42)"
    assert view.perm == "[My Layer].[My View](id:42)"


def test_semantic_view_perm_none_by_default() -> None:
    """Test SemanticView perm is None when not set."""
    view = SemanticView()
    assert view.perm is None


def test_semantic_view_get_perm() -> None:
    """Test SemanticView.get_perm() format: [layer].[view](id:N)."""
    layer = SemanticLayer()
    layer.name = "My Layer"
    layer.uuid = uuid.UUID("87654321-4321-8765-4321-876543218765")

    view = SemanticView()
    view.id = 42
    view.name = "My View"
    view.semantic_layer = layer
    assert view.get_perm() == "[My Layer].[My View](id:42)"


def test_semantic_view_get_perm_without_layer() -> None:
    """Test get_perm uses 'unknown' when no semantic_layer."""
    view = SemanticView()
    view.id = 1
    view.name = "Orphan View"
    view.semantic_layer = None  # type: ignore
    assert view.get_perm() == "[unknown].[Orphan View](id:1)"


def test_semantic_view_uid(
    mock_implementation: MagicMock,
    mock_dimensions: list[Dimension],
    mock_metrics: list[Metric],
) -> None:
    """Test SemanticView uid property."""
    view = SemanticView()
    view.name = "Test View"
    view.uuid = uuid.uuid4()
    view.semantic_layer_uuid = uuid.uuid4()

    with patch.object(
        SemanticView,
        "implementation",
        new_callable=lambda: property(lambda s: mock_implementation),
    ):
        assert view.uid == "semantic_view_uid_123"


def test_semantic_view_metrics(
    mock_implementation: MagicMock,
    mock_metrics: list[Metric],
) -> None:
    """Test SemanticView metrics property."""
    view = SemanticView()

    with patch.object(
        SemanticView,
        "implementation",
        new_callable=lambda: property(lambda s: mock_implementation),
    ):
        metrics = view.metrics
        assert len(metrics) == 2
        assert metrics[0].metric_name == "revenue"
        assert metrics[0].expression == "SUM(orders.amount)"
        assert metrics[0].description == "Total revenue"
        assert metrics[1].metric_name == "order_count"


def test_semantic_view_columns(
    mock_implementation: MagicMock,
    mock_dimensions: list[Dimension],
) -> None:
    """Test SemanticView columns property."""
    view = SemanticView()

    with patch.object(
        SemanticView,
        "implementation",
        new_callable=lambda: property(lambda s: mock_implementation),
    ):
        columns = view.columns
        assert len(columns) == 2
        assert columns[0].column_name == "order_date"
        assert columns[0].type == "date32[day]"
        assert columns[0].is_dttm is True
        assert columns[0].description == "Date of the order"
        assert columns[1].column_name == "category"
        assert columns[1].type == "string"
        assert columns[1].is_dttm is False


def test_semantic_view_column_names(
    mock_implementation: MagicMock,
    mock_dimensions: list[Dimension],
) -> None:
    """Test SemanticView column_names property."""
    view = SemanticView()

    with patch.object(
        SemanticView,
        "implementation",
        new_callable=lambda: property(lambda s: mock_implementation),
    ):
        column_names = view.column_names
        assert column_names == ["order_date", "category"]


def test_semantic_view_get_time_grains(
    mock_implementation: MagicMock,
    mock_dimensions: list[Dimension],
) -> None:
    """Test SemanticView get_time_grains property."""
    view = SemanticView()

    with patch.object(
        SemanticView,
        "implementation",
        new_callable=lambda: property(lambda s: mock_implementation),
    ):
        time_grains = view.get_time_grains()
        assert len(time_grains) == 1
        assert time_grains[0]["name"] == "Day"
        assert time_grains[0]["duration"] == "P1D"


def test_semantic_view_has_drill_by_columns_all_exist(
    mock_implementation: MagicMock,
    mock_dimensions: list[Dimension],
) -> None:
    """Test has_drill_by_columns when all columns exist."""
    view = SemanticView()

    with patch.object(
        SemanticView,
        "implementation",
        new_callable=lambda: property(lambda s: mock_implementation),
    ):
        assert view.has_drill_by_columns(["order_date", "category"]) is True


def test_semantic_view_has_drill_by_columns_some_missing(
    mock_implementation: MagicMock,
    mock_dimensions: list[Dimension],
) -> None:
    """Test has_drill_by_columns when some columns are missing."""
    view = SemanticView()

    with patch.object(
        SemanticView,
        "implementation",
        new_callable=lambda: property(lambda s: mock_implementation),
    ):
        assert view.has_drill_by_columns(["order_date", "nonexistent"]) is False


def test_semantic_view_has_drill_by_columns_empty(
    mock_implementation: MagicMock,
    mock_dimensions: list[Dimension],
) -> None:
    """Test has_drill_by_columns with empty list."""
    view = SemanticView()

    with patch.object(
        SemanticView,
        "implementation",
        new_callable=lambda: property(lambda s: mock_implementation),
    ):
        assert view.has_drill_by_columns([]) is True


def test_semantic_view_data(
    mock_implementation: MagicMock,
    mock_dimensions: list[Dimension],
    mock_metrics: list[Metric],
) -> None:
    """Test SemanticView data property."""
    from superset.semantic_layers.models import SemanticLayer

    layer = SemanticLayer()
    layer.name = "My Semantic Layer"
    layer.uuid = uuid.UUID("87654321-4321-8765-4321-876543218765")
    layer.perm = "[My Semantic Layer](id:87654321432187654321876543218765)"

    view = SemanticView()
    view.name = "Orders View"
    view.description = "View of order data"
    view.id = 1
    view.uuid = uuid.UUID("12345678-1234-5678-1234-567812345678")
    view.semantic_layer_uuid = uuid.UUID("87654321-4321-8765-4321-876543218765")
    view.semantic_layer = layer
    view.cache_timeout = 3600

    with patch.object(
        SemanticView,
        "implementation",
        new_callable=lambda: property(lambda s: mock_implementation),
    ):
        data = view.data

        # Check core fields
        assert data["id"] == 1
        assert data["uid"] == "semantic_view_uid_123"
        assert data["type"] == "semantic_view"
        assert data["name"] == "Orders View"
        assert data["description"] == "View of order data"
        assert data["cache_timeout"] == 3600
        assert data["database"] == {}
        assert data["parent"] == {"name": "My Semantic Layer"}

        # Check columns
        assert len(data["columns"]) == 2
        assert data["columns"][0]["column_name"] == "order_date"
        assert data["columns"][0]["type"] == "date32[day]"
        assert data["columns"][0]["is_dttm"] is True
        assert data["columns"][0]["type_generic"] == GenericDataType.TEMPORAL
        assert data["columns"][1]["column_name"] == "category"
        assert data["columns"][1]["type"] == "string"
        assert data["columns"][1]["type_generic"] == GenericDataType.STRING

        # Check metrics
        assert len(data["metrics"]) == 2
        assert data["metrics"][0]["metric_name"] == "revenue"
        assert data["metrics"][0]["expression"] == "SUM(orders.amount)"
        assert data["metrics"][1]["metric_name"] == "order_count"

        # Check column_types and column_names
        assert data["column_types"] == [
            GenericDataType.TEMPORAL,
            GenericDataType.STRING,
        ]
        assert data["column_names"] == ["order_date", "category"]

        # Check other fields
        assert data["table_name"] == "Orders View"
        assert data["datasource_name"] == "Orders View"
        assert data["offset"] == 0


def test_semantic_view_get_query_result(
    mock_implementation: MagicMock,
) -> None:
    """Test SemanticView get_query_result method."""
    view = SemanticView()

    mock_query_object = MagicMock()
    mock_result = MagicMock()

    with patch(
        "superset.semantic_layers.models.get_results",
        return_value=mock_result,
    ) as mock_get_results:
        result = view.get_query_result(mock_query_object)

        mock_get_results.assert_called_once_with(mock_query_object)
        assert result == mock_result


def test_semantic_view_data_for_slices(
    mock_implementation: MagicMock,
    mock_dimensions: list[Dimension],
    mock_metrics: list[Metric],
) -> None:
    """Test SemanticView data_for_slices returns same as data."""
    from superset.semantic_layers.models import SemanticLayer

    layer = SemanticLayer()
    layer.name = "My Semantic Layer"
    layer.uuid = uuid.UUID("87654321-4321-8765-4321-876543218765")
    layer.perm = "[My Semantic Layer](id:87654321432187654321876543218765)"

    view = SemanticView()
    view.name = "Orders View"
    view.description = "View of order data"
    view.id = 1
    view.uuid = uuid.UUID("12345678-1234-5678-1234-567812345678")
    view.semantic_layer_uuid = uuid.UUID("87654321-4321-8765-4321-876543218765")
    view.semantic_layer = layer
    view.cache_timeout = 3600

    with patch.object(
        SemanticView,
        "implementation",
        new_callable=lambda: property(lambda s: mock_implementation),
    ):
        assert view.data_for_slices([]) == view.data


def test_semantic_view_catalog_perm() -> None:
    """Test SemanticView catalog_perm returns None."""
    view = SemanticView()
    assert view.catalog_perm is None


def test_semantic_view_schema_perm() -> None:
    """Test SemanticView schema_perm returns None."""
    view = SemanticView()
    assert view.schema_perm is None


def test_semantic_view_schema() -> None:
    """Test SemanticView schema returns None."""
    view = SemanticView()
    assert view.schema is None


def test_semantic_view_url() -> None:
    """Test SemanticView url property."""
    view = SemanticView()
    view.uuid = uuid.UUID("12345678-1234-5678-1234-567812345678")
    assert view.url == "/semantic_view/12345678-1234-5678-1234-567812345678/"


def test_semantic_view_explore_url() -> None:
    """Test SemanticView explore_url property."""
    view = SemanticView()
    view.id = 42
    assert (
        view.explore_url == "/explore/?datasource_type=semantic_view&datasource_id=42"
    )


def test_semantic_view_implementation() -> None:
    """Test SemanticView implementation property."""
    view = SemanticView()
    view.name = "Test View"
    view.configuration = '{"key": "value"}'

    mock_semantic_layer = MagicMock()
    mock_semantic_view_impl = MagicMock()
    mock_semantic_layer.implementation.get_semantic_view.return_value = (
        mock_semantic_view_impl
    )
    view.semantic_layer = mock_semantic_layer

    # Clear cached property if it exists
    if "implementation" in view.__dict__:
        del view.__dict__["implementation"]

    result = view.implementation

    mock_semantic_layer.implementation.get_semantic_view.assert_called_once_with(
        "Test View",
        {"key": "value"},
    )
    assert result == mock_semantic_view_impl


def test_semantic_view_get_compatible_metrics(
    mock_implementation: MagicMock,
    mock_dimensions: list[Dimension],
    mock_metrics: list[Metric],
) -> None:
    """Test SemanticView get_compatible_metrics maps names to objects and back."""
    view = SemanticView()

    mock_implementation.get_compatible_metrics.return_value = {mock_metrics[0]}

    with patch.object(
        SemanticView,
        "implementation",
        new_callable=lambda: property(lambda s: mock_implementation),
    ):
        result = view.get_compatible_metrics(
            selected_metrics=["revenue", "missing_metric"],
            selected_dimensions=["order_date", "missing_dimension"],
        )

    assert result == ["revenue"]
    args = mock_implementation.get_compatible_metrics.call_args.args
    assert args[0] == {mock_metrics[0]}
    assert args[1] == {mock_dimensions[0]}


def test_semantic_view_get_compatible_dimensions(
    mock_implementation: MagicMock,
    mock_dimensions: list[Dimension],
    mock_metrics: list[Metric],
) -> None:
    """Test SemanticView get_compatible_dimensions maps names to objects and back."""
    view = SemanticView()

    mock_implementation.get_compatible_dimensions.return_value = {mock_dimensions[1]}

    with patch.object(
        SemanticView,
        "implementation",
        new_callable=lambda: property(lambda s: mock_implementation),
    ):
        result = view.get_compatible_dimensions(
            selected_metrics=["order_count", "missing_metric"],
            selected_dimensions=["category", "missing_dimension"],
        )

    assert result == ["category"]
    args = mock_implementation.get_compatible_dimensions.call_args.args
    assert args[0] == {mock_metrics[1]}
    assert args[1] == {mock_dimensions[1]}


# =============================================================================
# SemanticLayer.get_perm tests
# =============================================================================


def test_semantic_layer_get_perm() -> None:
    """Test SemanticLayer.get_perm() format."""
    layer = SemanticLayer()
    layer.name = "My Layer"
    layer.uuid = uuid.UUID("abcdef12-3456-7890-abcd-ef1234567890")
    assert layer.get_perm() == "[My Layer](id:abcdef1234567890abcdef1234567890)"


def test_semantic_layer_get_perm_special_characters() -> None:
    """Test get_perm with special characters in the layer name."""
    layer = SemanticLayer()
    layer.name = "Layer [with] (parens)"
    layer.uuid = uuid.UUID("11111111-1111-1111-1111-111111111111")
    assert (
        layer.get_perm()
        == "[Layer [with] (parens)](id:11111111111111111111111111111111)"
    )


# =============================================================================
# SemanticView.raise_for_access tests
# =============================================================================


def test_semantic_view_raise_for_access_all_datasources(app: Any) -> None:
    """Test raise_for_access passes when user has all_datasource_access."""
    from superset import security_manager

    layer = SemanticLayer()
    layer.name = "Layer"
    layer.uuid = uuid.UUID("abcdef12-3456-7890-abcd-ef1234567890")
    layer.perm = layer.get_perm()

    view = SemanticView()
    view.semantic_layer = layer

    with patch.object(
        security_manager, "can_access_all_datasources", return_value=True
    ):
        view.raise_for_access()


def test_semantic_view_raise_for_access_view_perm(app: Any) -> None:
    """Test raise_for_access passes when user has view-level perm."""
    from superset import security_manager

    layer = SemanticLayer()
    layer.name = "Layer"
    layer.uuid = uuid.UUID("abcdef12-3456-7890-abcd-ef1234567890")
    layer.perm = layer.get_perm()

    view = SemanticView()
    view.id = 1
    view.name = "My View"
    view.semantic_layer = layer
    view.perm = "[Layer].[My View](id:1)"

    with (
        patch.object(
            security_manager, "can_access_all_datasources", return_value=False
        ),
        patch.object(
            security_manager, "can_access", return_value=True
        ) as mock_can_access,
    ):
        view.raise_for_access()
        mock_can_access.assert_called_once_with(
            "datasource_access", "[Layer].[My View](id:1)"
        )


def test_semantic_view_raise_for_access_layer_perm(app: Any) -> None:
    """Test raise_for_access passes via layer perm when view perm is denied."""
    from superset import security_manager

    layer = SemanticLayer()
    layer.name = "Layer"
    layer.uuid = uuid.UUID("abcdef12-3456-7890-abcd-ef1234567890")
    layer.perm = layer.get_perm()

    view = SemanticView()
    view.id = 1
    view.name = "My View"
    view.semantic_layer = layer
    view.perm = "[Layer].[My View](id:1)"

    def side_effect(permission: str, perm: str) -> bool:
        # Deny view perm, allow layer perm
        return perm == layer.perm

    with (
        patch.object(
            security_manager, "can_access_all_datasources", return_value=False
        ),
        patch.object(
            security_manager, "can_access", side_effect=side_effect
        ) as mock_can_access,
    ):
        view.raise_for_access()
        assert mock_can_access.call_count == 2


def test_semantic_view_raise_for_access_denied(app: Any) -> None:
    """Test raise_for_access raises SupersetSecurityException when denied."""
    from superset import security_manager
    from superset.exceptions import SupersetSecurityException

    layer = SemanticLayer()
    layer.name = "Layer"
    layer.uuid = uuid.UUID("abcdef12-3456-7890-abcd-ef1234567890")
    layer.perm = layer.get_perm()

    view = SemanticView()
    view.id = 1
    view.name = "My View"
    view.semantic_layer = layer
    view.perm = "[Layer].[My View](id:1)"

    with (
        patch.object(
            security_manager, "can_access_all_datasources", return_value=False
        ),
        patch.object(security_manager, "can_access", return_value=False),
    ):
        with pytest.raises(SupersetSecurityException):
            view.raise_for_access()


# =============================================================================
# create_missing_perms backfill tests
# =============================================================================


def test_create_missing_perms_backfills_semantic_layer_perm(app: Any) -> None:
    """Test that create_missing_perms sets perm on layers with perm=NULL."""
    from superset import security_manager
    from superset.extensions import db

    layer = SemanticLayer()
    layer.name = "Backfill Layer"
    layer.uuid = uuid.UUID("aaaa1111-2222-3333-4444-555566667777")
    layer.type = "test"
    layer.perm = None  # simulate pre-existing layer without perm

    db.session.add(layer)
    db.session.flush()

    try:
        with (
            patch.object(security_manager, "_get_all_pvms", return_value=[]),
            patch.object(security_manager, "add_permission_view_menu") as mock_add_pvm,
        ):
            security_manager.create_missing_perms()

        expected_perm = "[Backfill Layer](id:aaaa1111222233334444555566667777)"
        assert layer.perm == expected_perm
        mock_add_pvm.assert_any_call("datasource_access", expected_perm)
    finally:
        db.session.rollback()


def test_create_missing_perms_backfills_semantic_view_perm(app: Any) -> None:
    """Test that create_missing_perms sets perm on views with perm=NULL."""
    from superset import security_manager
    from superset.extensions import db

    layer = SemanticLayer()
    layer.name = "Backfill Layer"
    layer.uuid = uuid.UUID("aaaa1111-2222-3333-4444-555566667777")
    layer.type = "test"
    layer.perm = "[Backfill Layer](id:aaaa1111222233334444555566667777)"

    view = SemanticView()
    view.name = "Backfill View"
    view.semantic_layer_uuid = layer.uuid
    view.perm = None  # simulate pre-existing view without perm

    db.session.add(layer)
    db.session.add(view)
    db.session.flush()

    try:
        with (
            patch.object(security_manager, "_get_all_pvms", return_value=[]),
            patch.object(security_manager, "add_permission_view_menu") as mock_add_pvm,
        ):
            security_manager.create_missing_perms()

        expected_perm = f"[Backfill Layer].[Backfill View](id:{view.id})"
        assert view.perm == expected_perm
        mock_add_pvm.assert_any_call("datasource_access", expected_perm)
    finally:
        db.session.rollback()


# =============================================================================
# SemanticView.get_perm with explicit layer_name
# =============================================================================


def test_semantic_view_get_perm_explicit_layer_name() -> None:
    """Test get_perm with explicit layer_name parameter."""
    view = SemanticView()
    view.id = 5
    view.name = "My View"
    view.semantic_layer = None  # type: ignore
    assert (
        view.get_perm(layer_name="Explicit Layer") == "[Explicit Layer].[My View](id:5)"
    )


# =============================================================================
# Event listener tests
# =============================================================================


def test_semantic_view_after_insert_sets_perm(app: Any) -> None:
    """Test that the after_insert event listener sets the perm column."""
    from superset.extensions import db

    layer = SemanticLayer()
    layer.name = "Event Layer"
    layer.uuid = uuid.UUID("eeee1111-2222-3333-4444-555566667777")
    layer.type = "test"

    view = SemanticView()
    view.name = "Event View"
    view.semantic_layer_uuid = layer.uuid

    db.session.add(layer)
    db.session.add(view)
    db.session.flush()

    try:
        assert view.perm == f"[Event Layer].[Event View](id:{view.id})"
    finally:
        db.session.rollback()


def test_semantic_view_before_update_updates_perm(app: Any) -> None:
    """Test that renaming a view updates its perm via the before_update event."""
    from superset.extensions import db

    layer = SemanticLayer()
    layer.name = "Update Layer"
    layer.uuid = uuid.UUID("dddd1111-2222-3333-4444-555566667777")
    layer.type = "test"

    view = SemanticView()
    view.name = "Old Name"
    view.semantic_layer_uuid = layer.uuid

    db.session.add(layer)
    db.session.add(view)
    db.session.flush()

    try:
        view.name = "New Name"
        db.session.flush()
        assert view.perm == f"[Update Layer].[New Name](id:{view.id})"
    finally:
        db.session.rollback()


def test_semantic_layer_after_delete_calls_security_manager() -> None:
    """Test SemanticLayer.after_delete delegates to security manager."""
    from superset import security_manager

    mapper = MagicMock()
    connection = MagicMock()
    target = MagicMock(spec=SemanticLayer)

    with patch.object(security_manager, "semantic_layer_after_delete") as mock_hook:
        SemanticLayer.after_delete(mapper, connection, target)

    mock_hook.assert_called_once_with(mapper, connection, target)


def test_semantic_view_after_delete_calls_security_manager() -> None:
    """Test SemanticView.after_delete delegates to security manager."""
    from superset import security_manager

    mapper = MagicMock()
    connection = MagicMock()
    target = MagicMock(spec=SemanticView)

    with patch.object(security_manager, "semantic_view_after_delete") as mock_hook:
        SemanticView.after_delete(mapper, connection, target)

    mock_hook.assert_called_once_with(mapper, connection, target)


def test_semantic_layer_rename_cascades_to_view_perms(app: Any) -> None:
    """Test that renaming a layer cascades the perm update to its views."""
    from superset.extensions import db

    layer = SemanticLayer()
    layer.name = "Old Layer"
    layer.uuid = uuid.UUID("cccc1111-2222-3333-4444-555566667777")
    layer.type = "test"

    view = SemanticView()
    view.name = "Cascade View"
    view.semantic_layer_uuid = layer.uuid

    db.session.add(layer)
    db.session.add(view)
    db.session.flush()

    assert view.perm == f"[Old Layer].[Cascade View](id:{view.id})"

    try:
        layer.name = "New Layer"
        db.session.flush()

        # Cascade update is via raw SQL, so refresh the ORM object
        db.session.refresh(view)
        assert view.perm == f"[New Layer].[Cascade View](id:{view.id})"
    finally:
        db.session.rollback()


# =============================================================================
# build_semantic_view_query dual perm tests
# =============================================================================


def test_build_semantic_view_query_view_perm_grants_access(app: Any) -> None:
    """Test that view-level perm grants access in build_semantic_view_query."""
    from superset import security_manager
    from superset.daos.datasource import DatasourceDAO
    from superset.extensions import db

    layer = SemanticLayer()
    layer.name = "Query Layer"
    layer.uuid = uuid.UUID("bbbb1111-2222-3333-4444-555566667777")
    layer.type = "test"

    view = SemanticView()
    view.name = "Query View"
    view.semantic_layer_uuid = layer.uuid

    db.session.add(layer)
    db.session.add(view)
    db.session.flush()

    try:
        # Only grant the view-level perm (not the layer perm)
        with (
            patch.object(
                security_manager, "can_access_all_datasources", return_value=False
            ),
            patch.object(
                security_manager,
                "user_view_menu_names",
                return_value={view.perm},
            ),
        ):
            query = DatasourceDAO.build_semantic_view_query(name_filter=None)
            results = db.session.execute(query).fetchall()

        item_ids = [row.item_id for row in results]
        assert view.id in item_ids
    finally:
        db.session.rollback()


def test_build_semantic_view_query_layer_perm_grants_access(app: Any) -> None:
    """Test that layer-level perm grants access in build_semantic_view_query."""
    from superset import security_manager
    from superset.daos.datasource import DatasourceDAO
    from superset.extensions import db

    layer = SemanticLayer()
    layer.name = "Query Layer 2"
    layer.uuid = uuid.UUID("aaaa2222-3333-4444-5555-666677778888")
    layer.type = "test"

    view = SemanticView()
    view.name = "Query View 2"
    view.semantic_layer_uuid = layer.uuid

    db.session.add(layer)
    db.session.add(view)
    db.session.flush()

    try:
        # Only grant the layer-level perm (not the view perm)
        with (
            patch.object(
                security_manager, "can_access_all_datasources", return_value=False
            ),
            patch.object(
                security_manager,
                "user_view_menu_names",
                return_value={layer.perm},
            ),
        ):
            query = DatasourceDAO.build_semantic_view_query(name_filter=None)
            results = db.session.execute(query).fetchall()

        item_ids = [row.item_id for row in results]
        assert view.id in item_ids
    finally:
        db.session.rollback()


def test_build_semantic_view_query_no_perm_excludes(app: Any) -> None:
    """Test that views are excluded when user has neither view nor layer perm."""
    from superset import security_manager
    from superset.daos.datasource import DatasourceDAO
    from superset.extensions import db

    layer = SemanticLayer()
    layer.name = "Query Layer 3"
    layer.uuid = uuid.UUID("aaaa3333-4444-5555-6666-777788889999")
    layer.type = "test"

    view = SemanticView()
    view.name = "Query View 3"
    view.semantic_layer_uuid = layer.uuid

    db.session.add(layer)
    db.session.add(view)
    db.session.flush()

    try:
        with (
            patch.object(
                security_manager, "can_access_all_datasources", return_value=False
            ),
            patch.object(
                security_manager,
                "user_view_menu_names",
                return_value={"[unrelated](id:xxx)"},
            ),
        ):
            query = DatasourceDAO.build_semantic_view_query(name_filter=None)
            results = db.session.execute(query).fetchall()

        item_ids = [row.item_id for row in results]
        assert view.id not in item_ids
    finally:
        db.session.rollback()