stack/tests/aco/test_express_data_quality.py

"""Tests for aco.express.data_quality — testing_summary function.

Verifies the join logic that combines dbt_tests with
elementary_test_results filtered to the first-ranked test
invocation (rank=1 by generated_at).
"""

from __future__ import annotations

from datetime import datetime

import polars as pl
import pytest

from aco.express.data_quality import testing_summary as _testing_summary

# ── fixtures ────────────────────────────────────────────────────────


@pytest.fixture
def dbt_tests_df() -> pl.DataFrame:
    """Minimal dbt_tests with unique_id as join key."""
    return pl.DataFrame(
        {
            "unique_id": ["test_a", "test_b", "test_c"],
            "name": ["not_null_id", "unique_id", "accepted_values"],
        }
    )


@pytest.fixture
def elementary_test_results_df() -> pl.DataFrame:
    """Elementary test results spanning two invocations."""
    return pl.DataFrame(
        {
            "test_unique_id": [
                "test_a",
                "test_b",
                "test_a",
                "test_b",
            ],
            "invocation_id": [
                "inv_old",
                "inv_old",
                "inv_new",
                "inv_new",
            ],
            "status": [
                "pass",
                "fail",
                "pass",
                "pass",
            ],
        }
    )


@pytest.fixture
def dbt_invocations_df() -> pl.DataFrame:
    """Two test invocations plus one non-test invocation."""
    return pl.DataFrame(
        {
            "invocation_id": [
                "inv_old",
                "inv_new",
                "inv_run",
            ],
            "command": ["test", "test", "run"],
            "generated_at": [
                datetime(2024, 1, 1, 10, 0),
                datetime(2024, 1, 2, 10, 0),
                datetime(2024, 1, 3, 10, 0),
            ],
        }
    )


# ── TestTestingSummary ──────────────────────────────────────────────


class TestTestingSummary:
    """Tests for the testing_summary join function."""

    def test_returns_dataframe(
        self,
        dbt_tests_df,
        elementary_test_results_df,
        dbt_invocations_df,
    ) -> None:
        result = _testing_summary(
            dbt_tests_df,
            elementary_test_results_df,
            dbt_invocations_df,
        )
        assert isinstance(result, pl.DataFrame)

    def test_left_join_preserves_all_dbt_tests(
        self,
        dbt_tests_df,
        elementary_test_results_df,
        dbt_invocations_df,
    ) -> None:
        """All dbt_tests rows kept even without matching results."""
        result = _testing_summary(
            dbt_tests_df,
            elementary_test_results_df,
            dbt_invocations_df,
        )
        result_ids = result["unique_id"].to_list()
        assert "test_a" in result_ids
        assert "test_b" in result_ids
        # test_c has no result rows but is still present
        assert "test_c" in result_ids

    def test_unmatched_test_has_null_status(
        self,
        dbt_tests_df,
        elementary_test_results_df,
        dbt_invocations_df,
    ) -> None:
        """test_c has no elementary result, so status is null."""
        result = _testing_summary(
            dbt_tests_df,
            elementary_test_results_df,
            dbt_invocations_df,
        )
        row_c = result.filter(pl.col("unique_id") == "test_c")
        assert row_c["status"][0] is None

    def test_only_first_ranked_invocation_included(
        self,
        dbt_tests_df,
        elementary_test_results_df,
        dbt_invocations_df,
    ) -> None:
        """Only the rank-1 (earliest) invocation joins."""
        result = _testing_summary(
            dbt_tests_df,
            elementary_test_results_df,
            dbt_invocations_df,
        )
        # rank(method="ordinal") ascending: rank 1 = inv_old
        inv_ids = result.filter(~pl.col("invocation_id").is_null())[
            "invocation_id"
        ].to_list()
        assert all(i == "inv_old" for i in inv_ids)

    def test_non_test_invocations_excluded(
        self,
        dbt_tests_df,
        elementary_test_results_df,
        dbt_invocations_df,
    ) -> None:
        """The 'run' command invocation is ignored."""
        result = _testing_summary(
            dbt_tests_df,
            elementary_test_results_df,
            dbt_invocations_df,
        )
        inv_ids = result.filter(~pl.col("invocation_id").is_null())[
            "invocation_id"
        ].to_list()
        assert "inv_run" not in inv_ids

    def test_row_count_matches_dbt_tests(
        self,
        dbt_tests_df,
        elementary_test_results_df,
        dbt_invocations_df,
    ) -> None:
        """Left join should produce exactly len(dbt_tests) rows
        when each test has at most one result per invocation."""
        result = _testing_summary(
            dbt_tests_df,
            elementary_test_results_df,
            dbt_invocations_df,
        )
        assert len(result) == len(dbt_tests_df)

    def test_empty_invocations(self, dbt_tests_df) -> None:
        """No invocations at all: all result columns are null."""
        empty_etr = pl.DataFrame(
            {
                "test_unique_id": pl.Series([], dtype=pl.Utf8),
                "invocation_id": pl.Series([], dtype=pl.Utf8),
                "status": pl.Series([], dtype=pl.Utf8),
            }
        )
        empty_inv = pl.DataFrame(
            {
                "invocation_id": pl.Series([], dtype=pl.Utf8),
                "command": pl.Series([], dtype=pl.Utf8),
                "generated_at": pl.Series([], dtype=pl.Datetime),
            }
        )
        result = _testing_summary(dbt_tests_df, empty_etr, empty_inv)
        assert len(result) == len(dbt_tests_df)
        assert result["status"].is_null().all()

    def test_empty_dbt_tests(
        self,
        elementary_test_results_df,
        dbt_invocations_df,
    ) -> None:
        """No dbt_tests at all: result is empty."""
        empty_tests = pl.DataFrame(
            {
                "unique_id": pl.Series([], dtype=pl.Utf8),
                "name": pl.Series([], dtype=pl.Utf8),
            }
        )
        result = _testing_summary(
            empty_tests,
            elementary_test_results_df,
            dbt_invocations_df,
        )
        assert len(result) == 0

    def test_single_invocation(self, dbt_tests_df) -> None:
        """With exactly one test invocation, it is the latest."""
        etr = pl.DataFrame(
            {
                "test_unique_id": ["test_a"],
                "invocation_id": ["inv_only"],
                "status": ["pass"],
            }
        )
        inv = pl.DataFrame(
            {
                "invocation_id": ["inv_only"],
                "command": ["test"],
                "generated_at": [datetime(2024, 6, 1, 12, 0)],
            }
        )
        result = _testing_summary(dbt_tests_df, etr, inv)
        assert len(result) == len(dbt_tests_df)
        matched = result.filter(pl.col("unique_id") == "test_a")
        assert matched["status"][0] == "pass"