hyh
119 lines
4.8 KiB
Python
119 lines
4.8 KiB
Python
"""端到端测试: 创建作业要求 → 上传项目包 → 检测 → 报告"""
|
|
import io
|
|
import tarfile
|
|
import textwrap
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
from django.contrib.auth.models import User
|
|
from rest_framework.test import APITestCase
|
|
|
|
from requirements.models import Requirement, Submission
|
|
|
|
FLASK_PROJECT = textwrap.dedent("""\
|
|
from flask import Flask
|
|
app = Flask(__name__)
|
|
|
|
@app.route("/")
|
|
def index():
|
|
return "ok"
|
|
|
|
if __name__ == "__main__":
|
|
app.run(host="0.0.0.0", port=8321)
|
|
""")
|
|
|
|
|
|
def make_tar_bytes(name: str, files: dict) -> bytes:
|
|
buf = io.BytesIO()
|
|
with tarfile.open(fileobj=buf, mode="w") as tf:
|
|
for fname, content in files.items():
|
|
data = content.encode()
|
|
info = tarfile.TarInfo(name=f"{name}/{fname}")
|
|
info.size = len(data)
|
|
tf.addfile(info, io.BytesIO(data))
|
|
return buf.getvalue()
|
|
|
|
|
|
class EndToEndTests(APITestCase):
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
cls.user = User.objects.create_user("tester", password="pass12345")
|
|
|
|
def setUp(self):
|
|
self._login()
|
|
|
|
def _login(self):
|
|
from rest_framework.authtoken.models import Token
|
|
token, _ = Token.objects.get_or_create(user=self.user)
|
|
self.client.credentials(HTTP_AUTHORIZATION=f"Token {token.key}")
|
|
|
|
def test_create_requirement_analyzes_spec(self):
|
|
self._login()
|
|
with mock.patch("requirements.llm.requests.post") as post:
|
|
post.return_value = mock.Mock(
|
|
json=lambda: {"choices": [{"message": {"content": (
|
|
'{"ai_inspection": true, '
|
|
'"stability": {"enabled": true, "requirements": "基准并发20错误率≤1%"}, '
|
|
'"functional": {"enabled": true, "points": ["首页可访问", "返回 ok"]}, '
|
|
'"performance": {"enabled": false, "requirements": ""}, '
|
|
'"security": {"enabled": false, "requirements": ""}}'
|
|
)}}]},
|
|
raise_for_status=lambda: None,
|
|
)
|
|
resp = self.client.post("/api/requirements/create/", {
|
|
"title": "作业1-分析",
|
|
"content": "实现一个 web 服务, 首页返回 ok, 要求稳定度达到 99%",
|
|
}, format="json")
|
|
self.assertEqual(resp.status_code, 201, resp.content)
|
|
requirement = Requirement.objects.get(title="作业1-分析")
|
|
self.assertTrue(requirement.spec["stability"]["enabled"])
|
|
self.assertIn("首页可访问", requirement.spec["functional"]["points"])
|
|
|
|
def test_submission_full_pipeline(self):
|
|
self._login()
|
|
requirement = Requirement.objects.create(
|
|
title="作业1",
|
|
content="实现 web 服务",
|
|
spec={
|
|
"ai_inspection": True,
|
|
"stability": {"enabled": True, "requirements": "基准错误率≤1%"},
|
|
"functional": {"enabled": True, "points": ["首页可访问"]},
|
|
"performance": {"enabled": False, "requirements": ""},
|
|
"security": {"enabled": False, "requirements": ""},
|
|
},
|
|
)
|
|
tar_bytes = make_tar_bytes("proj", {"app.py": FLASK_PROJECT})
|
|
with mock.patch("requirements.llm.ai_summarize", return_value={
|
|
"summary": "代码结构清晰, 功能点完备",
|
|
"issues": [{"point": "首页可访问", "status": "pass", "detail": "/ 路由存在"}],
|
|
}), mock.patch.dict("django.conf.settings.__dict__", {
|
|
"STABILITY_BASE_DURATION": 2, "STABILITY_RAMPUP_DURATION": 2, "STABILITY_CONCURRENCY": 4,
|
|
}):
|
|
resp = self.client.post(
|
|
f"/api/requirements/{requirement.pk}/submissions/",
|
|
{"name": "tester-sub", "file": io.BytesIO(tar_bytes)},
|
|
format="multipart",
|
|
)
|
|
self.assertEqual(resp.status_code, 201, resp.content)
|
|
submission = Submission.objects.get()
|
|
if submission.status != "completed":
|
|
submission.report["error"] = submission.report.get("error", "?")
|
|
raise AssertionError(f"status={submission.status} report={submission.report}")
|
|
self.assertIn("ai_inspection", submission.report)
|
|
self.assertIn("stability", submission.report)
|
|
report = self.client.get(f"/api/submissions/{submission.pk}/report/")
|
|
self.assertEqual(report.status_code, 200)
|
|
self.assertIn("stability", report.json())
|
|
|
|
|
|
class AuthTests(APITestCase):
|
|
def test_requires_login(self):
|
|
resp = self.client.get("/api/requirements/")
|
|
self.assertIn(resp.status_code, (401, 403))
|
|
|
|
def test_token_login(self):
|
|
User.objects.create_user("alice", password="pass12345")
|
|
resp = self.client.post("/api/auth/token/", {"username": "alice", "password": "pass12345"})
|
|
self.assertEqual(resp.status_code, 200)
|
|
self.assertIn("token", resp.json())
|