test: S-01统一E2E入口+缺口覆盖11例(visitors/logs/搜索边界/通知/反模式自查,61例全绿)
This commit is contained in:
dsp
@@ -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())
|
||||
@@ -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", "<script>alert(1)</script>"]:
|
||||
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 缺少软删/状态过滤关键字"
|
||||
Reference in New Issue
Block a user