OJplatform/requirements/tests.py

"""端到端测试: 创建作业要求 → 上传项目包 → 检测 → 报告"""
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()
        llm_json = (
            '{"ai_inspection": true, '
            '"stability": {"enabled": true, "requirements": "基准并发20错误率≤1%"}, '
            '"functional": {"enabled": true, "points": ["首页可访问", "返回 ok"]}, '
            '"performance": {"enabled": false, "requirements": ""}, '
            '"security": {"enabled": false, "requirements": ""}}'
        )
        with mock.patch("requirements.llm.resolve_llm", return_value=("http://llm.test/v1", "k", "m")), \
             mock.patch("requirements.llm._chat", return_value=llm_json):
            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_create_requirement_custom_llm(self):
        self._login()
        override = {"mode": "custom", "base_url": "http://custom.test/v1",
                    "api_key": "ck", "model": "custom-model"}
        llm_json = '{"ai_inspection": true, "functional": {"enabled": true, "points": ["自定义功能点"]}}'
        captured = {}
        with mock.patch("requirements.llm.resolve_llm", side_effect=lambda ov: captured.update(ov) or ("http://custom.test/v1", "ck", "custom-model")), \
             mock.patch("requirements.llm._chat", return_value=llm_json):
            resp = self.client.post("/api/requirements/create/", {
                "title": "custom-llm-作业",
                "content": "任意",
                "llm": override,
            }, format="json")
        self.assertEqual(resp.status_code, 201, resp.content)
        self.assertEqual(captured["mode"], "custom")
        requirement = Requirement.objects.get(title="custom-llm-作业")
        self.assertEqual(requirement.llm_config["mode"], "custom")
        self.assertIn("自定义功能点", requirement.spec["functional"]["points"])

    def test_custom_llm_requires_fields(self):
        self._login()
        resp = self.client.post("/api/requirements/create/", {
            "title": "缺字段", "content": "任意",
            "llm": {"mode": "custom", "base_url": "", "model": ""},
        }, format="json")
        self.assertEqual(resp.status_code, 400)

    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())