tests/test_dataset_quality.py

from demoprep_app.dataset.contracts import DatasetBundle, DatasetColumn, DatasetTable
from demoprep_app.dataset.generators.template_generator import TemplateDatasetGenerator
from demoprep_app.dataset.quality import validate_dataset_quality
from demoprep_app.scenario.contract import DimensionSpec, ScenarioContract


def test_dataset_quality_blocks_numbered_placeholder_dimension_values():
    dataset = DatasetBundle(
        scenario=_scenario(),
        tables=[
            DatasetTable(
                "SALES_REPS",
                "one row per sales rep",
                [
                    DatasetColumn("SALES_REP_KEY", "dimension_key", "NUMBER", nullable=False),
                    DatasetColumn("SALES_REP_NAME", "sales_rep", "VARCHAR(255)", nullable=False),
                ],
                [{"SALES_REP_KEY": 1, "SALES_REP_NAME": "Alex Chen Sales Rep 07"}],
            )
        ],
    )

    report = validate_dataset_quality(dataset)

    assert not report.ok
    assert "Numbered placeholder" in report.summary()


def test_dataset_quality_blocks_generic_numeric_suffix_dimension_values():
    dataset = DatasetBundle(
        scenario=_scenario(),
        tables=[
            DatasetTable(
                "STORES",
                "one row per store",
                [
                    DatasetColumn("STORE_KEY", "dimension_key", "NUMBER", nullable=False),
                    DatasetColumn("STORE_NAME", "store", "VARCHAR(255)", nullable=False),
                ],
                [{"STORE_KEY": 1, "STORE_NAME": "Nike Northeast Metro 01"}],
            )
        ],
    )

    report = validate_dataset_quality(dataset)

    assert not report.ok
    assert "Generic numeric-suffix placeholder" in report.summary()


def test_template_generator_records_fallback_dimension_events():
    scenario = _scenario(
        dimensions=[
            DimensionSpec(name="ACCOUNTS", semantic_role="account", values=["Strategic Shipper"]),
        ]
    )

    dataset = TemplateDatasetGenerator().generate(scenario, row_count=25)

    fallback_events = dataset.scenario.metadata.get("data_fallback_events", [])
    assert fallback_events
    assert fallback_events[0]["dimension"] == "ACCOUNTS"
    assert fallback_events[0]["semantic_family"] == "organization"
    assert fallback_events[0]["source"] == "table_padding"
    assert "Strategic Shipper" in {row["ACCOUNT_NAME"] for row in dataset.table_map()["ACCOUNTS"].rows}


def test_template_generator_records_internal_semantic_expansion_events():
    scenario = ScenarioContract(
        company_name="Generic Consulting",
        company_url="https://example.com",
        use_case="Operations",
        scenario_type="professional_services_engagements",
        fact_grain="client-service-line-sector-month",
        demo_audience="Practice leaders",
        business_problem="Understand engagement performance.",
        metric_formulas={},
        dashboard_questions=[],
        seasonality=[],
        metadata={"seed": 123},
    )
    dataset = TemplateDatasetGenerator().generate(scenario, row_count=25)

    fallback_events = dataset.scenario.metadata.get("data_fallback_events", [])

    assert any(event["dimension"] in {"SERVICE_LINES", "SECTORS", "CONSULTANTS"} for event in fallback_events)
    assert all(event.get("semantic_family") for event in fallback_events)
    assert all(event.get("source") in {"semantic_family_pool", "generated_semantic_family_pool"} for event in fallback_events)


def test_dataset_quality_blocks_semantic_contamination():
    dataset = DatasetBundle(
        scenario=_scenario(),
        tables=[
            DatasetTable(
                "CONSULTANTS",
                "one row per consultant",
                [
                    DatasetColumn("CONSULTANT_KEY", "dimension_key", "NUMBER", nullable=False),
                    DatasetColumn("CONSULTANT_NAME", "consultant", "VARCHAR(255)", nullable=False),
                ],
                [{"CONSULTANT_KEY": 1, "CONSULTANT_NAME": "APAC"}],
            )
        ],
    )

    report = validate_dataset_quality(dataset)

    assert not report.ok
    assert "person dimension" in report.summary()


