test: дымовой create->modify->delete для всех сервисов полигона

test_smoke_all_services.py — прогоняет все 3 стенда (dev/test/prod), для каждого сервиса: create -> modify -> delete, с поллингом.
2026-08-04 08:54:33 +04:00
parent b89c0138e4
commit 1052cf3506
1 changed files with 125 additions and 0 deletions
@@ -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}")