pbi-cli/tests/test_tmdl_diff.py

"""Tests for pbi_cli.core.tmdl_diff."""

from __future__ import annotations

from pathlib import Path
from typing import Any

import pytest

from pbi_cli.core.errors import PbiCliError
from pbi_cli.core.tmdl_diff import diff_tmdl_folders

# ---------------------------------------------------------------------------
# Fixture helpers
# ---------------------------------------------------------------------------

_MODEL_TMDL = """\
model Model
\tculture: en-US
\tdefaultPowerBIDataSourceVersion: powerBI_V3
\tsourceQueryCulture: en-US

ref table Sales
ref cultureInfo en-US
"""

_RELATIONSHIPS_TMDL = """\
relationship abc-def-111
\tlineageTag: xyz
\tfromColumn: Sales.ProductID
\ttoColumn: Product.ProductID

relationship abc-def-222
\tfromColumn: Sales.CustomerID
\ttoColumn: Customer.CustomerID
"""

_SALES_TMDL = """\
table Sales
\tlineageTag: tbl-001

\tmeasure 'Total Revenue' = SUM(Sales[Amount])
\t\tformatString: "$#,0"
\t\tlineageTag: msr-001

\tcolumn Amount
\t\tdataType: decimal
\t\tlineageTag: col-001
\t\tsummarizeBy: sum
\t\tsourceColumn: Amount

\tpartition Sales = m
\t\tmode: import
\t\tsource
\t\t\tlet
\t\t\t    Source = Csv.Document(...)
\t\t\tin
\t\t\t    Source
"""

_DATE_TMDL = """\
table Date
\tlineageTag: tbl-002

\tcolumn Date
\t\tdataType: dateTime
\t\tlineageTag: col-002
\t\tsummarizeBy: none
\t\tsourceColumn: Date
"""

# Inline TMDL snippets reused across multiple tests
_NEW_MEASURE_SNIPPET = (
    "\n\tmeasure 'YTD Revenue'"
    " = CALCULATE([Total Revenue], DATESYTD('Date'[Date]))"
    "\n\t\tlineageTag: msr-new\n"
)
_TOTAL_REVENUE_BLOCK = (
    "\n\tmeasure 'Total Revenue' = SUM(Sales[Amount])"
    '\n\t\tformatString: "$#,0"'
    "\n\t\tlineageTag: msr-001\n"
)
_NEW_COL_SNIPPET = (
    "\n\tcolumn Region"
    "\n\t\tdataType: string"
    "\n\t\tsummarizeBy: none"
    "\n\t\tsourceColumn: Region\n"
)
_AMOUNT_COL_BLOCK = (
    "\n\tcolumn Amount"
    "\n\t\tdataType: decimal"
    "\n\t\tlineageTag: col-001"
    "\n\t\tsummarizeBy: sum"
    "\n\t\tsourceColumn: Amount\n"
)
_NEW_REL_SNIPPET = (
    "\nrelationship abc-def-999"
    "\n\tfromColumn: Sales.RegionID"
    "\n\ttoColumn: Region.ID\n"
)
_TRIMMED_RELS = (
    "relationship abc-def-111"
    "\n\tfromColumn: Sales.ProductID"
    "\n\ttoColumn: Product.ProductID\n"
)
_REL_222_BASE = (
    "relationship abc-def-222"
    "\n\tfromColumn: Sales.CustomerID"
    "\n\ttoColumn: Customer.CustomerID"
)
_REL_222_CHANGED = (
    "relationship abc-def-222"
    "\n\tfromColumn: Sales.CustomerID"
    "\n\ttoColumn: Customer.CustomerID"
    "\n\tcrossFilteringBehavior: bothDirections"
)


def _make_tmdl_folder(
    root: Path,
    *,
    model_text: str = _MODEL_TMDL,
    relationships_text: str = _RELATIONSHIPS_TMDL,
    tables: dict[str, str] | None = None,
) -> Path:
    """Create a minimal TMDL folder under root and return its path."""
    if tables is None:
        tables = {"Sales": _SALES_TMDL, "Date": _DATE_TMDL}
    root.mkdir(parents=True, exist_ok=True)
    (root / "model.tmdl").write_text(model_text, encoding="utf-8")
    (root / "database.tmdl").write_text("database\n\tcompatibilityLevel: 1600\n", encoding="utf-8")
    (root / "relationships.tmdl").write_text(relationships_text, encoding="utf-8")
    tables_dir = root / "tables"
    tables_dir.mkdir()
    for name, text in tables.items():
        (tables_dir / f"{name}.tmdl").write_text(text, encoding="utf-8")
    return root


