tests/test_data_quality_gates.py

from tests.e2e_quality import (
    SYNTHETIC_FAILURE_SCORE,
    SYNTHETIC_WARNING_SCORE_CAP,
    detect_synthetic_dimension_values,
    grade_data_quality,
)


def test_detect_synthetic_route_number_values():
    sample_data = """
Table: SHIPMENTS
Columns: ROUTE_NAME, SHIPMENTS
  {'ROUTE_NAME': 'North Corridor Route 31', 'SHIPMENTS': 120}
  {'ROUTE_NAME': 'Central Express Route 21', 'SHIPMENTS': 98}
  {'ROUTE_NAME': 'Central Express Route 31', 'SHIPMENTS': 87}
"""

    result = detect_synthetic_dimension_values(sample_data)

    assert result["fail"] is False
    assert result["warn"] is True
    assert result["count"] == 3
    assert result["occurrences"] == 3
    assert "North Corridor Route 31" in result["examples"]


def test_grade_data_quality_caps_low_volume_synthetic_numeric_suffixes(monkeypatch):
    monkeypatch.setattr(
        "tests.e2e_quality._call_grader",
        lambda prompt: {
            "score": 96,
            "reasoning": "Otherwise solid data.",
            "strengths": ["Good scale"],
            "weaknesses": [],
        },
    )
    sample_data = """
Table: ROUTES
Columns: ROUTE_NAME
  {'ROUTE_NAME': 'North Corridor Route 31'}
"""

    result = grade_data_quality(
        company="Maersk",
        vertical="Transportation & Logistics",
        line="Shipping",
        function="Finance",
        model_tml="model: {}",
        sample_data=sample_data,
    )

    assert result["score"] == SYNTHETIC_WARNING_SCORE_CAP
    assert result["synthetic_dimension_warning"]["warn"] is True
    assert "Synthetic numeric-suffix dimension values detected at low volume" in result["weaknesses"]
    assert "Capped at" in result["reasoning"]


def test_grade_data_quality_fails_repeated_synthetic_numeric_suffixes():
    sample_data = "\n".join(
        [
            "Table: ROUTES",
            "Columns: ROUTE_NAME",
            *[
                "  {'ROUTE_NAME': 'North Corridor Route 31'}"
                for _ in range(10)
            ],
        ]
    )

    result = grade_data_quality(
        company="Maersk",
        vertical="Transportation & Logistics",
        line="Shipping",
        function="Finance",
        model_tml="model: {}",
        sample_data=sample_data,
    )

    assert result["score"] == SYNTHETIC_FAILURE_SCORE
    assert result["synthetic_dimension_failure"]["fail"] is True
    assert result["synthetic_dimension_failure"]["occurrences"] == 10
    assert "North Corridor Route 31" in result["weaknesses"]
    assert "Automatic data-quality failure" in result["reasoning"]


def test_detect_synthetic_route_gate_ignores_natural_named_values_without_suffixes():
    sample_data = """
Table: SHIPMENTS
Columns: ROUTE_NAME, CUSTOMER_NAME
  {'ROUTE_NAME': 'Pacific Northwest Corridor', 'CUSTOMER_NAME': 'Acme Logistics'}
  {'ROUTE_NAME': 'Atlanta to Savannah Intermodal', 'CUSTOMER_NAME': 'Northstar Foods'}
"""

    result = detect_synthetic_dimension_values(sample_data)

    assert result == {
        "fail": False,
        "warn": False,
        "examples": [],
        "count": 0,
        "occurrences": 0,
    }