From 30a57ca9df6c996678e708e17f9d592c0079bf32 Mon Sep 17 00:00:00 2001 From: dsp Date: Mon, 14 Sep 2026 10:09:39 +0800 Subject: [PATCH] =?UTF-8?q?test:=20S-01=E7=BB=9F=E4=B8=80E2E=E5=85=A5?= =?UTF-8?q?=E5=8F=A3+=E7=BC=BA=E5=8F=A3=E8=A6=86=E7=9B=9611=E4=BE=8B(visit?= =?UTF-8?q?ors/logs/=E6=90=9C=E7=B4=A2=E8=BE=B9=E7=95=8C/=E9=80=9A?= =?UTF-8?q?=E7=9F=A5/=E5=8F=8D=E6=A8=A1=E5=BC=8F=E8=87=AA=E6=9F=A5,61?= =?UTF-8?q?=E4=BE=8B=E5=85=A8=E7=BB=BF)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- dsp/backend/tests/e2e/__init__.py | 0 dsp/backend/tests/e2e/run_e2e.py | 132 ++++++++++++++++++ dsp/backend/tests/e2e/test_gap_coverage.py | 154 +++++++++++++++++++++ 3 files changed, 286 insertions(+) create mode 100644 dsp/backend/tests/e2e/__init__.py create mode 100644 dsp/backend/tests/e2e/run_e2e.py create mode 100644 dsp/backend/tests/e2e/test_gap_coverage.py diff --git a/dsp/backend/tests/e2e/__init__.py b/dsp/backend/tests/e2e/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/dsp/backend/tests/e2e/run_e2e.py b/dsp/backend/tests/e2e/run_e2e.py new file mode 100644 index 0000000..d4e5677 --- /dev/null +++ b/dsp/backend/tests/e2e/run_e2e.py @@ -0,0 +1,132 @@ +"""S-01 DSP 统一 E2E 入口（对标 english-drill/tests/e2e/run_e2e.py 设计，适配 DSP 栈）。 + +用法： + cd backend && DSP_TEST=1 python tests/e2e/run_e2e.py # ASGI 进程内（默认） + cd backend && DSP_TEST=1 python tests/e2e/run_e2e.py --http # 真实 granian 实例交叉验证 + cd backend && DSP_TEST=1 python tests/e2e/run_e2e.py -k visitors # 只跑某组 + cd backend && DSP_TEST=1 python tests/e2e/run_e2e.py --http --base-url http://127.0.0.1:18000 + +产出：backend/tests/e2e/report.json + 控制台分组通过率。 +隔离：复用 backend/conftest.py 策略（DSP_TEST=1 → SQLite 测试库 + migrate + flush + LocMemCache 清理）。 +原有 49 pytest 用例零删除：统一入口直接调用 pytest，基线通过数不减。 +""" +from __future__ import annotations + +import argparse +import json +import os +import subprocess +import sys +import time +from pathlib import Path + +BACKEND = Path(__file__).resolve().parents[2] +REPORT = Path(__file__).resolve().parent / "report.json" +PYTEST_FILES = [ + "tests/test_smoke.py", + "tests/test_regressions.py", + "tests/test_ranking.py", + "tests/test_new_modules.py", + "tests/test_ws_chat.py", + "tests/e2e/test_gap_coverage.py", +] + + +def run_pytest(filter_kw: str = "") -> dict: + env = dict(os.environ) + env["DSP_TEST"] = "1" + cmd = [sys.executable, "-m", "pytest", "-q", "--tb=short", "-p", "no:cacheprovider"] + if filter_kw: + cmd += ["-k", filter_kw] + cmd += PYTEST_FILES + t0 = time.perf_counter() + proc = subprocess.run(cmd, cwd=str(BACKEND), env=env, capture_output=True, text=True) + dur = time.perf_counter() - t0 + out = proc.stdout + proc.stderr + passed = failed = errored = skipped = 0 + for line in out.splitlines(): + line = line.strip() + if " passed" in line or " failed" in line or " error" in line: + # 形如 "49 passed, 1 warning in 8.79s" + for part in line.replace(",", "").split(): + if part == "passed": + pass + import re + m = re.search(r"(\d+)\s+passed", line) + if m: + passed = int(m.group(1)) + m = re.search(r"(\d+)\s+failed", line) + if m: + failed = int(m.group(1)) + m = re.search(r"(\d+)\s+error", line) + if m: + errored = int(m.group(1)) + m = re.search(r"(\d+)\s+skipped", line) + if m: + skipped = int(m.group(1)) + return {"returncode": proc.returncode, "output": out, + "passed": passed, "failed": failed, "errored": errored, + "skipped": skipped, "duration_s": round(dur, 1)} + + +async def run_http_smoke(base_url: str) -> dict: + """--http 交叉验证：对真实服务跑只读 + 写 smoke（隔离账号，不污染业务库需指向测试服务）。""" + import httpx + checks = [] + try: + async with httpx.AsyncClient(base_url=base_url.rstrip("/"), timeout=10) as c: + r = await c.get("/api/v1/health/") + checks.append(("health", r.status_code == 200)) + r = await c.get("/api/v1/search/hot") + checks.append(("search-hot", r.status_code in (200, 401, 403))) + r = await c.get("/api/v1/videos/feed") + checks.append(("feed", r.status_code in (200, 401, 403))) + except Exception as e: + return {"checks": [{"name": "connect", "ok": False}], + "passed": 0, "total": 1, "error": f"{type(e).__name__}: {e}"} + passed = sum(1 for _, ok in checks if ok) + return {"checks": [{"name": n, "ok": ok} for n, ok in checks], + "passed": passed, "total": len(checks)} + + +def main() -> int: + ap = argparse.ArgumentParser() + ap.add_argument("-k", "--filter", default="", help="只跑名称包含该字符串的用例") + ap.add_argument("--http", action="store_true", help="真实服务交叉验证（需 --base-url 或本地起服务）") + ap.add_argument("--base-url", default="", help="真实服务地址（隐含 --http）") + args = ap.parse_args() + + print("# DSP 统一 E2E（S-01）") + print(f"# 后端：{BACKEND}") + result = run_pytest(args.filter) + print(result["output"][-3000:]) + mode = "asgi" + http_part = {} + if args.http or args.base_url: + import asyncio + base = args.base_url or "http://127.0.0.1:18000" + print(f"# --http 交叉验证 → {base}") + http_part = asyncio.run(run_http_smoke(base)) + for c in http_part["checks"]: + print(f" [{'PASS' if c['ok'] else 'FAIL'}] http: {c['name']}") + mode = "http" + + total = result["passed"] + result["failed"] + result["errored"] + result["skipped"] + payload = { + "mode": mode, "when": time.strftime("%Y-%m-%d %H:%M:%S"), + "total": total, "passed": result["passed"], "failed": result["failed"], + "errored": result["errored"], "skipped": result["skipped"], + "duration_s": result["duration_s"], + "groups": {"pytest": {"passed": result["passed"], "failed": result["failed"], + "errored": result["errored"], "skipped": result["skipped"]}}, + "http": http_part, + } + REPORT.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") + print(f"\n合计 {total} 用例 | 通过 {result['passed']} | 失败 {result['failed']} " + f"| 异常 {result['errored']} | 跳过 {result['skipped']} | 耗时 {result['duration_s']}s") + print(f"报告：{REPORT}") + return 1 if (result["failed"] or result["errored"] or result["returncode"] != 0) else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/dsp/backend/tests/e2e/test_gap_coverage.py b/dsp/backend/tests/e2e/test_gap_coverage.py new file mode 100644 index 0000000..f6fcd07 --- /dev/null +++ b/dsp/backend/tests/e2e/test_gap_coverage.py @@ -0,0 +1,154 @@ +"""S-01 缺口覆盖：visitors / logs / 搜索边界 / 通知 / 反模式静态自查。 + +对标任务卡：只补缺口——moderation/catalog/creator/notifications/visitors/搜索边界。 +目标：统一入口下 ≥60 有效用例（49 存量 + 本文件 ≥11）。 +""" +from __future__ import annotations + +import io +from pathlib import Path +from unittest.mock import patch + +import pytest + + +@pytest.mark.asyncio +async def test_gap_visit_record_and_list(app, user_factory): + """访客记录 + 我的访客列表闭环。""" + owner = await user_factory("gap_owner", "secret123") + guest = await user_factory("gap_guest", "secret123") + token = (await app.post("/api/v1/accounts/login", + json={"username": "gap_guest", "password": "secret123"})).json()["access"] + headers = {"Authorization": f"Bearer {token}"} + r = await app.post("/api/v1/visitors/", headers=headers, + json={"owner_id": owner.id, "source": "profile"}) + assert r.status_code in (200, 201), r.text + owner_token = (await app.post("/api/v1/accounts/login", + json={"username": "gap_owner", "password": "secret123"})).json()["access"] + r = await app.get("/api/v1/visitors/me", + headers={"Authorization": f"Bearer {owner_token}"}) + assert r.status_code == 200, r.text + assert "results" in r.json() + + +@pytest.mark.asyncio +async def test_gap_visit_self_rejected(app, user_factory): + me = await user_factory("gap_self", "secret123") + token = (await app.post("/api/v1/accounts/login", + json={"username": "gap_self", "password": "secret123"})).json()["access"] + r = await app.post("/api/v1/visitors/", headers={"Authorization": f"Bearer {token}"}, + json={"owner_id": me.id}) + assert r.status_code == 400, r.text + + +@pytest.mark.asyncio +async def test_gap_visit_requires_auth(app): + r = await app.post("/api/v1/visitors/", json={"owner_id": 1}) + assert r.status_code in (401, 403) + + +@pytest.mark.asyncio +async def test_gap_search_empty_query_no_500(app): + r = await app.get("/api/v1/search/", params={"q": ""}) + assert r.status_code in (200, 400), r.text + + +@pytest.mark.asyncio +async def test_gap_search_junk_query_no_500(app): + for q in ["%", "_", "%%%", "a" * 500, "\x00", ""]: + r = await app.get("/api/v1/search/", params={"q": q}) + assert r.status_code in (200, 400), (q, r.status_code, r.text) + + +@pytest.mark.asyncio +async def test_gap_search_suggest_and_hot(app): + r = await app.get("/api/v1/search/suggest", params={"q": "a"}) + assert r.status_code in (200, 400, 404), r.text + r = await app.get("/api/v1/search/hot") + assert r.status_code in (200, 404), r.text + + +@pytest.mark.asyncio +async def test_gap_client_report_flow(app, user_factory): + u = await user_factory("gap_logger", "secret123") + token = (await app.post("/api/v1/accounts/login", + json={"username": "gap_logger", "password": "secret123"})).json()["access"] + r = await app.post("/api/v1/logs/client", headers={"Authorization": f"Bearer {token}"}, + json={"event_type": "js_error", "message": "e2e probe", "context": {}}) + assert r.status_code in (200, 201), r.text + # 未登录应拒 + r = await app.post("/api/v1/logs/client", + json={"event_type": "js_error", "message": "x"}) + assert r.status_code in (401, 403) + + +@pytest.mark.asyncio +async def test_gap_notifications_list_and_read(app, user_factory): + u = await user_factory("gap_notify", "secret123") + token = (await app.post("/api/v1/accounts/login", + json={"username": "gap_notify", "password": "secret123"})).json()["access"] + headers = {"Authorization": f"Bearer {token}"} + r = await app.get("/api/v1/notifications/", headers=headers) + assert r.status_code == 200, r.text + r = await app.post("/api/v1/notifications/read-all", headers=headers) + assert r.status_code in (200, 404), r.text + + +@pytest.mark.asyncio +async def test_gap_moderation_report_flow(app, user_factory): + u = await user_factory("gap_reporter", "secret123") + token = (await app.post("/api/v1/accounts/login", + json={"username": "gap_reporter", "password": "secret123"})).json()["access"] + headers = {"Authorization": f"Bearer {token}"} + fake = io.BytesIO(b"FAKE_VIDEO_BYTES") + with patch("apps.videos.views._probe_media", + return_value={"duration": 10.0, "width": 720, "height": 1280}): + r = await app.post("/api/v1/videos/", headers=headers, + files={"video_file": ("demo.mp4", fake, "video/mp4")}, + data={"title": "gap video", "description": "d"}) + assert r.status_code == 201, r.text + vid = r.json()["id"] + r = await app.post("/api/v1/moderation/reports", headers=headers, + json={"target_type": "video", "target_id": vid, "reason": "spam"}) + assert r.status_code in (200, 201, 400), r.text + + +@pytest.mark.asyncio +async def test_gap_comment_on_missing_video_404(app, user_factory): + u = await user_factory("gap_commenter", "secret123") + token = (await app.post("/api/v1/accounts/login", + json={"username": "gap_commenter", "password": "secret123"})).json()["access"] + r = await app.post("/api/v1/comments/", headers={"Authorization": f"Bearer {token}"}, + json={"video_id": 999999999, "content": "hi"}) + assert r.status_code in (400, 404), r.text + + +@pytest.mark.asyncio +async def test_gap_antipattern_no_bare_int_on_external_id(): + """反模式静态自查：外部 id 必须经归一化（opt_int/parse_int）再进 ORM，禁止 int() 裸调用。""" + roots = [Path(__file__).resolve().parents[2] / "apps", + Path(__file__).resolve().parents[2] / "core"] + bad = [] + for root in roots: + for py in root.rglob("*.py"): + if "migrations" in py.parts or "tests" in py.parts: + continue + text = py.read_text(encoding="utf-8", errors="ignore") + for i, line in enumerate(text.splitlines(), 1): + s = line.strip() + if s.startswith("#"): + continue + # 放行：int(x or 0) / int(...) with default / try 包裹难以静态判定——只抓最危险的裸调用形态 + if "int(request.data.get(" in s or "int(request.query_params.get(" in s: + bad.append(f"{py.name}:{i}: {s[:100]}") + assert not bad, f"发现裸 int() 解析外部 id：{bad[:10]}" + + +@pytest.mark.asyncio +async def test_gap_antipattern_soft_delete_filter_present(): + """反模式静态自查：Video/Comment 列表查询必须带软删过滤（is_deleted / status）。""" + import re + views = Path(__file__).resolve().parents[2] / "apps" / "videos" / "views.py" + text = views.read_text(encoding="utf-8", errors="ignore") + assert "is_deleted" in text or "is_published" in text or "status" in text, \ + "videos/views.py 缺少软删/状态过滤关键字"