def _make_semantic_model_folder(
    root: Path,
    **kwargs: Any,
) -> Path:
    """Create a SemanticModel-layout folder (definition/ subdirectory)."""
    root.mkdir(parents=True, exist_ok=True)
    defn_dir = root / "definition"
    defn_dir.mkdir()
    _make_tmdl_folder(defn_dir, **kwargs)
    (root / ".platform").write_text("{}", encoding="utf-8")
    return root


# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------


class TestDiffTmdlFolders:
    def test_identical_folders_returns_no_changes(self, tmp_path: Path) -> None:
        base = _make_tmdl_folder(tmp_path / "base")
        head = _make_tmdl_folder(tmp_path / "head")
        result = diff_tmdl_folders(str(base), str(head))
        assert result["changed"] is False
        assert result["summary"]["tables_added"] == 0
        assert result["summary"]["tables_removed"] == 0
        assert result["summary"]["tables_changed"] == 0

    def test_lineage_tag_only_change_is_not_reported(self, tmp_path: Path) -> None:
        base = _make_tmdl_folder(tmp_path / "base")
        changed_sales = _SALES_TMDL.replace("tbl-001", "NEW-TAG").replace("msr-001", "NEW-MSR")
        head = _make_tmdl_folder(
            tmp_path / "head",
            tables={"Sales": changed_sales, "Date": _DATE_TMDL},
        )
        result = diff_tmdl_folders(str(base), str(head))
        assert result["changed"] is False

    def test_table_added(self, tmp_path: Path) -> None:
        product_tmdl = "table Product\n\tlineageTag: tbl-003\n\n\tcolumn ID\n\t\tdataType: int64\n"
        base = _make_tmdl_folder(tmp_path / "base")
        head = _make_tmdl_folder(
            tmp_path / "head",
            tables={"Sales": _SALES_TMDL, "Date": _DATE_TMDL, "Product": product_tmdl},
        )
        result = diff_tmdl_folders(str(base), str(head))
        assert result["changed"] is True
        assert "Product" in result["tables"]["added"]
        assert result["tables"]["removed"] == []

    def test_table_removed(self, tmp_path: Path) -> None:
        base = _make_tmdl_folder(tmp_path / "base")
        head = _make_tmdl_folder(tmp_path / "head", tables={"Sales": _SALES_TMDL})
        result = diff_tmdl_folders(str(base), str(head))
        assert "Date" in result["tables"]["removed"]

    def test_measure_added(self, tmp_path: Path) -> None:
        modified_sales = _SALES_TMDL + _NEW_MEASURE_SNIPPET
        base = _make_tmdl_folder(tmp_path / "base")
        head = _make_tmdl_folder(
            tmp_path / "head",
            tables={"Sales": modified_sales, "Date": _DATE_TMDL},
        )
        result = diff_tmdl_folders(str(base), str(head))
        assert result["changed"] is True
        sales_diff = result["tables"]["changed"]["Sales"]
        assert "YTD Revenue" in sales_diff["measures_added"]

    def test_measure_removed(self, tmp_path: Path) -> None:
        stripped_sales = _SALES_TMDL.replace(_TOTAL_REVENUE_BLOCK, "")
        base = _make_tmdl_folder(tmp_path / "base")
        head = _make_tmdl_folder(
            tmp_path / "head",
            tables={"Sales": stripped_sales, "Date": _DATE_TMDL},
        )
        result = diff_tmdl_folders(str(base), str(head))
        sales_diff = result["tables"]["changed"]["Sales"]
        assert "Total Revenue" in sales_diff["measures_removed"]

    def test_measure_expression_changed(self, tmp_path: Path) -> None:
        modified_sales = _SALES_TMDL.replace(
            "measure 'Total Revenue' = SUM(Sales[Amount])",
            "measure 'Total Revenue' = SUMX(Sales, Sales[Amount] * Sales[Qty])",
        )
        base = _make_tmdl_folder(tmp_path / "base")
        head = _make_tmdl_folder(
            tmp_path / "head",
            tables={"Sales": modified_sales, "Date": _DATE_TMDL},
        )
        result = diff_tmdl_folders(str(base), str(head))
        sales_diff = result["tables"]["changed"]["Sales"]
        assert "Total Revenue" in sales_diff["measures_changed"]

    def test_column_added(self, tmp_path: Path) -> None:
        modified_sales = _SALES_TMDL + _NEW_COL_SNIPPET
        base = _make_tmdl_folder(tmp_path / "base")
        head = _make_tmdl_folder(
            tmp_path / "head",
            tables={"Sales": modified_sales, "Date": _DATE_TMDL},
        )
        result = diff_tmdl_folders(str(base), str(head))
        sales_diff = result["tables"]["changed"]["Sales"]
        assert "Region" in sales_diff["columns_added"]

    def test_column_removed(self, tmp_path: Path) -> None:
        stripped = _SALES_TMDL.replace(_AMOUNT_COL_BLOCK, "")
        base = _make_tmdl_folder(tmp_path / "base")
        head = _make_tmdl_folder(
            tmp_path / "head",
            tables={"Sales": stripped, "Date": _DATE_TMDL},
        )
        result = diff_tmdl_folders(str(base), str(head))
        sales_diff = result["tables"]["changed"]["Sales"]
        assert "Amount" in sales_diff["columns_removed"]

    def test_relationship_added(self, tmp_path: Path) -> None:
        base = _make_tmdl_folder(tmp_path / "base")
        head = _make_tmdl_folder(
            tmp_path / "head",
            relationships_text=_RELATIONSHIPS_TMDL + _NEW_REL_SNIPPET,
        )
        result = diff_tmdl_folders(str(base), str(head))
        assert "Sales.RegionID -> Region.ID" in result["relationships"]["added"]

    def test_relationship_removed(self, tmp_path: Path) -> None:
        base = _make_tmdl_folder(tmp_path / "base")
        head = _make_tmdl_folder(tmp_path / "head", relationships_text=_TRIMMED_RELS)
        result = diff_tmdl_folders(str(base), str(head))
        assert "Sales.CustomerID -> Customer.CustomerID" in result["relationships"]["removed"]

    def test_relationship_changed(self, tmp_path: Path) -> None:
        changed_rels = _RELATIONSHIPS_TMDL.replace(_REL_222_BASE, _REL_222_CHANGED)
        base = _make_tmdl_folder(tmp_path / "base")
        head = _make_tmdl_folder(tmp_path / "head", relationships_text=changed_rels)
        result = diff_tmdl_folders(str(base), str(head))
        assert "Sales.CustomerID -> Customer.CustomerID" in result["relationships"]["changed"]

    def test_model_property_changed(self, tmp_path: Path) -> None:
        changed_model = _MODEL_TMDL.replace("culture: en-US", "culture: fr-FR")
        base = _make_tmdl_folder(tmp_path / "base")
        head = _make_tmdl_folder(tmp_path / "head", model_text=changed_model)
        result = diff_tmdl_folders(str(base), str(head))
        assert result["summary"]["model_changed"] is True
        assert any("culture" in p for p in result["model"]["changed_properties"])

    def test_semantic_model_layout(self, tmp_path: Path) -> None:
        """Handles the SemanticModel folder layout (definition/ subdirectory)."""
        base = _make_semantic_model_folder(tmp_path / "MyModel.SemanticModel.base")
        head = _make_semantic_model_folder(tmp_path / "MyModel.SemanticModel.head")
        result = diff_tmdl_folders(str(base), str(head))
        assert result["changed"] is False

    def test_missing_base_folder_raises(self, tmp_path: Path) -> None:
        head = _make_tmdl_folder(tmp_path / "head")
        with pytest.raises(PbiCliError, match="Base folder not found"):
            diff_tmdl_folders(str(tmp_path / "nonexistent"), str(head))

    def test_missing_head_folder_raises(self, tmp_path: Path) -> None:
        base = _make_tmdl_folder(tmp_path / "base")
        with pytest.raises(PbiCliError, match="Head folder not found"):
            diff_tmdl_folders(str(base), str(tmp_path / "nonexistent"))

    def test_result_keys_present(self, tmp_path: Path) -> None:
        base = _make_tmdl_folder(tmp_path / "base")
        head = _make_tmdl_folder(tmp_path / "head")
        result = diff_tmdl_folders(str(base), str(head))
        assert "base" in result
        assert "head" in result
        assert "changed" in result
        assert "summary" in result
        assert "tables" in result
        assert "relationships" in result
        assert "model" in result

    def test_no_relationships_file(self, tmp_path: Path) -> None:
        """Handles missing relationships.tmdl gracefully."""
        base = _make_tmdl_folder(tmp_path / "base", relationships_text="")
        head = _make_tmdl_folder(tmp_path / "head", relationships_text="")
        result = diff_tmdl_folders(str(base), str(head))
        assert result["relationships"] == {"added": [], "removed": [], "changed": []}