78 lines
2.9 KiB
Python
78 lines
2.9 KiB
Python
#!/usr/bin/env python3
|
||
"""Smoke checks для Enduro Trails. Запуск: python scripts/smoke_check.py"""
|
||
import sys, sqlite3, os, json, urllib.request, urllib.error
|
||
|
||
DB_PATH = os.environ.get("DATA_PATH", "../data/centralfederal.sqlite")
|
||
API_BASE = os.environ.get("API_BASE", "http://localhost:5558")
|
||
errors = []
|
||
|
||
print("==> Схема БД...")
|
||
try:
|
||
conn = sqlite3.connect(DB_PATH)
|
||
cur = conn.cursor()
|
||
|
||
cur.execute("SELECT name FROM sqlite_master WHERE type='table'")
|
||
tables = {r[0] for r in cur.fetchall()}
|
||
for t in ("trails", "poi"):
|
||
if t not in tables: errors.append(f"Таблица {t} не найдена")
|
||
|
||
cur.execute("PRAGMA table_info(trails)")
|
||
trail_cols = {r[1] for r in cur.fetchall()}
|
||
for col in ("min_lon","max_lon","min_lat","max_lat","length_m","mtb_scale"):
|
||
if col not in trail_cols: errors.append(f"trails.{col} не найдена")
|
||
|
||
cur.execute("PRAGMA table_info(poi)")
|
||
poi_cols = {r[1] for r in cur.fetchall()}
|
||
for col in ("lon","lat"):
|
||
if col not in poi_cols: errors.append(f"poi.{col} не найдена")
|
||
|
||
cur.execute("SELECT name FROM sqlite_master WHERE type='index'")
|
||
indexes = {r[0] for r in cur.fetchall()}
|
||
for idx in ("idx_trails_bbox","idx_poi_coords"):
|
||
if idx not in indexes: errors.append(f"Индекс {idx} не найден")
|
||
|
||
cur.execute("SELECT COUNT(*) FROM trails"); n = cur.fetchone()[0]
|
||
print(f" trails: {n:,}")
|
||
if n == 0: errors.append("trails пуста")
|
||
|
||
cur.execute("SELECT COUNT(*) FROM poi"); n = cur.fetchone()[0]
|
||
print(f" poi: {n:,}")
|
||
|
||
cur.execute("SELECT COUNT(*) FROM trails WHERE min_lon IS NULL")
|
||
null_bbox = cur.fetchone()[0]
|
||
if null_bbox > 0: errors.append(f"{null_bbox} trails с NULL bbox")
|
||
|
||
conn.close()
|
||
except Exception as e:
|
||
errors.append(f"Ошибка БД: {e}")
|
||
|
||
print("==> API...")
|
||
try:
|
||
with urllib.request.urlopen(f"{API_BASE}/api/health", timeout=5) as r:
|
||
data = json.loads(r.read())
|
||
if not data.get("db_exists"): errors.append("db_exists=false")
|
||
else: print(" health: OK")
|
||
|
||
tile_url = f"{API_BASE}/api/tiles/10/619/320.mvt"
|
||
with urllib.request.urlopen(tile_url, timeout=10) as r:
|
||
td = r.read()
|
||
if len(td) == 0: errors.append("Тайл z10/619/320 пустой")
|
||
else: print(f" tile z10/619/320: {len(td)} bytes OK")
|
||
|
||
try:
|
||
urllib.request.urlopen(f"{API_BASE}/api/tiles/99/0/0.mvt", timeout=5)
|
||
errors.append("z=99 должен вернуть 400")
|
||
except urllib.error.HTTPError as e:
|
||
if e.code == 400: print(" validation z=99: 400 OK")
|
||
else: errors.append(f"z=99 вернул {e.code}")
|
||
except Exception as e:
|
||
errors.append(f"Ошибка API: {e}")
|
||
|
||
print()
|
||
if errors:
|
||
print(f"FAILED ({len(errors)} ошибок):")
|
||
for e in errors: print(f" - {e}")
|
||
sys.exit(1)
|
||
else:
|
||
print("OK — все проверки прошли")
|