def test_dataset_quality_blocks_duplicate_unrelated_dimension_sets():
    shared = ["Enterprise", "Commercial", "Digital", "Public Sector", "Strategic"]
    dataset = DatasetBundle(
        scenario=_scenario(),
        tables=[
            DatasetTable(
                "CLIENTS",
                "one row per client",
                [
                    DatasetColumn("CLIENT_KEY", "dimension_key", "NUMBER", nullable=False),
                    DatasetColumn("CLIENT_NAME", "client", "VARCHAR(255)", nullable=False),
                ],
                [{"CLIENT_KEY": idx, "CLIENT_NAME": value} for idx, value in enumerate(shared, start=1)],
            ),
            DatasetTable(
                "SERVICE_LINES",
                "one row per service line",
                [
                    DatasetColumn("SERVICE_LINE_KEY", "dimension_key", "NUMBER", nullable=False),
                    DatasetColumn("SERVICE_LINE_NAME", "service_line", "VARCHAR(255)", nullable=False),
                ],
                [{"SERVICE_LINE_KEY": idx, "SERVICE_LINE_NAME": value} for idx, value in enumerate(shared, start=1)],
            ),
        ],
    )

    report = validate_dataset_quality(dataset)

    assert not report.ok
    assert any("Unrelated dimensions share nearly identical value sets" in issue.message for issue in report.issues)


def test_dataset_quality_blocks_mdf_story_without_mdf_measures():
    dataset = DatasetBundle(
        scenario=ScenarioContract(
            company_name="Wells Fargo",
            company_url="https://wellsfargo.com",
            use_case="Banking Marketing MDF",
            scenario_type="marketing_funnel",
            fact_grain="campaign-segment-month",
            business_problem="Track MDF spend and co-marketing partner performance.",
            metadata={"seed": 123},
        ),
        tables=[
            DatasetTable(
                "MARKETING_FUNNEL",
                "campaign-segment-month",
                [
                    DatasetColumn("MONTH_KEY", "month_key", "NUMBER", nullable=False),
                    DatasetColumn("SPEND_USD", "spend", "NUMBER(12,2)", nullable=False),
                    DatasetColumn("CONVERSIONS", "conversions", "NUMBER", nullable=False),
                ],
                [{"MONTH_KEY": 1, "SPEND_USD": 1000.0, "CONVERSIONS": 25}],
                is_fact=True,
            )
        ],
    )

    report = validate_dataset_quality(dataset)

    assert not report.ok
    assert "MDF-specific measures" in report.summary()


def test_slack_notification_called_when_fallback_events_exist(monkeypatch):
    import slack_notifier

    calls = []
    monkeypatch.setattr(slack_notifier, "notify_deployment_event", lambda *args, **kwargs: calls.append((args, kwargs)) or (True, "sent"))

    ok, message = slack_notifier.notify_data_fallback_event(
        company="Echo Global",
        use_case="Shipping Sales",
        scenario_type="shipping_sales",
        fallback_events=[{"dimension": "ACCOUNTS", "fallback_values": ["Blue Ridge Retail Group"]}],
    )

    assert ok is True
    assert message == "sent"
    assert calls
    assert calls[0][0][0] == "DemoPrep data fallback used"
    assert calls[0][0][1] == "Warning"


def test_dataset_quality_failure_notification(monkeypatch):
    import slack_notifier

    calls = []
    monkeypatch.setattr(slack_notifier, "notify_deployment_event", lambda *args, **kwargs: calls.append((args, kwargs)) or (True, "sent"))

    ok, message = slack_notifier.notify_dataset_quality_failure(
        company="Deloitte",
        use_case="Professional Services",
        scenario_type="professional_services_engagements",
        summary="BLOCKER: bad data",
    )

    assert ok is True
    assert message == "sent"
    assert calls[0][0][0] == "DemoPrep dataset quality gate blocked deploy"
    assert calls[0][0][1] == "Failed"


def test_dataset_preflight_renders_matrix_report():
    from demoprep_app.dataset.preflight import render_preflight_markdown, run_dataset_preflight

    results = run_dataset_preflight(scenario_types=["professional_services_engagements"], row_count_guidance=25)
    markdown = render_preflight_markdown(results)

    assert len(results) == 1
    assert results[0].scenario_type == "professional_services_engagements"
    assert results[0].ok
    assert "DemoPrep Dataset Preflight" in markdown


def _scenario(dimensions=None):
    return ScenarioContract(
        company_name="Echo Global",
        company_url="https://echo.com",
        use_case="Shipping Sales",
        scenario_type="shipping_sales",
        fact_grain="account-service-lane-month",
        demo_audience="Sales leaders",
        business_problem="Understand shipping revenue and margin.",
        metric_formulas={},
        dashboard_questions=[],
        seasonality=[],
        dimensions=dimensions or [],
        metadata={"seed": 123},
    )