test: дымовой тест АВТОТЕСТА (не полигона) — create + история
This commit is contained in:
@@ -1,125 +1,83 @@
|
||||
"""
|
||||
Дымовой тест: create -> modify -> delete для ВСЕХ сервисов в полигоне (DEV/TEST/PROD).
|
||||
Дымовой тест АВТОТЕСТА в эмуляции: create через /api/test, история через /api/history.
|
||||
|
||||
Запуск: python3 tests/test_smoke_all_services.py
|
||||
Бьёт в РАЗВЁРНУТЫЙ автотест (atest.pythonk8s.dev.nubes.ru), эмуляция TEST.
|
||||
Операции идут через app-autotest -> executor -> save_run -> история в БД.
|
||||
"""
|
||||
import sys, os, time
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'site'))
|
||||
import time, requests
|
||||
|
||||
import requests
|
||||
|
||||
POLYGON = "https://polygon.pythonk8s.dev.nubes.ru"
|
||||
STANDS = ["dev", "test", "prod"]
|
||||
PREFIX = "autotest-smoke-"
|
||||
|
||||
results = {}
|
||||
BASE = "https://atest.pythonk8s.dev.nubes.ru"
|
||||
COOKIES = {"mode": "polygon", "polygon_stand": "test"}
|
||||
|
||||
|
||||
def _post(base, path, json_data=None):
|
||||
r = requests.post(f"{base}{path}", json=json_data or {}, timeout=10)
|
||||
if r.status_code not in (200, 201):
|
||||
raise Exception(f"HTTP {r.status_code}: {r.text[:100]}")
|
||||
return r
|
||||
def test_services():
|
||||
r = requests.get(f"{BASE}/api/services", cookies=COOKIES, timeout=10)
|
||||
assert r.status_code == 200
|
||||
svc = r.json()
|
||||
assert len(svc) > 0
|
||||
return svc
|
||||
|
||||
|
||||
def _get(base, path):
|
||||
r = requests.get(f"{base}{path}", timeout=10)
|
||||
r.raise_for_status()
|
||||
return r.json()
|
||||
def test_create(svc_id):
|
||||
r = requests.post(f"{BASE}/api/test",
|
||||
json={"serviceId": svc_id, "operation": "create", "svcOperationId": 18,
|
||||
"params": {}, "displayName": f"autotest-smoke-{svc_id}"},
|
||||
cookies=COOKIES, timeout=30)
|
||||
assert r.status_code == 200, f"Create failed: {r.text[:200]}"
|
||||
data = r.json()
|
||||
assert data["status"] == "RUNNING"
|
||||
return data["opUid"], data.get("instanceUid")
|
||||
|
||||
|
||||
def _uid(r):
|
||||
loc = r.headers.get("Location", "")
|
||||
return loc.rstrip("/").split("/")[-1]
|
||||
|
||||
|
||||
def _poll(base, op_uid, timeout=5):
|
||||
def poll(op_uid, timeout=10):
|
||||
deadline = time.time() + timeout
|
||||
while time.time() < deadline:
|
||||
data = _get(base, f"/instanceOperations/{op_uid}?fields=dtFinish,isSuccessful")
|
||||
op = data.get("instanceOperation", {})
|
||||
if op.get("dtFinish"):
|
||||
return op.get("isSuccessful", False)
|
||||
time.sleep(0.1)
|
||||
raise Exception("poll timeout")
|
||||
r = requests.get(f"{BASE}/api/test/status/{op_uid}", cookies=COOKIES, timeout=5)
|
||||
d = r.json()
|
||||
if d.get("done") or d.get("status") in ("OK", "FAIL"):
|
||||
return d
|
||||
time.sleep(0.5)
|
||||
return {"status": "TIMEOUT"}
|
||||
|
||||
|
||||
def test_stand(stand):
|
||||
base = f"{POLYGON}/{stand}/api/v1/svc"
|
||||
print(f"\n{'='*50}")
|
||||
print(f"Стенд: {stand.upper()}")
|
||||
print(f"{'='*50}")
|
||||
|
||||
services = _get(base, "/services").get("results", [])
|
||||
print(f"Сервисов: {len(services)}")
|
||||
|
||||
ok = fail = skip = 0
|
||||
|
||||
for svc in services:
|
||||
sid = svc["svcId"]
|
||||
sname = svc["svc"]
|
||||
|
||||
try:
|
||||
ops = _get(base, f"/services/{sid}").get("svc", {}).get("operations", [])
|
||||
create_op = next((o for o in ops if o["operation"] == "create"), None)
|
||||
modify_op = next((o for o in ops if o["operation"] == "modify"), None)
|
||||
delete_op = next((o for o in ops if o["operation"] == "delete"), None)
|
||||
except Exception as e:
|
||||
print(f" FAIL {sid:3d} {sname:<35s} — ops: {e}")
|
||||
fail += 1
|
||||
continue
|
||||
|
||||
if not create_op or not delete_op:
|
||||
print(f" SKIP {sid:3d} {sname:<35s} — нет create/delete")
|
||||
skip += 1
|
||||
continue
|
||||
|
||||
try:
|
||||
r = _post(base, "/instances", {"serviceId": sid, "displayName": f"{PREFIX}{sid}"})
|
||||
uid = _uid(r)
|
||||
|
||||
if modify_op:
|
||||
r = _post(base, "/instanceOperations",
|
||||
{"instanceUid": uid, "operation": "modify", "svcOperationId": modify_op["svcOperationId"]})
|
||||
mop = _uid(r)
|
||||
_post(base, f"/instanceOperations/{mop}/run")
|
||||
if not _poll(base, mop):
|
||||
raise Exception("modify failed")
|
||||
|
||||
r = _post(base, "/instanceOperations",
|
||||
{"instanceUid": uid, "operation": "delete", "svcOperationId": delete_op["svcOperationId"]})
|
||||
dop = _uid(r)
|
||||
_post(base, f"/instanceOperations/{dop}/run")
|
||||
if not _poll(base, dop):
|
||||
raise Exception("delete failed")
|
||||
|
||||
extra = " +modify" if modify_op else ""
|
||||
print(f" OK {sid:3d} {sname:<35s} create{extra} -> delete")
|
||||
ok += 1
|
||||
except Exception as e:
|
||||
print(f" FAIL {sid:3d} {sname:<35s} — {e}")
|
||||
fail += 1
|
||||
|
||||
results[stand] = {"ok": ok, "fail": fail, "skip": skip, "total": len(services)}
|
||||
print(f"\n OK={ok} FAIL={fail} SKIP={skip} TOTAL={len(services)}")
|
||||
def test_history():
|
||||
r = requests.get(f"{BASE}/api/history", cookies=COOKIES, timeout=10)
|
||||
assert r.status_code == 200
|
||||
rows = r.json()
|
||||
assert len(rows) > 0, "История пустая!"
|
||||
return rows
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
for s in STANDS:
|
||||
try:
|
||||
test_stand(s)
|
||||
except Exception as e:
|
||||
print(f" СТЕНД {s} УПАЛ: {e}")
|
||||
results[s] = {"error": str(e)}
|
||||
print("=" * 60)
|
||||
print("Дымовой тест АВТОТЕСТА (эмуляция TEST)")
|
||||
print("=" * 60)
|
||||
|
||||
print(f"\n{'='*50}")
|
||||
print("ИТОГО:")
|
||||
total_ok = total_fail = 0
|
||||
for s, r in results.items():
|
||||
if "error" in r:
|
||||
print(f" {s}: ОШИБКА — {r['error']}")
|
||||
else:
|
||||
print(f" {s}: OK={r['ok']} FAIL={r['fail']} SKIP={r['skip']}")
|
||||
total_ok += r["ok"]
|
||||
total_fail += r["fail"]
|
||||
print(f" ВСЕГО: OK={total_ok} FAIL={total_fail}")
|
||||
svc = test_services()
|
||||
print(f"Сервисов: {len(svc)}")
|
||||
|
||||
ok = fail = 0
|
||||
for s in svc[:3]:
|
||||
sid, sname = s["svcId"], s["svc"]
|
||||
try:
|
||||
op_uid, iuid = test_create(sid)
|
||||
result = poll(op_uid)
|
||||
if result.get("status") == "OK":
|
||||
print(f" OK {sid} {sname}")
|
||||
ok += 1
|
||||
else:
|
||||
print(f" FAIL {sid} {sname} -> {result}")
|
||||
fail += 1
|
||||
except Exception as e:
|
||||
print(f" FAIL {sid} {sname} -> {e}")
|
||||
fail += 1
|
||||
|
||||
print(f"\nCreate: OK={ok} FAIL={fail}")
|
||||
|
||||
rows = test_history()
|
||||
print(f"История: {len(rows)} записей")
|
||||
for r in rows[:3]:
|
||||
print(f" {r['created_at'][:19]} {r['op_name']:8s} {r['status']}")
|
||||
|
||||
print(f"\n{'='*60}")
|
||||
print(f"РЕЗУЛЬТАТ: {'PASS' if fail == 0 else 'FAIL'}")
|
||||
|
||||
Reference in New Issue
Block a user