fix: empty table detection + test suite robustness

Empty tables (Wells Fargo, Best Buy):
- Extend empty-table check to ALL tables, not just fact tables
- is_fact_table heuristic (>=2 FKs) fails when DDL has no FOREIGN KEY
constraints — fact_table_names was empty so the check never triggered
- Now checks all non-date tables; 0-row dim table triggers retry
- Final check also covers dim tables: raises RuntimeError if still empty
- Add Step 3b generation summary (per-table row counts before write)
- Log schema name in populate log_end so fetch_run_diagnostics can find it
- Improved insert_rows logging when filtered_columns is empty

Test suite fixes:
- Liveboard Name selector: try textarea/input/generic aria-label variants
- Custom tab Context selector: 1.5s wait after vertical change + try
placeholder and aria-label selectors with visibility check
- Tag/share checks: guard against empty API response body before .json()

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

legitdata_bridge.py

@@ -197,7 +197,14 @@ class KeyPairSnowflakeWriter:
                 filtered_indices.append(i)
         
         if not filtered_columns:
-            print(f"Warning: No columns with values to insert for {table_name}")
+            # Log the first row so we can see what the generator produced
+            if rows:
+                sample = dict(zip(columns, rows[0]))
+                print(f"ERROR: No columns with non-None values to insert for {table_name}")
+                print(f"  columns expected: {columns[:8]}")
+                print(f"  first row sample: {str(sample)[:300]}")
+            else:
+                print(f"Warning: No rows to insert for {table_name}")
             return 0
         
         col_list = ', '.join(filtered_columns)
@@ -435,6 +442,11 @@ class DemoLegitGenerator:
             # Register PK values for FK references
             self._gen._register_pk_values(table, rows)
         
+        print("\n=== Step 3b: Generation Summary ===")
+        for tname, trows in generated_data.items():
+            status = "✅" if trows else "❌ EMPTY"
+            print(f"  {status} {tname}: {len(trows)} rows in memory")
+
         print("\n=== Step 4: Writing to Database ===")
         results = self._write_to_database(generated_data, truncate_first)
         
