naeel
58d54f2a95
- loader.py: STANDS dict, get_services/get_ops_index per stand, POLYGON_STANDS env
- mock_state.py: get_state(stand_id) — ленивая инициализация
- app.py: StandMiddleware — перезапись PATH_INFO до Flask-роутинга
- Все роуты: LocalProxy(state), LocalProxy(SERVICES/OPS_INDEX) через g.stand_id
- openapi.py: _build_servers() — по одному серверу на стенд
- services/stand{1,2,3}/ — 37 YAML на стенд из STANDS/dev и STANDS/test
- tests/test_api.py: +4 multi-stand теста (изоляция, fail-next, OpenAPI servers)
- Обратная совместимость: /api/v1/svc/... → стенд по умолчанию
420 lines
16 KiB
Python
420 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
test_api.py — smoke + edge-case тесты Polygon API.
|
|
|
|
Запуск: python3 tests/test_api.py
|
|
Требует: pip install requests
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
|
|
try:
|
|
import requests
|
|
except ImportError:
|
|
print("❌ Нужен requests: pip install requests")
|
|
sys.exit(1)
|
|
|
|
BASE = os.getenv("POLYGON_ENDPOINT", "https://polygon.pythonk8s.dev.nubes.ru")
|
|
AUTH = os.getenv("MOCK_AUTH_TOKEN", "test-token-123")
|
|
H = {"Content-Type": "application/json"}
|
|
MH = {**H, "X-Mock-Auth": AUTH}
|
|
|
|
passed = 0
|
|
failed = 0
|
|
errors = []
|
|
|
|
|
|
def t(name, fn):
|
|
global passed, failed
|
|
try:
|
|
fn()
|
|
passed += 1
|
|
print(f" ✅ {name}")
|
|
except AssertionError as e:
|
|
failed += 1
|
|
msg = f" ❌ {name}: {e}"
|
|
print(msg)
|
|
errors.append(msg)
|
|
except Exception as e:
|
|
failed += 1
|
|
msg = f" 💥 {name}: {type(e).__name__}: {e}"
|
|
print(msg)
|
|
errors.append(msg)
|
|
|
|
|
|
def st(r, expected, note=""):
|
|
extra = f" ({note})" if note else ""
|
|
assert r.status_code == expected, f"expected {expected}, got {r.status_code}: {r.text[:200]}{extra}"
|
|
|
|
|
|
def js(r):
|
|
try:
|
|
return r.json()
|
|
except Exception:
|
|
raise AssertionError(f"not JSON: {r.text[:200]}")
|
|
|
|
|
|
# ═══════════════════════════════════════
|
|
# Setup
|
|
# ═══════════════════════════════════════
|
|
print("=" * 60)
|
|
print(f"Polygon API Tests — {BASE}")
|
|
print("=" * 60)
|
|
|
|
r = requests.post(f"{BASE}/api/v1/svc/_mock/reset", headers=MH)
|
|
st(r, 200)
|
|
print()
|
|
|
|
# Состояние между тестами
|
|
svc_list = {}
|
|
inst_uid = None
|
|
op_uid = None
|
|
svc_op_id = None
|
|
fail_inst_uid = None
|
|
fail_op_uid = None
|
|
|
|
|
|
# ═══════════════════════════════════════
|
|
# 1. HEALTH
|
|
# ═══════════════════════════════════════
|
|
print("── 1. Health ──")
|
|
|
|
def test_health():
|
|
st(requests.get(f"{BASE}/health"), 200)
|
|
t("GET /health → 200 OK", test_health)
|
|
|
|
|
|
# ═══════════════════════════════════════
|
|
# 2. SERVICES
|
|
# ═══════════════════════════════════════
|
|
print("\n── 2. Services ──")
|
|
|
|
def test_svc_list():
|
|
global svc_list
|
|
d = js(requests.get(f"{BASE}/api/v1/svc/services"))
|
|
svc_list = d
|
|
assert "results" in d, "no results key"
|
|
assert len(d["results"]) >= 37, f"only {len(d['results'])} services"
|
|
t("GET /services → 200, >=37 сервисов", test_svc_list)
|
|
|
|
def test_svc_fields():
|
|
for s in svc_list["results"]:
|
|
for k in ("svcId", "svc", "svcShort"):
|
|
assert k in s, f"missing {k} in {s}"
|
|
t("GET /services → каждый имеет svcId, svc, svcShort", test_svc_fields)
|
|
|
|
def test_svc_detail():
|
|
d = js(requests.get(f"{BASE}/api/v1/svc/services/1"))
|
|
assert "svc" in d
|
|
assert "operations" in d["svc"]
|
|
assert len(d["svc"]["operations"]) > 0
|
|
t("GET /services/1 → 200 + operations", test_svc_detail)
|
|
|
|
def test_svc_404():
|
|
st(requests.get(f"{BASE}/api/v1/svc/services/99999"), 404)
|
|
t("GET /services/99999 → 404", test_svc_404)
|
|
|
|
|
|
# ═══════════════════════════════════════
|
|
# 3. INSTANCES
|
|
# ═══════════════════════════════════════
|
|
print("\n── 3. Instances ──")
|
|
|
|
def test_inst_empty():
|
|
d = js(requests.get(f"{BASE}/api/v1/svc/instances"))
|
|
assert d["total"] == 0
|
|
t("GET /instances (после reset) → total=0", test_inst_empty)
|
|
|
|
def test_inst_no_service_id():
|
|
st(requests.post(f"{BASE}/api/v1/svc/instances", json={}, headers=H), 400)
|
|
t("POST /instances (без serviceId) → 400", test_inst_no_service_id)
|
|
|
|
def test_inst_bad_service():
|
|
st(requests.post(f"{BASE}/api/v1/svc/instances", json={"serviceId": 99999}, headers=H), 404)
|
|
t("POST /instances (serviceId=99999) → 404", test_inst_bad_service)
|
|
|
|
def test_inst_create():
|
|
global inst_uid
|
|
r = requests.post(f"{BASE}/api/v1/svc/instances",
|
|
json={"serviceId": 1, "displayName": "test-bolvanka"}, headers=H)
|
|
st(r, 201)
|
|
assert "Location" in r.headers
|
|
d = js(r)
|
|
assert "instanceUid" in d
|
|
inst_uid = d["instanceUid"]
|
|
t("POST /instances → 201 + Location + instanceUid", test_inst_create)
|
|
|
|
def test_inst_list_one():
|
|
d = js(requests.get(f"{BASE}/api/v1/svc/instances"))
|
|
assert d["total"] == 1
|
|
t("GET /instances → total=1", test_inst_list_one)
|
|
|
|
def test_inst_pagesize_clamp():
|
|
d = js(requests.get(f"{BASE}/api/v1/svc/instances?pageSize=500"))
|
|
assert d["pageSize"] == 200
|
|
t("GET /instances?pageSize=500 → pageSize=200 (clamped)", test_inst_pagesize_clamp)
|
|
|
|
def test_inst_negative_page():
|
|
d = js(requests.get(f"{BASE}/api/v1/svc/instances?page=-1"))
|
|
assert d["page"] >= 1
|
|
t("GET /instances?page=-1 → не падает", test_inst_negative_page)
|
|
|
|
def test_inst_detail():
|
|
d = js(requests.get(f"{BASE}/api/v1/svc/instances/{inst_uid}"))
|
|
inst = d["instance"]
|
|
assert inst["status"] == "creating"
|
|
assert inst["displayName"] == "test-bolvanka"
|
|
t("GET /instances/{uid} → status=creating, имя верное", test_inst_detail)
|
|
|
|
def test_inst_fields():
|
|
d = js(requests.get(f"{BASE}/api/v1/svc/instances/{inst_uid}?fields=dtCreate,status"))
|
|
assert "instance" in d
|
|
t("GET /instances/{uid}?fields=... → 200", test_inst_fields)
|
|
|
|
def test_inst_404():
|
|
st(requests.get(f"{BASE}/api/v1/svc/instances/00000000-0000-0000-0000-000000000000"), 404)
|
|
t("GET /instances/nonexistent → 404", test_inst_404)
|
|
|
|
|
|
# ═══════════════════════════════════════
|
|
# 4. OPERATIONS
|
|
# ═══════════════════════════════════════
|
|
print("\n── 4. Operations ──")
|
|
|
|
def test_op_default():
|
|
global svc_op_id
|
|
d = js(requests.get(f"{BASE}/api/v1/svc/instanceOperations/default/18"))
|
|
assert "svcOperation" in d
|
|
assert "cfsParams" in d["svcOperation"]
|
|
assert len(d["svcOperation"]["cfsParams"]) > 0
|
|
svc_op_id = 18
|
|
t("GET /instanceOperations/default/18 → 200 + cfsParams", test_op_default)
|
|
|
|
def test_op_create():
|
|
global op_uid
|
|
r = requests.post(f"{BASE}/api/v1/svc/instanceOperations",
|
|
json={"instanceUid": inst_uid, "operation": "create",
|
|
"svcOperationId": svc_op_id}, headers=H)
|
|
st(r, 201)
|
|
assert "Location" in r.headers
|
|
d = js(r)
|
|
assert "instanceOperationUid" in d
|
|
op_uid = d["instanceOperationUid"]
|
|
t("POST /instanceOperations → 201 + Location", test_op_create)
|
|
|
|
def test_op_null_fields():
|
|
d = js(requests.get(f"{BASE}/api/v1/svc/instanceOperations/{op_uid}"))
|
|
op = d["instanceOperation"]
|
|
assert op["dtStart"] is None, f"dtStart={op['dtStart']}"
|
|
assert op["dtFinish"] is None, f"dtFinish={op['dtFinish']}"
|
|
assert op["isSuccessful"] is None, f"isSuccessful={op['isSuccessful']}"
|
|
t("GET /op/{uid} → dtStart=null, dtFinish=null, isSuccessful=null", test_op_null_fields)
|
|
|
|
def test_op_set_param():
|
|
st(requests.post(f"{BASE}/api/v1/svc/instanceOperationCfsParams",
|
|
json={"instanceOperationUid": op_uid,
|
|
"svcOperationCfsParamId": 242,
|
|
"paramValue": "test-realm"}, headers=H), 200)
|
|
t("POST /instanceOperationCfsParams → 200", test_op_set_param)
|
|
|
|
def test_op_validate():
|
|
st(requests.get(f"{BASE}/api/v1/svc/instanceOperations/{op_uid}/validate-cfs"), 200)
|
|
t("GET /validate-cfs → 200", test_op_validate)
|
|
|
|
def test_op_run():
|
|
d = js(requests.post(f"{BASE}/api/v1/svc/instanceOperations/{op_uid}/run", headers=H))
|
|
assert d["ok"] is True
|
|
t("POST /run → ok=true", test_op_run)
|
|
|
|
def test_op_double_run():
|
|
st(requests.post(f"{BASE}/api/v1/svc/instanceOperations/{op_uid}/run", headers=H), 409)
|
|
t("POST /run (повторно) → 409", test_op_double_run)
|
|
|
|
def test_op_after_run():
|
|
d = js(requests.get(f"{BASE}/api/v1/svc/instanceOperations/{op_uid}"))
|
|
op = d["instanceOperation"]
|
|
assert op["dtFinish"] is not None
|
|
assert op["isSuccessful"] is True
|
|
t("GET /op/{uid} (после run) → dtFinish!=null, isSuccessful=true", test_op_after_run)
|
|
|
|
|
|
# ═══════════════════════════════════════
|
|
# 5. FAIL-NEXT
|
|
# ═══════════════════════════════════════
|
|
print("\n── 5. Fail-next ──")
|
|
|
|
def test_fail_set():
|
|
d = js(requests.post(f"{BASE}/api/v1/svc/_mock/fail-next", headers=MH))
|
|
assert d["fail_next"] is True
|
|
t("POST /_mock/fail-next → 200, fail_next=true", test_fail_set)
|
|
|
|
def test_fail_inst():
|
|
global fail_inst_uid
|
|
r = requests.post(f"{BASE}/api/v1/svc/instances",
|
|
json={"serviceId": 1, "displayName": "fail-test"}, headers=H)
|
|
st(r, 201)
|
|
fail_inst_uid = js(r)["instanceUid"]
|
|
t("POST /instances (fail-next) → 201", test_fail_inst)
|
|
|
|
def test_fail_op():
|
|
global fail_op_uid
|
|
r = requests.post(f"{BASE}/api/v1/svc/instanceOperations",
|
|
json={"instanceUid": fail_inst_uid, "operation": "create",
|
|
"svcOperationId": 18}, headers=H)
|
|
st(r, 201)
|
|
fail_op_uid = js(r)["instanceOperationUid"]
|
|
t("POST /instanceOperations (fail-next) → 201", test_fail_op)
|
|
|
|
def test_fail_run():
|
|
d = js(requests.post(f"{BASE}/api/v1/svc/instanceOperations/{fail_op_uid}/run", headers=H))
|
|
assert d["ok"] is False
|
|
assert "error" in d
|
|
assert "mock failure" in d["error"]
|
|
t("POST /run (fail-next) → ok=false, error='mock failure'", test_fail_run)
|
|
|
|
|
|
# ═══════════════════════════════════════
|
|
# 6. AUTH
|
|
# ═══════════════════════════════════════
|
|
print("\n── 6. Auth (_mock/*) ──")
|
|
|
|
# Детектируем, включена ли auth на сервере
|
|
_auth_enabled = requests.post(f"{BASE}/api/v1/svc/_mock/reset", headers=H).status_code == 403
|
|
|
|
if _auth_enabled:
|
|
def test_auth_no_token():
|
|
st(requests.post(f"{BASE}/api/v1/svc/_mock/reset", headers=H), 403)
|
|
t("POST /_mock/reset (без токена) → 403", test_auth_no_token)
|
|
|
|
def test_auth_bad_token():
|
|
st(requests.post(f"{BASE}/api/v1/svc/_mock/reset",
|
|
headers={**H, "X-Mock-Auth": "wrong-token"}), 403)
|
|
t("POST /_mock/reset (неверный токен) → 403", test_auth_bad_token)
|
|
else:
|
|
print(" ⚠️ Auth отключена (MOCK_AUTH_TOKEN не задан) — тесты 403 пропущены")
|
|
|
|
def test_auth_good_token():
|
|
st(requests.post(f"{BASE}/api/v1/svc/_mock/reset", headers=MH), 200)
|
|
t("POST /_mock/reset (верный токен) → 200", test_auth_good_token)
|
|
|
|
def test_no_auth_services():
|
|
st(requests.get(f"{BASE}/api/v1/svc/services"), 200)
|
|
t("GET /services (без токена) → 200", test_no_auth_services)
|
|
|
|
def test_no_auth_create():
|
|
r = requests.post(f"{BASE}/api/v1/svc/instances",
|
|
json={"serviceId": 1, "displayName": "no-auth"}, headers=H)
|
|
st(r, 201)
|
|
t("POST /instances (без токена) → 201", test_no_auth_create)
|
|
|
|
|
|
# ═══════════════════════════════════════
|
|
# 7. MOCK STATE
|
|
# ═══════════════════════════════════════
|
|
print("\n── 7. Mock (state/services/delay) ──")
|
|
|
|
def test_state():
|
|
d = js(requests.get(f"{BASE}/api/v1/svc/_mock/state", headers=MH))
|
|
assert "instances" in d
|
|
assert "operations" in d
|
|
t("GET /_mock/state → instances + operations", test_state)
|
|
|
|
def test_mock_services():
|
|
d = js(requests.get(f"{BASE}/api/v1/svc/_mock/services", headers=MH))
|
|
assert d["count"] >= 37
|
|
t("GET /_mock/services → count >= 37", test_mock_services)
|
|
|
|
def test_delay_ok():
|
|
d = js(requests.post(f"{BASE}/api/v1/svc/_mock/delay/0.1", headers=MH))
|
|
assert d["delay"] == 0.1
|
|
t("POST /_mock/delay/0.1 → delay=0.1", test_delay_ok)
|
|
|
|
def test_delay_too_high():
|
|
st(requests.post(f"{BASE}/api/v1/svc/_mock/delay/100", headers=MH), 400)
|
|
t("POST /_mock/delay/100 → 400", test_delay_too_high)
|
|
|
|
def test_delay_negative():
|
|
st(requests.post(f"{BASE}/api/v1/svc/_mock/delay/-1", headers=MH), 400)
|
|
t("POST /_mock/delay/-1 → 400", test_delay_negative)
|
|
|
|
|
|
# ═══════════════════════════════════════
|
|
# 8. MULTI-STAND (v0.5.0+)
|
|
# ═══════════════════════════════════════
|
|
print("\n── 8. Multi-Stand (stand1/stand2/stand3) ──")
|
|
|
|
STANDS = ["stand1", "stand2", "stand3"]
|
|
stand_instances = {}
|
|
|
|
def test_stands_services():
|
|
for sid in STANDS:
|
|
r = requests.get(f"{BASE}/{sid}/api/v1/svc/services")
|
|
st(r, 200, f"{sid} services")
|
|
t("GET /{stand}/api/v1/svc/services → 200 для всех стендов", test_stands_services)
|
|
|
|
def test_stands_isolated():
|
|
"""Каждый стенд — свои инстансы."""
|
|
for sid in STANDS:
|
|
requests.post(f"{BASE}/{sid}/api/v1/svc/_mock/reset", headers=MH)
|
|
# Создаём по 1 инстансу на стенд
|
|
for i, sid in enumerate(STANDS):
|
|
r = requests.post(f"{BASE}/{sid}/api/v1/svc/instances",
|
|
json={"serviceId": 1, "displayName": f"inst-on-{sid}"}, headers=H)
|
|
st(r, 201)
|
|
stand_instances[sid] = js(r)["instanceUid"]
|
|
# Проверяем что на каждом стенде ровно 1 инстанс
|
|
for sid in STANDS:
|
|
d = js(requests.get(f"{BASE}/{sid}/api/v1/svc/instances"))
|
|
assert d["total"] == 1, f"{sid} has {d['total']} instances, expected 1"
|
|
t("Изоляция: каждый стенд — свои инстансы", test_stands_isolated)
|
|
|
|
def test_stands_openapi():
|
|
r = requests.get(f"{BASE}/api/v1/svc/openapi.json")
|
|
spec = js(r)
|
|
servers = [s["url"] for s in spec["servers"]]
|
|
assert len(servers) == 3, f"expected 3 servers, got {len(servers)}"
|
|
for sid in STANDS:
|
|
assert any(f"/{sid}" in u for u in servers), f"{sid} not in servers"
|
|
t("GET /openapi.json → 3 servers (по одному на стенд)", test_stands_openapi)
|
|
|
|
def test_stands_fail_next_isolated():
|
|
"""fail-next на одном стенде не влияет на другой."""
|
|
requests.post(f"{BASE}/stand1/api/v1/svc/_mock/reset", headers=MH)
|
|
requests.post(f"{BASE}/stand2/api/v1/svc/_mock/reset", headers=MH)
|
|
# fail-next только на stand1
|
|
requests.post(f"{BASE}/stand1/api/v1/svc/_mock/fail-next", headers=MH)
|
|
# Создаём операции на обоих стендах
|
|
for sid, label in [("stand1", "fail"), ("stand2", "ok")]:
|
|
r = requests.post(f"{BASE}/{sid}/api/v1/svc/instances",
|
|
json={"serviceId": 1, "displayName": label}, headers=H)
|
|
inst_uid = js(r)["instanceUid"]
|
|
r = requests.post(f"{BASE}/{sid}/api/v1/svc/instanceOperations",
|
|
json={"instanceUid": inst_uid, "operation": "create", "svcOperationId": 18}, headers=H)
|
|
op_uid = js(r)["instanceOperationUid"]
|
|
d = js(requests.post(f"{BASE}/{sid}/api/v1/svc/instanceOperations/{op_uid}/run", headers=H))
|
|
if sid == "stand1":
|
|
assert d["ok"] is False, f"stand1 should fail, got ok={d['ok']}"
|
|
else:
|
|
assert d["ok"] is True, f"stand2 should succeed, got ok={d['ok']}"
|
|
t("Изоляция fail-next: stand1 падает, stand2 работает", test_stands_fail_next_isolated)
|
|
|
|
|
|
# ═══════════════════════════════════════
|
|
# ИТОГ
|
|
# ═══════════════════════════════════════
|
|
print()
|
|
print("=" * 60)
|
|
total = passed + failed
|
|
print(f" PASS: {passed}/{total}")
|
|
if failed:
|
|
print(f" FAIL: {failed}/{total}")
|
|
for e in errors:
|
|
print(f" {e}")
|
|
else:
|
|
print(" 🎉 Все тесты пройдены!")
|
|
print("=" * 60)
|
|
|
|
sys.exit(0 if failed == 0 else 1)
|