diff --git a/tests/test_smoke_all_services.py b/tests/test_smoke_all_services.py new file mode 100644 index 0000000..fbbe0f9 --- /dev/null +++ b/tests/test_smoke_all_services.py @@ -0,0 +1,125 @@ +""" +Дымовой тест: create -> modify -> delete для ВСЕХ сервисов в полигоне (DEV/TEST/PROD). + +Запуск: python3 tests/test_smoke_all_services.py +""" +import sys, os, time +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'site')) + +import requests + +POLYGON = "https://polygon.pythonk8s.dev.nubes.ru" +STANDS = ["dev", "test", "prod"] +PREFIX = "autotest-smoke-" + +results = {} + + +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 _get(base, path): + r = requests.get(f"{base}{path}", timeout=10) + r.raise_for_status() + return r.json() + + +def _uid(r): + loc = r.headers.get("Location", "") + return loc.rstrip("/").split("/")[-1] + + +def _poll(base, op_uid, timeout=5): + 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") + + +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)}") + + +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(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}")