"""Integration preflight — run FIRST, before touching chat_interface.py. Exercises the seam between the new blueprint pipeline and your real repo classes (ScenarioContract, DatasetBundle/Column/Table, the DDL compiler, the Snowflake loader signature) with NO LLM and NO Snowflake connection. All-green means the risky part of the integration is done. PYTHONPATH=. python tests/integration_preflight.py """ from __future__ import annotations import inspect import sys import traceback results: list[tuple[str, bool]] = [] def check(name): def wrap(fn): print(f"\n-- {name} " + "-" * max(0, 55 - len(name))) try: ok, note = fn() except Exception as e: ok, note = False, f"{type(e).__name__}: {e}" traceback.print_exc() print(f" [{'PASS' if ok else 'FAIL'}] {note}") results.append((name, ok)) return fn return wrap MINI_BLUEPRINT = { "company_name": "Preflight Co", "company_url": "preflight.example", "use_case": "wiring check", "business_domain": "generic", "demo_audience": "engineer", "business_problem": "verify integration", "date": {"grain": "month", "months_of_history": 12, "include_date_dimension": False}, "dimensions": [ {"name": "REGION", "attribute_columns": ["TIER"], "values": [ {"name": "Northwest", "attributes": {"TIER": "A"}, "performance": 1.3}, {"name": "Southwest", "attributes": {"TIER": "A"}}, {"name": "Midwest", "attributes": {"TIER": "B"}}, {"name": "Northeast", "attributes": {"TIER": "B"}, "performance": 0.8}]}, {"name": "PRODUCT_LINE", "values": ["Signature Blend", "Cold Brew Kit", "Single Origin", "Seasonal Reserve"]}, ], "facts": [{ "name": "MONTHLY_SALES", "grain": "one row per region per product line per month", "dimension_names": ["REGION", "PRODUCT_LINE"], "date_column": "MONTH_DATE", "measures": [ {"name": "UNITS", "kind": "base", "distribution": "lognormal", "params": {"mean": 500, "sigma": 0.5}, "fmt": "int"}, {"name": "REVENUE", "kind": "base", "distribution": "lognormal", "params": {"mean": 12000, "sigma": 0.6}, "fmt": "currency"}]}], "seasonality": {"monthly": {str(m): 1.0 for m in range(1, 13)}, "trend_pct_per_year": 6, "narrative": ["steady"]}, "insights": [ {"id": "nw", "headline": "Northwest leads", "insight_type": "segment_outlier", "dimension": "REGION", "dimension_value": "Northwest", "measure": "UNITS", "magnitude": 1.3, "spotter_question": "units by region", "expected_finding": "NW tallest"}, {"id": "ne", "headline": "Northeast trails", "insight_type": "segment_laggard", "dimension": "REGION", "dimension_value": "Northeast", "measure": "REVENUE", "magnitude": 0.75, "spotter_question": "revenue by region", "expected_finding": "NE lowest"}, {"id": "res", "headline": "Reserve overperforms", "insight_type": "segment_outlier", "dimension": "PRODUCT_LINE", "dimension_value": "Seasonal Reserve", "measure": "REVENUE", "magnitude": 1.4, "spotter_question": "revenue by product line", "expected_finding": "Reserve tallest"}], "dashboard_questions": ["Revenue by region?", "Units by product line?", "How is Northwest doing?", "Where is revenue weakest?"], } @check("1. Real dataset contracts import & construct") def _c1(): from demoprep_app.dataset.contracts import DatasetBundle, DatasetColumn, DatasetTable col = list(inspect.signature(DatasetColumn).parameters) tbl = list(inspect.signature(DatasetTable).parameters) print(f" DatasetColumn params: {col}") print(f" DatasetTable params: {tbl}") missing = ({"name", "data_type"} - set(col)) | ({"name", "columns", "rows"} - set(tbl)) if missing: return False, f"engine.py expects fields not present: {missing}" if not hasattr(DatasetBundle, "table_map"): return False, "DatasetBundle has no table_map(); add it or adjust callers." return True, "column/table/bundle compatible." @check("2. Real ScenarioContract accepts compat-shim kwargs") def _c2(): from demoprep_app.scenario.blueprint import DemoBlueprint from demoprep_app.dataset.engine import _compat_scenario from demoprep_app.scenario.contract import ScenarioContract c = _compat_scenario(DemoBlueprint.from_dict(MINI_BLUEPRINT, seed=1)) print(f" ScenarioContract params: {list(inspect.signature(ScenarioContract).parameters)}") ok = isinstance(c, ScenarioContract) return ok, "compat shim built a real ScenarioContract." if ok else "did not return a ScenarioContract." @check("3. Engine generates a coherent multi-table bundle") def _c3(): from demoprep_app.scenario.blueprint import DemoBlueprint from demoprep_app.dataset.engine import BlueprintEngine bp = DemoBlueprint.from_dict(MINI_BLUEPRINT, seed=7) if bp.problems(): return False, f"mini blueprint invalid: {bp.problems()}" bundle = BlueprintEngine().generate(bp, row_count=3000) names = [t.name for t in bundle.tables] fact = bundle.table_map()["MONTHLY_SALES"] print(f" tables: {names}; fact rows: {len(fact.rows)}; sample: {fact.rows[0]}") if set(names) != {"REGION", "PRODUCT_LINE", "MONTHLY_SALES"}: return False, f"unexpected tables: {names}" if "MONTH_DATE" not in fact.rows[0]: return False, "custom date_column not honored" return True, f"{len(bundle.tables)} tables, {len(fact.rows)} fact rows." @check("4. Validator proves planted insights") def _c4(): from demoprep_app.scenario.blueprint import DemoBlueprint from demoprep_app.dataset.engine import BlueprintEngine from demoprep_app.dataset.validator import validate_bundle bp = DemoBlueprint.from_dict(MINI_BLUEPRINT, seed=7) report = validate_bundle(bp, BlueprintEngine().generate(bp, row_count=3000)) for line in report.summary().splitlines(): print(" " + line) return report.passed, "insights demo-visible." if report.passed else "validation did not pass." @check("5. Real DDL compiler consumes the bundle") def _c5(): from demoprep_app.scenario.blueprint import DemoBlueprint from demoprep_app.dataset.engine import BlueprintEngine from demoprep_app.ddl import DatasetDdlCompiler bp = DemoBlueprint.from_dict(MINI_BLUEPRINT, seed=7) ddl = DatasetDdlCompiler().compile(BlueprintEngine().generate(bp, row_count=500)) creates = ddl.upper().count("CREATE TABLE") print(f" DDL length: {len(ddl)}; CREATE TABLE count: {creates}") ok = creates >= 3 return ok, f"compiler produced DDL for {creates} tables." if ok else f"expected >=3, got {creates}." @check("6. Snowflake loader signature matches call site") def _c6(): from demoprep_app.integrations.snowflake import populate_dataset_bundle params = list(inspect.signature(populate_dataset_bundle).parameters) print(f" populate_dataset_bundle params: {params}") ok = len(params) >= 3 return ok, "loader signature compatible." if ok else f"expected >=3 params, got {params}" @check("7. build_demo entry point importable (LLM not invoked)") def _c7(): from demoprep_app.pipeline.build_demo import build_demo, DemoBuild # noqa: F401 sig = list(inspect.signature(build_demo).parameters) print(f" build_demo params: {sig}") missing = [r for r in ("company_name", "company_url", "use_case", "user_request", "llm_model") if r not in sig] return not missing, "build_demo present with expected signature." if not missing else f"missing params: {missing}" def main(): print("Integration preflight (no LLM, no Snowflake)") _c1(); _c2(); _c3(); _c4(); _c5(); _c6(); _c7() passed = sum(1 for _, ok in results if ok) print("\n" + "=" * 55) for name, ok in results: print(f" [{'PASS' if ok else 'FAIL'}] {name}") print("=" * 55) if passed == len(results): print(f"ALL {len(results)} PASSED - safe to wire chat_interface.py (see SINGLE_PIPELINE.md).") return 0 print(f"{passed}/{len(results)} passed. Most failures point at a field-name mismatch " "in engine._compat_scenario or the DatasetColumn/DatasetTable constructors.") return 1 if __name__ == "__main__": sys.exit(main())