diff --git a/tests/test_smoke_all_services.py b/tests/test_smoke_all_services.py index fbbe0f9..1793801 100644 --- a/tests/test_smoke_all_services.py +++ b/tests/test_smoke_all_services.py @@ -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'}")