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}, )