hyh
85 lines
2.9 KiB
Python
85 lines
2.9 KiB
Python
"""真实 API 冒烟测试: 起服务 → 登录 → 建要求 → 提交 → 轮询报告"""
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
|
|
import requests
|
|
|
|
BASE = "http://127.0.0.1:8765"
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
|
|
|
|
def _wait_up(timeout: int = 40) -> bool:
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
try:
|
|
requests.get(BASE + "/api/requirements/", timeout=1)
|
|
return True
|
|
except requests.RequestException:
|
|
time.sleep(1)
|
|
return False
|
|
|
|
|
|
def main() -> int:
|
|
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "ojplatform.settings")
|
|
import django
|
|
django.setup()
|
|
server = subprocess.Popen(
|
|
[sys.executable, "manage.py", "runserver", "127.0.0.1:8765", "--noreload"],
|
|
cwd=HERE, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
|
)
|
|
try:
|
|
if not _wait_up():
|
|
print("FAIL: 服务未启动")
|
|
return 1
|
|
from django.contrib.auth.models import User
|
|
from rest_framework.authtoken.models import Token
|
|
user, _ = User.objects.get_or_create(username="smoke_user")
|
|
if not user.has_usable_password():
|
|
user.set_password("smoke12345")
|
|
user.save()
|
|
token, _ = Token.objects.get_or_create(user=user)
|
|
|
|
s = requests.Session()
|
|
s.headers["Authorization"] = f"Token {token.key}"
|
|
|
|
r = s.post(BASE + "/api/requirements/create/", json={
|
|
"title": "冒烟作业", "content": "实现 web 服务, 首页返回 ok",
|
|
}, timeout=30)
|
|
assert r.status_code == 201, r.text
|
|
req_id = r.json()["id"]
|
|
print("requirement:", req_id)
|
|
|
|
tar_path = sys.argv[1] if len(sys.argv) > 1 else "/tmp/opencode/smoke/proj.tar"
|
|
with open(tar_path, "rb") as fh:
|
|
r = s.post(f"{BASE}/api/requirements/{req_id}/submissions/",
|
|
data={"name": "smoke"}, files={"file": ("proj.tar", fh)},
|
|
timeout=600)
|
|
assert r.status_code == 201, r.text
|
|
sub_id = r.json()["id"]
|
|
|
|
deadline = time.time() + 400
|
|
while time.time() < deadline:
|
|
r = s.get(f"{BASE}/api/submissions/{sub_id}/report/", timeout=10)
|
|
if r.status_code == 200:
|
|
report = r.json()
|
|
stab = report.get("stability", {})
|
|
print("stability pass:", stab.get("pass"),
|
|
"| base_err:", stab.get("base_error_rate"),
|
|
"| ramp:", stab.get("rampup_crash_rate"),
|
|
"| reqs:", stab.get("total_requests"))
|
|
overall = report.get("overall", {})
|
|
print("overall:", overall.get("passed_checks"), "/", overall.get("decided_checks"))
|
|
print("SMOKE OK")
|
|
return 0
|
|
time.sleep(3)
|
|
print("TIMEOUT")
|
|
return 1
|
|
finally:
|
|
server.terminate()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|