@@ -640,11 +652,25 @@ def populate_demo_data(
             if _slog:
                 _slog.log_verbose("populate", "first attempt failed", error=first_attempt_error)
 
-        # Check all tables for 0 rows even if no exception
+        # Check all tables for 0 rows even if no exception.
+        # NOTE: is_fact_table uses a FK-count heuristic; if DDL has no explicit FOREIGN KEY
+        # constraints (common in Snowflake schemas), ALL tables look like dim tables and
+        # fact_table_names is empty — so we must also check dim tables.
         fact_table_names = {t.name for t in schema.fact_tables}
+        all_table_names  = {t.name for t in schema.tables}
+        date_table_names = {t.name for t in schema.tables
+                            if t.name.upper() in ('DATES', 'DATE', 'DATE_DIM', 'DIM_DATE', 'CALENDAR')}
+
         empty_facts = [t for t in fact_table_names if results.get(t, 0) == 0]
-        if not first_attempt_error and empty_facts:
-            first_attempt_error = f"Fact table(s) empty after first attempt: {', '.join(empty_facts)}"
+        # Also catch dim tables (excluding date tables, which may legitimately be smaller)
+        empty_dims  = [t for t in (all_table_names - fact_table_names - date_table_names)
+                       if results.get(t, 0) == 0]
+
+        if not first_attempt_error and (empty_facts or empty_dims):
+            first_attempt_error = (
+                f"Empty table(s) after first attempt — "
+                f"facts: {empty_facts or 'none'}, dims: {empty_dims or 'none'}"
+            )
             log(f"⚠️ {first_attempt_error}")
 
         # Attempt 2 if anything failed
@@ -667,9 +693,12 @@ def populate_demo_data(
             log(f"  {status} {table_name}: {count:,} rows")
 
         empty_facts_final = [t for t in fact_table_names if results.get(t, 0) == 0]
-        if empty_facts_final:
+        empty_dims_final  = [t for t in (all_table_names - fact_table_names - date_table_names)
+                             if results.get(t, 0) == 0]
+        if empty_facts_final or empty_dims_final:
             raise RuntimeError(
-                f"Fact table(s) still empty after 2 attempts: {', '.join(empty_facts_final)}. "
+                f"Empty table(s) still present after 2 attempts — "
+                f"facts: {empty_facts_final or 'none'}, dims: {empty_dims_final or 'none'}. "
                 f"First attempt error: {first_attempt_error}"
             )
 
@@ -688,7 +717,9 @@ Data contextually generated for:
 """
         log(message)
         if _slog and _t_populate is not None:
-            _slog.log_end("populate", _t_populate, tables=len(results), total_rows=total_rows)
+            _slog.log_end("populate", _t_populate,
+                          tables=len(results), total_rows=total_rows,
+                          schema=schema_name)
         return True, message, results
 
     except Exception as e:

tests/e2e_quality.py

@@ -95,9 +95,14 @@ def _parse_json(text: str) -> dict:
 # ---------------------------------------------------------------------------
 # Test case generation
 # ---------------------------------------------------------------------------
-def generate_ai_test_case(config: dict) -> dict:
+def generate_ai_test_case(config: dict, exclude_list: list = None) -> dict:
     """AI picks vertical, line, function, and a matching company."""
     prompt = config["ai_generated"]["generation_prompt"]
+    if exclude_list:
+        prompt = prompt.replace(
+            "Do NOT pick: Nike, Wells Fargo.",
+            f"Do NOT pick: {', '.join(exclude_list)}."
+        )
     data = _parse_json(_llm(prompt))
     return {
         "name":        f"AI: {data['company']} — {data['vertical']} / {data['line']} / {data['function']}",
@@ -192,10 +197,9 @@ def build_test_suite(config: dict) -> list:
     ai_count = config["ai_generated"].get("count", 2)
     for i in range(ai_count):
         try:
-            tc = generate_ai_test_case(config)
-            # Deduplicate: if this company was already picked, regenerate once
+            tc = generate_ai_test_case(config, exclude_list=used_companies)
             if tc["company"] in used_companies:
-                tc = generate_ai_test_case(config)
+                tc = generate_ai_test_case(config, exclude_list=used_companies)
             used_companies.append(tc["company"])
             suite.append(tc)
         except Exception as e:
@@ -204,7 +208,9 @@ def build_test_suite(config: dict) -> list:
 
     for i in range(2):
         try:
-            suite.append(generate_custom_test_case(config, used_companies))
+            tc = generate_custom_test_case(config, used_companies)
+            used_companies.append(tc["company"])  # prevent second custom from picking same
+            suite.append(tc)
         except Exception as e:
             print(f"  ⚠️  Custom test case {i+1} failed ({e}), skipping")
 
@@ -280,6 +286,34 @@ def submit_job(page: Page, test_case: dict):
     page.get_by_role('tab', name='App', exact=True).click(timeout=10000)
     page.wait_for_timeout(1000)
 
+    # Set liveboard name FIRST — right panel is independent of the form tabs
+    lb_name = f"QA — {test_case['company']} {test_case.get('function', 'Demo')}"
+    try:
+        # Gradio Textbox with label="Liveboard Name" — try multiple selectors
+        lb_el = None
+        for sel in [
+            'textarea[aria-label="Liveboard Name"]',
+            'input[aria-label="Liveboard Name"]',
+            '[aria-label="Liveboard Name"]',
+        ]:
+            try:
+                el = page.locator(sel).first
+                if el.is_visible(timeout=2000):
+                    lb_el = el
+                    break
+            except Exception:
+                pass
+        if lb_el:
+            lb_el.scroll_into_view_if_needed(timeout=3000)
+            lb_el.click(click_count=3, timeout=3000)
+            lb_el.fill(lb_name)
+            page.wait_for_timeout(300)
+        else:
+            raise Exception("No matching element found")
+    except Exception:
+        lb_name += " (name not set)"
+        print(f"  ⚠️  Liveboard Name field not found — continuing without it")
+
     # Select TS environment first (required for deploy)
     select_gradio_dropdown(page, "TS Environment", test_case.get("ts_environment", "secloud - primary"))
 
@@ -287,8 +321,23 @@ def submit_job(page: Page, test_case: dict):
     select_gradio_dropdown(page, "Vertical", test_case["vertical"])
 
     if test_case["vertical"] == "* CUSTOM *":
-        # Custom mode: fill the Context textarea, skip Line/Function
-        ctx_el = page.locator('textarea[aria-label="Context"]').first
+        # Custom mode: wait for UI to settle after vertical dropdown change, then fill Context
+        page.wait_for_timeout(1500)
+        ctx_el = None
+        for sel in [
+            'textarea[placeholder="Any extra context for the demo..."]',
+            'textarea[aria-label="Context"]',
+            'input[aria-label="Context"]',
+        ]:
+            try:
+                el = page.locator(sel).first
+                if el.is_visible(timeout=3000):
+                    ctx_el = el
+                    break
+            except Exception:
+                pass
+        if ctx_el is None:
+            raise Exception('Context textarea not found — tried placeholder and aria-label="Context"')
         ctx_el.click(click_count=3, timeout=5000)
         ctx_el.fill(test_case.get("context", ""))
         page.wait_for_timeout(300)
@@ -302,17 +351,6 @@ def submit_job(page: Page, test_case: dict):
     url_el.fill(test_case["company_url"])
     page.wait_for_timeout(300)
 
-    # Set liveboard name so test runs are identifiable in ThoughtSpot
-    lb_name = f"QA — {test_case['company']} {test_case['function']}"
-    try:
-        lb_el = page.locator('input[aria-label="Liveboard Name"]').first
-        lb_el.click(click_count=3, timeout=3000)
-        lb_el.fill(lb_name)
-        page.wait_for_timeout(300)
-    except Exception:
-        lb_name += " (name not set)"
-        print(f"  ⚠️  Liveboard Name field not found — continuing without it")
-
     # Click GO
     page.click('button:has-text("→ GO")', timeout=10000)
     print(f"  ✅ Form submitted: {test_case['vertical']} / {test_case['line']} / {test_case['function']} — {test_case['company_url']} | lb: {lb_name}")
@@ -439,18 +477,51 @@ def get_snowflake_sample(db: str, schema: str) -> str:
         cursor = conn.cursor()
         cursor.execute(f'SHOW TABLES IN SCHEMA "{db}"."{schema}"')
         tables = [row[1] for row in cursor.fetchall()]
-        parts  = []
-        for table in tables[:6]:
+
+        # Count rows in every table first — so we can prioritize the fact table
+        row_counts = {}
+        for table in tables:
+            try:
+                cursor.execute(f'SELECT COUNT(*) FROM "{db}"."{schema}"."{table}"')
+                row_counts[table] = cursor.fetchone()[0]
+            except Exception:
+                row_counts[table] = 0
+
+        # Sort descending — fact table (most rows) sampled first
+        tables_sorted = sorted(tables, key=lambda t: row_counts.get(t, 0), reverse=True)
+
+        # Build row-count summary header so grader knows what's populated
+        header = ["Table row counts:"]
+        for t in tables_sorted:
+            header.append(f"  {t}: {row_counts.get(t, 0)} rows")
+        empty_tables = [t for t in tables_sorted if row_counts.get(t, 0) == 0]
+        if empty_tables:
+            header.append(
+                f"\n⚠️  WARNING: {len(empty_tables)} table(s) have 0 rows: "
+                f"{', '.join(empty_tables)}"
+            )
+
+        parts = ["\n".join(header)]
+
+        # Sample from tables that actually have data (up to 6); fall back to first 3 if all empty
+        tables_with_data = [t for t in tables_sorted if row_counts.get(t, 0) > 0]
+        to_sample = tables_with_data[:6] if tables_with_data else tables_sorted[:3]
+
+        for table in to_sample:
             try:
                 cursor.execute(f'SELECT * FROM "{db}"."{schema}"."{table}" LIMIT 30')
                 cols = [d[0] for d in cursor.description]
                 rows = cursor.fetchall()
-                parts.append(f"\nTable: {table}  ({len(rows)} sample rows)")
+                parts.append(
+                    f"\nTable: {table}  "
+                    f"({row_counts.get(table, 0)} total rows, {len(rows)} sampled)"
+                )
                 parts.append(f"Columns: {', '.join(cols)}")
                 for row in rows[:15]:
                     parts.append("  " + str(dict(zip(cols, row))))
             except Exception as e:
                 parts.append(f"\nTable: {table} — error: {e}")
+
         cursor.close()
         conn.close()
         return "\n".join(parts) if parts else "No tables found"
@@ -483,18 +554,23 @@ Analytics function: {function}
 The goal is a compelling demo with realistic data, outliers that drive a narrative,
 and a schema that supports the key KPIs for this use case.
 
-MODEL TML:
-{model_tml[:3000]}
+MODEL TML (full schema, column definitions, and relationships):
+{model_tml[:7000]}
 
-SNOWFLAKE SAMPLE DATA:
-{sample_data[:3000]}
+SNOWFLAKE DATA — actual row counts and sample rows:
+{sample_data[:4000]}
 
-Grade 0–100:
-1. REALISM (20 pts): Values look like real {company} data at scale.
-2. STORY POTENTIAL (30 pts): Outliers and trends exist that would anchor a demo conversation.
-3. TIME COVERAGE (20 pts): 12–24 months of history with meaningful trends.
-4. SCHEMA FITNESS (15 pts): Star schema supports the right KPIs for {line} {function}.
-5. COMPLETENESS (15 pts): Tables fully populated, dimensions have 20+ members.
+Grade 0–100 using the actual data above. Do NOT hedge with phrases like "constrained by
+partial TML" or "missing sample data" — the full TML and real row counts are provided.
+Score based on what you can observe.
+
+1. REALISM (20 pts): Values look like real {company} data at realistic scale and ranges.
+2. STORY POTENTIAL (30 pts): Outliers, trends, or anomalies exist that anchor a demo narrative.
+3. TIME COVERAGE (20 pts): 12–24 months of history with meaningful trends over time.
+4. SCHEMA FITNESS (15 pts): Star schema design supports the key KPIs for {line} {function}.
+5. COMPLETENESS (15 pts): Tables are populated. Dimensions have 20+ distinct members.
+   RULE: If the row counts above show any key table at 0 rows, score COMPLETENESS = 0 for
+   that criteria. If the fact table is 0 rows, also deduct heavily from STORY POTENTIAL.
 
 Return ONLY valid JSON:
 {{"score": 0, "reasoning": "...", "strengths": ["..."], "weaknesses": ["..."]}}"""
@@ -750,22 +826,25 @@ def verify_group1_settings(result: dict) -> dict:
     if expected_tag and ts_base and model_guid:
         try:
             session = ts_authenticate(ts_base)
-            # v1 API: metadata/list returns tags per object
             resp = session.get(
                 f"{ts_base}/tspublic/v1/metadata/list",
                 params={"type": "LOGICAL_TABLE", "batchsize": 1,
                         "offset": 0, "pattern": model_guid},
             )
-            headers_data = resp.json().get("headers", [])
-            obj_tags = []
-            for h in headers_data:
-                if h.get("id") == model_guid:
-                    obj_tags = [t.get("name", "") for t in (h.get("tags") or [])]
-                    break
-            checks["tag_name"] = {
-                "expected": expected_tag, "actual": obj_tags,
-                "pass": expected_tag in obj_tags,
-            }
+            body = resp.text.strip()
+            if not body:
+                checks["tag_name"] = {"pass": None, "note": "tag check skipped: empty API response"}
+            else:
+                headers_data = resp.json().get("headers", [])
+                obj_tags = []
+                for h in headers_data:
+                    if h.get("id") == model_guid:
+                        obj_tags = [t.get("name", "") for t in (h.get("tags") or [])]
+                        break
+                checks["tag_name"] = {
+                    "expected": expected_tag, "actual": obj_tags,
+                    "pass": expected_tag in obj_tags,
+                }
         except Exception as e:
             checks["tag_name"] = {"pass": None, "note": f"tag check failed: {e}"}
 
@@ -779,15 +858,19 @@ def verify_group1_settings(result: dict) -> dict:
                 json={"metadata": [{"type": "LOGICAL_TABLE", "identifier": model_guid}]},
                 timeout=15,
             )
-            perms = resp.json()
-            principals = []
-            for item in (perms if isinstance(perms, list) else []):
-                for p in (item.get("permissions") or []):
-                    principals.append(p.get("principal", {}).get("name", ""))
-            checks["share_with"] = {
-                "expected": expected_share, "actual": principals,
-                "pass": any(expected_share.lower() in p.lower() for p in principals),
-            }
+            body = resp.text.strip()
+            if not body:
+                checks["share_with"] = {"pass": None, "note": "share check skipped: empty API response"}
+            else:
+                perms = resp.json()
+                principals = []
+                for item in (perms if isinstance(perms, list) else []):
+                    for p in (item.get("permissions") or []):
+                        principals.append(p.get("principal", {}).get("name", ""))
+                checks["share_with"] = {
+                    "expected": expected_share, "actual": principals,
+                    "pass": any(expected_share.lower() in p.lower() for p in principals),
+                }
         except Exception as e:
             checks["share_with"] = {"pass": None, "note": f"share check failed: {e}"}
 
@@ -911,29 +994,32 @@ def fetch_run_diagnostics(start_time: float) -> dict:
         return {"found": False, "reason": f"Diagnostics query failed: {e}"}
 
 
-def check_snowflake_schema(company: str, start_time: float) -> dict:
+def check_snowflake_schema(company: str, start_time: float, schema_override: str = None) -> dict:
     """
-    Look for a Snowflake schema matching this company created around test time.
-    Schema naming: {PREFIX}_{MMDDHHMI}_{RAND}_sch  e.g. WEL_04280450_B10_sch
+    Get row counts for the Snowflake schema created during this test run.
+    Uses schema_override (from session_logs) if available — no guessing needed.
+    Falls back to prefix+date matching only if session_logs didn't record the schema.
     """
     try:
         from snowflake_auth import get_snowflake_connection
         from datetime import datetime
 
-        prefix   = company[:3].upper()
-        date_str = datetime.fromtimestamp(start_time).strftime("%m%d")
-
         conn   = get_snowflake_connection()
         cursor = conn.cursor()
-        cursor.execute("SHOW SCHEMAS IN DATABASE DEMOBUILD")
-        all_schemas = [row[1] for row in cursor.fetchall()]
-
-        matching = [s for s in all_schemas if s.startswith(prefix) and date_str in s]
-        if not matching:
-            cursor.close(); conn.close()
-            return {"found": False, "prefix": prefix, "date": date_str}
 
-        schema = sorted(matching)[-1]  # most recent
+        if schema_override:
+            schema = schema_override
+        else:
+            # Fall back: guess from company prefix + date
+            prefix   = company[:3].upper()
+            date_str = datetime.fromtimestamp(start_time).strftime("%m%d")
+            cursor.execute("SHOW SCHEMAS IN DATABASE DEMOBUILD")
+            all_schemas = [row[1] for row in cursor.fetchall()]
+            matching = [s for s in all_schemas if s.startswith(prefix) and date_str in s]
+            if not matching:
+                cursor.close(); conn.close()
+                return {"found": False, "prefix": prefix, "date": date_str}
+            schema = sorted(matching)[-1]  # most recent
         cursor.execute(f'SHOW TABLES IN SCHEMA DEMOBUILD."{schema}"')
         tables = cursor.fetchall()
 
@@ -1063,21 +1149,23 @@ def run_single_test(page: Page, test_case: dict, config: dict) -> dict:
 
     result["duration_seconds"] = round(time.time() - start)
 
-    # Always check Snowflake row counts — included in JSON every run
-    sf = check_snowflake_schema(result["company"], start)
+    # Always fetch session_logs — gives us the real schema name + any errors
+    diag = fetch_run_diagnostics(start)
+    result["diagnostics"] = diag
+
+    # Use the real schema name from session_logs; fall back to prefix-guessing only if missing
+    known_schema = diag.get("snowflake_schema") if diag.get("found") else None
+    sf = check_snowflake_schema(result["company"], start, schema_override=known_schema)
     result["snowflake_check"] = sf
     if sf.get("found"):
         print(f"  📦 Snowflake: {sf['schema']}  ({sf['total_rows']} rows, {len(sf['tables'])} tables)")
         for t in sf["tables"]:
             print(f"    {t['table']}: {t['rows']} rows")
     else:
-        print(f"  📦 Snowflake: no schema found (prefix={sf.get('prefix', '?')}, date={sf.get('date', '?')})")
+        print(f"  📦 Snowflake: schema not found (logged as: {known_schema or 'not in logs'})")
 
-    # Fetch Supabase diagnostics on timeout or error only
+    # Print diagnostics on timeout or error
     if result["timed_out"] or result["error"]:
-        print("  🔍 Fetching failure diagnostics...")
-        diag = fetch_run_diagnostics(start)
-        result["diagnostics"] = diag
         print_diagnostics(diag, {})  # sf already printed above
 
     # Stage scoring
@@ -1160,7 +1248,7 @@ def run_quality_suite(max_tests: int = None):
         print("✅ Logged in\n")
 
         for i, test_case in enumerate(suite, 1):
-            label = {"fixed": "🔒", "random": "🎲", "ai_generated": "🤖"}[test_case["type"]]
+            label = {"fixed": "🔒", "random": "🎲", "ai_generated": "🤖", "custom": "✏️"}[test_case["type"]]
             print(f"{'─'*62}")
             print(f"[{i}/{len(suite)}] {label} {test_case['name']}")
 
@@ -1207,7 +1295,7 @@ def run_quality_suite(max_tests: int = None):
     print(f"\n{'='*62}")
     print(f"  COMPLETE  —  Avg: {avg}/100  Grade: {grade}")
     for r in results:
-        label = {"fixed": "🔒", "random": "🎲", "ai_generated": "🤖"}[r["type"]]
+        label = {"fixed": "🔒", "random": "🎲", "ai_generated": "🤖", "custom": "✏️"}[r["type"]]
         ag    = r["ai_grading"]
         ds    = f"{ag['data_score']}/100" if ag.get("data_score") is not None else "n/a"
         ls    = f"{ag['liveboard_score']}/100" if ag.get("liveboard_score") is not None else "n/a"