138 lines
6.1 KiB
Python
138 lines
6.1 KiB
Python
"""仅在 Docker 本地测试容器中执行的真实 HTTP 冒烟检查。"""
|
|
import os
|
|
import sys
|
|
import time
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
import django
|
|
import requests
|
|
from bs4 import BeautifulSoup
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
if os.environ.get("DJANGO_SETTINGS_MODULE") != "wagtailcms.settings.docker_local":
|
|
raise SystemExit("拒绝执行：此脚本只允许用于 docker_local 隔离环境。")
|
|
|
|
django.setup()
|
|
|
|
from wagtail.models import Page
|
|
from apps.core.models import ContactPage, SiteSettings
|
|
from apps.forms.models import Lead
|
|
|
|
BACKEND = "http://localhost:8000"
|
|
FRONTEND = "http://frontend:3000"
|
|
COMPANY = "Docker 联调测试公司"
|
|
|
|
|
|
def require(condition, message):
|
|
if not condition:
|
|
raise AssertionError(message)
|
|
|
|
|
|
def csrf(response):
|
|
response.raise_for_status()
|
|
field = BeautifulSoup(response.text, "html.parser").find("input", {"name": "csrfmiddlewaretoken"})
|
|
require(field is not None, "响应没有 CSRF 表单令牌")
|
|
return field["value"]
|
|
|
|
|
|
def main():
|
|
session = requests.Session()
|
|
login_url = f"{BACKEND}/admin/login/"
|
|
token = csrf(session.get(login_url, timeout=30))
|
|
response = session.post(login_url, data={
|
|
"username": os.environ["DJANGO_SUPERUSER_USERNAME"],
|
|
"password": os.environ["DJANGO_SUPERUSER_PASSWORD"],
|
|
"csrfmiddlewaretoken": token,
|
|
}, headers={"Referer": login_url}, timeout=30)
|
|
response.raise_for_status()
|
|
require("sessionid" in session.cookies, "后台登录失败；已有管理员密码不会自动重置")
|
|
|
|
setup_url = f"{BACKEND}/admin/company-setup/"
|
|
payload = {
|
|
"company_name": COMPANY,
|
|
"contact_phone": "010-12345678",
|
|
"contact_email": "leads@example.test",
|
|
"address": "仅用于 Docker 本地联调",
|
|
"icp_number": "",
|
|
"create_demo": "on",
|
|
}
|
|
for attempt in range(2):
|
|
token = csrf(session.get(setup_url, timeout=30))
|
|
response = session.post(setup_url, data={**payload, "csrfmiddlewaretoken": token},
|
|
headers={"Referer": setup_url}, timeout=120)
|
|
response.raise_for_status()
|
|
require("企业资料已保存" in response.text, "开站向导保存失败")
|
|
require(SiteSettings.objects.filter(company_name=COMPANY).exists(), "企业资料未持久化")
|
|
if attempt == 0:
|
|
count = Page.objects.count()
|
|
page_state = list(Page.objects.order_by("pk").values_list(
|
|
"pk", "title", "live", "has_unpublished_changes", "latest_revision_id",
|
|
))
|
|
else:
|
|
require(Page.objects.count() == count, "重复开站产生重复页面")
|
|
require(list(Page.objects.order_by("pk").values_list(
|
|
"pk", "title", "live", "has_unpublished_changes", "latest_revision_id",
|
|
)) == page_state, "重复开站改变已有页面状态")
|
|
print("PASS: HTTP 登录、CSRF 开站表单及重复执行")
|
|
|
|
response = requests.post(f"{FRONTEND}/api/revalidate", json={"tags": ["wagtail-pages"]}, timeout=30)
|
|
require(response.status_code == 401, "缓存刷新接口未拒绝匿名请求")
|
|
response = requests.post(f"{FRONTEND}/api/revalidate", json={"tags": ["wagtail-pages", "site-settings"]},
|
|
headers={"Authorization": f"Bearer {os.environ['REVALIDATE_SECRET']}"}, timeout=30)
|
|
require(response.status_code == 200, "授权缓存刷新失败")
|
|
print("PASS: 前端缓存刷新鉴权和内部网络连接")
|
|
|
|
for path, text in [
|
|
("/", COMPANY), ("/products", "企业内容平台"),
|
|
("/products/content-platform", "企业内容平台"),
|
|
("/solutions/digital-workplace", "企业数字化协同"),
|
|
("/cases/demo-content-project", "虚构"),
|
|
("/blog/website-content-checklist", "上线内容清单"),
|
|
("/about", COMPANY), ("/contact", "需求描述"),
|
|
]:
|
|
for attempt in range(24):
|
|
response = requests.get(f"{FRONTEND}{path}", timeout=30)
|
|
if response.status_code == 200 and text in response.text:
|
|
break
|
|
time.sleep(5)
|
|
require(response.status_code == 200 and text in response.text, f"前端页面未显示真实内容：{path}")
|
|
print(f"PASS: 前端 {path}")
|
|
|
|
contact = ContactPage.objects.get(slug="contact")
|
|
form = next(block.value["form"] for block in contact.body if block.block_type == "form")
|
|
form.notification_email = "leads@example.test"
|
|
form.save(update_fields=["notification_email"])
|
|
marker = f"docker-smoke-{uuid.uuid4().hex}"
|
|
cors = requests.options(f"{BACKEND}/api/v1/custom/leads/", headers={
|
|
"Origin": "http://localhost:13000", "Access-Control-Request-Method": "POST",
|
|
"Access-Control-Request-Headers": "content-type",
|
|
}, timeout=30)
|
|
require(cors.headers.get("Access-Control-Allow-Origin") == "http://localhost:13000", "浏览器跨域预检未通过")
|
|
response = requests.post(f"{BACKEND}/api/v1/custom/leads/", json={
|
|
"form_id": form.pk, "data": {"name": "Docker 测试", "company": COMPANY,
|
|
"phone": "010-12345678", "email": "visitor@example.test", "message": marker},
|
|
"source_url": "http://localhost:13000/contact", "website": "",
|
|
}, headers={"Origin": "http://localhost:13000"}, timeout=30)
|
|
require(response.status_code == 201 and response.json().get("success"), "咨询提交失败")
|
|
require(Lead.objects.filter(form=form, data__message=marker).exists(), "咨询未入库")
|
|
found = False
|
|
for attempt in range(15):
|
|
messages = requests.get("http://mailpit:8025/api/v1/messages", timeout=10).json().get("messages", [])
|
|
for message in messages:
|
|
detail = requests.get(f"http://mailpit:8025/api/v1/message/{message['ID']}", timeout=10).json()
|
|
if marker in detail.get("Text", ""):
|
|
found = True
|
|
break
|
|
if found:
|
|
break
|
|
time.sleep(2)
|
|
require(found, "通知邮件未被 Mailpit 捕获")
|
|
print("PASS: 跨域预检、真实表单 API、PostgreSQL 线索入库及 SMTP 邮件捕获")
|
|
print("ALL LOCAL HTTP SMOKE CHECKS PASSED")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|