Spaces:
Running
Running
fix(e2e): remove remaining hardcoded DEMOBUILD refs in test harnesses
Browse filesFollow-up to the rotating-demo-database change: the data-grading fallback,
geo-scope check, and settings_test row check still pointed at DEMOBUILD.
All now derive the database from the model TML (via check_snowflake_schema's
new database field) or get_demo_database().
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
- tests/e2e_quality.py +8 -5
- tests/settings_test.py +4 -3
tests/e2e_quality.py
CHANGED
|
@@ -1074,7 +1074,8 @@ def run_ai_grading(run_context: dict, company: str, vertical: str, line: str, fu
|
|
| 1074 |
db = db or related_db
|
| 1075 |
schema = schema or related_schema
|
| 1076 |
if (not db or not schema) and schema_override:
|
| 1077 |
-
|
|
|
|
| 1078 |
print(f" ℹ️ Using schema resolved from model: {schema_override}")
|
| 1079 |
sample = get_snowflake_sample(db, schema) if db and schema else "Could not determine db/schema"
|
| 1080 |
print(" 🤖 Grading data quality...")
|
|
@@ -1213,16 +1214,17 @@ def verify_group1_settings(result: dict) -> dict:
|
|
| 1213 |
conn = get_snowflake_connection()
|
| 1214 |
cursor = conn.cursor()
|
| 1215 |
schema = sf["schema"]
|
|
|
|
| 1216 |
# Look for a column named COUNTRY, REGION, or STATE
|
| 1217 |
-
cursor.execute(f'SHOW TABLES IN SCHEMA
|
| 1218 |
tables = [row[1] for row in cursor.fetchall()]
|
| 1219 |
foreign_found = False
|
| 1220 |
checked = False
|
| 1221 |
for tname in tables:
|
| 1222 |
-
cursor.execute(f'SHOW COLUMNS IN TABLE
|
| 1223 |
cols = [row[2].upper() for row in cursor.fetchall()]
|
| 1224 |
if "COUNTRY" in cols:
|
| 1225 |
-
cursor.execute(f'SELECT DISTINCT "COUNTRY" FROM
|
| 1226 |
countries = [row[0] for row in cursor.fetchall() if row[0]]
|
| 1227 |
non_us = [c for c in countries if c not in ("USA", "US", "United States", "United States of America")]
|
| 1228 |
foreign_found = len(non_us) > 0
|
|
@@ -1634,7 +1636,8 @@ def check_snowflake_schema(company: str, start_time: float, schema_override: str
|
|
| 1634 |
|
| 1635 |
cursor.close()
|
| 1636 |
conn.close()
|
| 1637 |
-
return {"found": True, "
|
|
|
|
| 1638 |
|
| 1639 |
except Exception as e:
|
| 1640 |
return {"found": False, "error": str(e)}
|
|
|
|
| 1074 |
db = db or related_db
|
| 1075 |
schema = schema or related_schema
|
| 1076 |
if (not db or not schema) and schema_override:
|
| 1077 |
+
from snowflake_auth import get_demo_database
|
| 1078 |
+
db, schema = get_demo_database(), schema_override
|
| 1079 |
print(f" ℹ️ Using schema resolved from model: {schema_override}")
|
| 1080 |
sample = get_snowflake_sample(db, schema) if db and schema else "Could not determine db/schema"
|
| 1081 |
print(" 🤖 Grading data quality...")
|
|
|
|
| 1214 |
conn = get_snowflake_connection()
|
| 1215 |
cursor = conn.cursor()
|
| 1216 |
schema = sf["schema"]
|
| 1217 |
+
database = sf["database"]
|
| 1218 |
# Look for a column named COUNTRY, REGION, or STATE
|
| 1219 |
+
cursor.execute(f'SHOW TABLES IN SCHEMA "{database}"."{schema}"')
|
| 1220 |
tables = [row[1] for row in cursor.fetchall()]
|
| 1221 |
foreign_found = False
|
| 1222 |
checked = False
|
| 1223 |
for tname in tables:
|
| 1224 |
+
cursor.execute(f'SHOW COLUMNS IN TABLE "{database}"."{schema}"."{tname}"')
|
| 1225 |
cols = [row[2].upper() for row in cursor.fetchall()]
|
| 1226 |
if "COUNTRY" in cols:
|
| 1227 |
+
cursor.execute(f'SELECT DISTINCT "COUNTRY" FROM "{database}"."{schema}"."{tname}" LIMIT 20')
|
| 1228 |
countries = [row[0] for row in cursor.fetchall() if row[0]]
|
| 1229 |
non_us = [c for c in countries if c not in ("USA", "US", "United States", "United States of America")]
|
| 1230 |
foreign_found = len(non_us) > 0
|
|
|
|
| 1636 |
|
| 1637 |
cursor.close()
|
| 1638 |
conn.close()
|
| 1639 |
+
return {"found": True, "database": database, "schema": schema,
|
| 1640 |
+
"tables": table_info, "total_rows": total_rows}
|
| 1641 |
|
| 1642 |
except Exception as e:
|
| 1643 |
return {"found": False, "error": str(e)}
|
tests/settings_test.py
CHANGED
|
@@ -88,14 +88,15 @@ def write_setting(key: str, value: str):
|
|
| 88 |
def check_snowflake_rows(schema_name: str) -> dict:
|
| 89 |
"""Return {table: row_count} for every table in the schema."""
|
| 90 |
try:
|
| 91 |
-
from snowflake_auth import get_snowflake_connection
|
| 92 |
conn = get_snowflake_connection()
|
| 93 |
cursor = conn.cursor()
|
| 94 |
-
|
|
|
|
| 95 |
tables = [row[1] for row in cursor.fetchall()]
|
| 96 |
counts = {}
|
| 97 |
for t in tables:
|
| 98 |
-
cursor.execute(f'SELECT COUNT(*) FROM
|
| 99 |
counts[t] = cursor.fetchone()[0]
|
| 100 |
cursor.close()
|
| 101 |
conn.close()
|
|
|
|
| 88 |
def check_snowflake_rows(schema_name: str) -> dict:
|
| 89 |
"""Return {table: row_count} for every table in the schema."""
|
| 90 |
try:
|
| 91 |
+
from snowflake_auth import get_snowflake_connection, get_demo_database
|
| 92 |
conn = get_snowflake_connection()
|
| 93 |
cursor = conn.cursor()
|
| 94 |
+
database = get_demo_database() # demos live in the rotating <base>_<YYYY_MM> DB
|
| 95 |
+
cursor.execute(f'SHOW TABLES IN SCHEMA "{database}"."{schema_name}"')
|
| 96 |
tables = [row[1] for row in cursor.fetchall()]
|
| 97 |
counts = {}
|
| 98 |
for t in tables:
|
| 99 |
+
cursor.execute(f'SELECT COUNT(*) FROM "{database}"."{schema_name}"."{t}"')
|
| 100 |
counts[t] = cursor.fetchone()[0]
|
| 101 |
cursor.close()
|
| 102 |
conn.close()
|