ktx/python/ktx-sl/tests/test_table_identifier_parser.py

from semantic_layer.table_identifier_parser import (
    ParseTableIdentifierItem,
    parse_table_identifier_batch,
    parse_table_identifier_one,
)


def test_parse_table_identifier_supported_dialects_and_aliases() -> None:
    response = parse_table_identifier_batch(
        [
            ParseTableIdentifierItem(
                key="pg",
                sql_table_name="public.orders AS o",
                dialect="postgres",
            ),
            ParseTableIdentifierItem(
                key="bq",
                sql_table_name="analytics.orders",
                dialect="bigquery",
            ),
            ParseTableIdentifierItem(
                key="sf",
                sql_table_name="RAW.PUBLIC.ORDERS",
                dialect="snowflake",
            ),
        ]
    )

    assert response["pg"].ok is True
    assert response["pg"].schema_ == "public"
    assert response["pg"].name == "orders"
    assert response["pg"].canonical_table == "public.orders"
    assert response["bq"].ok is True
    assert response["bq"].schema_ == "analytics"
    assert response["bq"].name == "orders"
    assert response["sf"].ok is True
    assert response["sf"].catalog == "RAW"
    assert response["sf"].schema_ == "PUBLIC"
    assert response["sf"].name == "ORDERS"


def test_parse_table_identifier_rejects_non_physical_inputs() -> None:
    assert (
        parse_table_identifier_one("${orders.SQL_TABLE_NAME}", "postgres").reason
        == "looker_template_unresolved"
    )
    assert (
        parse_table_identifier_one("(select * from public.orders)", "postgres").reason
        == "derived_table_not_supported"
    )
    assert (
        parse_table_identifier_one(
            "public.orders join public.users on true", "postgres"
        ).reason
        == "multiple_table_references"
    )
    assert (
        parse_table_identifier_one("public.orders", "not-a-dialect").reason
        == "unsupported_dialect"
    )


def test_parse_table_identifier_preserves_batch_keys() -> None:
    response = parse_table_identifier_batch(
        [
            ParseTableIdentifierItem(
                key="z", sql_table_name="public.z", dialect="postgres"
            ),
            ParseTableIdentifierItem(
                key="a", sql_table_name="public.a", dialect="postgres"
            ),
        ]
    )

    assert list(response) == ["z", "a"]
    assert response["z"].name == "z"
    assert response["a"].name == "a"