补充测试覆盖率 + 新增部署引导命令 bootstrap_deployment
- 新增 apps/blog、apps/solutions、apps/home、apps/api 测试文件，覆盖标签/排序/ 草稿过滤、HomePage.subpage_types 回归测试、page_preview 预览接口 - apps/core/tests.py 新增健康检查端点、发布 webhook 信号、bootstrap_deployment 命令的测试用例，pytest 用例数从 40 增至 65，全部通过 - 修复 apps/api/urls.py 中 PagePreviewAPIViewSet.get_object() 未捕获 DoesNotExist 导致 500 而非文档承诺 404 的 bug - 新增 apps/core/management/commands/bootstrap_deployment.py：部署时按环境变量 幂等创建首个超级管理员账号并联动初始化 RBAC 权限组，.env.example 补充对应 环境变量说明 - 新增 OpenSearch + 中文分词本地 PoC 脚手架（docker-compose.opensearch.yml + docker/opensearch/Dockerfile），调研结论详见设计文档 §2.10 - 更新设计文档 §2.16 Phase 1 进度清单
This commit is contained in:
@@ -0,0 +1,47 @@
|
||||
"""apps.api 单元测试：预览专用接口 `page_preview`（apps/api/urls.py 的
|
||||
PagePreviewAPIViewSet），覆盖 §2.7/§2.15 描述的"仅凭有效 content_type+token
|
||||
才能读取草稿快照"行为。"""
|
||||
import pytest
|
||||
from rest_framework.test import APIClient
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def preview_token(home_page):
|
||||
"""为 home_page 生成一条预览快照记录，返回可用于查询接口的 token。"""
|
||||
preview = home_page.create_page_preview()
|
||||
preview.save()
|
||||
return preview.token
|
||||
|
||||
|
||||
def test_page_preview_returns_draft_snapshot_for_valid_token(home_page, preview_token):
|
||||
client = APIClient()
|
||||
response = client.get(
|
||||
"/api/v2/page_preview/",
|
||||
{"content_type": "home.homepage", "token": preview_token},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.data["title"] == home_page.title
|
||||
|
||||
|
||||
def test_page_preview_returns_404_for_unknown_token(home_page):
|
||||
client = APIClient()
|
||||
response = client.get(
|
||||
"/api/v2/page_preview/",
|
||||
{"content_type": "home.homepage", "token": "does-not-exist"},
|
||||
)
|
||||
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
def test_page_preview_returns_404_for_mismatched_content_type(home_page, preview_token):
|
||||
"""token 是为 home.homepage 生成的，用错误的 content_type 查询应查不到记录。"""
|
||||
client = APIClient()
|
||||
response = client.get(
|
||||
"/api/v2/page_preview/",
|
||||
{"content_type": "blog.blogpage", "token": preview_token},
|
||||
)
|
||||
|
||||
assert response.status_code == 404
|
||||
+11
-5
@@ -3,6 +3,7 @@ Wagtail API v2 路由配置。
|
||||
详见 documents/设计方案分析与完善版.md §2.7 Headless API 设计规范。
|
||||
"""
|
||||
from django.contrib.contenttypes.models import ContentType
|
||||
from django.http import Http404
|
||||
from rest_framework.response import Response
|
||||
from wagtail.api.v2.views import PagesAPIViewSet
|
||||
from wagtail.api.v2.router import WagtailAPIRouter
|
||||
@@ -40,11 +41,16 @@ class PagePreviewAPIViewSet(PagesAPIViewSet):
|
||||
return Response(serializer.data)
|
||||
|
||||
def get_object(self):
|
||||
app_label, model = self.request.GET["content_type"].split(".")
|
||||
content_type = ContentType.objects.get(app_label=app_label, model=model)
|
||||
page_preview = PagePreview.objects.get(
|
||||
content_type=content_type, token=self.request.GET["token"]
|
||||
)
|
||||
try:
|
||||
app_label, model = self.request.GET["content_type"].split(".")
|
||||
content_type = ContentType.objects.get(app_label=app_label, model=model)
|
||||
page_preview = PagePreview.objects.get(
|
||||
content_type=content_type, token=self.request.GET["token"]
|
||||
)
|
||||
except (KeyError, ValueError, ContentType.DoesNotExist, PagePreview.DoesNotExist):
|
||||
# 缺少参数/content_type 格式错误/未知或已过期的 token，统一返回 404，
|
||||
# 不暴露内部异常细节（详见类文档字符串）。
|
||||
raise Http404("No matching page preview found.")
|
||||
page = page_preview.as_page()
|
||||
if not page.pk:
|
||||
# 新建（尚未保存）页面的预览没有真实主键，填充占位值避免 API 路由生成 URL 时报错
|
||||
|
||||
@@ -0,0 +1,85 @@
|
||||
"""apps.blog 单元测试：页面树结构、标签与列表排序。"""
|
||||
import pytest
|
||||
from django.utils import timezone
|
||||
|
||||
from apps.blog.models import BlogIndexPage, BlogPage
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def blog_index(home_page):
|
||||
index = BlogIndexPage(title="博客", slug="blog", intro="技术与行业洞察")
|
||||
home_page.add_child(instance=index)
|
||||
return index
|
||||
|
||||
|
||||
def test_blog_page_can_be_created_under_index(blog_index):
|
||||
post = BlogPage(
|
||||
title="文章一",
|
||||
slug="post-1",
|
||||
published_at=timezone.now(),
|
||||
intro="摘要一",
|
||||
)
|
||||
blog_index.add_child(instance=post)
|
||||
|
||||
assert BlogPage.objects.live().descendant_of(blog_index).count() == 1
|
||||
|
||||
|
||||
def test_blog_page_supports_tags(blog_index):
|
||||
post = BlogPage(
|
||||
title="文章一",
|
||||
slug="post-1",
|
||||
published_at=timezone.now(),
|
||||
)
|
||||
blog_index.add_child(instance=post)
|
||||
|
||||
post.tags.add("Django", "Wagtail")
|
||||
post.save()
|
||||
|
||||
post.refresh_from_db()
|
||||
assert {t.name for t in post.tags.all()} == {"Django", "Wagtail"}
|
||||
|
||||
|
||||
def test_blog_index_context_orders_by_published_at_desc(blog_index):
|
||||
older = BlogPage(
|
||||
title="较早的文章",
|
||||
slug="older-post",
|
||||
published_at=timezone.now() - timezone.timedelta(days=5),
|
||||
)
|
||||
blog_index.add_child(instance=older)
|
||||
|
||||
newer = BlogPage(
|
||||
title="较新的文章",
|
||||
slug="newer-post",
|
||||
published_at=timezone.now(),
|
||||
)
|
||||
blog_index.add_child(instance=newer)
|
||||
|
||||
context = blog_index.get_context(request=None)
|
||||
titles = [p.title for p in context["posts"]]
|
||||
|
||||
assert titles == ["较新的文章", "较早的文章"]
|
||||
|
||||
|
||||
def test_blog_index_context_excludes_draft_posts(blog_index):
|
||||
live_post = BlogPage(
|
||||
title="已发布文章",
|
||||
slug="live-post",
|
||||
published_at=timezone.now(),
|
||||
)
|
||||
blog_index.add_child(instance=live_post)
|
||||
|
||||
draft_post = BlogPage(
|
||||
title="草稿文章",
|
||||
slug="draft-post",
|
||||
published_at=timezone.now(),
|
||||
live=False,
|
||||
)
|
||||
blog_index.add_child(instance=draft_post)
|
||||
|
||||
context = blog_index.get_context(request=None)
|
||||
titles = [p.title for p in context["posts"]]
|
||||
|
||||
assert "已发布文章" in titles
|
||||
assert "草稿文章" not in titles
|
||||
@@ -0,0 +1,56 @@
|
||||
"""
|
||||
部署引导命令：首次上线（或每次部署）时自动完成"能登录后台 + 拥有正确权限组"这两件事，
|
||||
避免每次部署都要人工登录服务器手动执行 `createsuperuser`。
|
||||
|
||||
用法：
|
||||
python manage.py bootstrap_deployment
|
||||
|
||||
行为（幂等，可安全重复执行，适合放进部署脚本/容器启动脚本的 migrate 之后）：
|
||||
1. 若环境变量 DJANGO_SUPERUSER_USERNAME / DJANGO_SUPERUSER_EMAIL /
|
||||
DJANGO_SUPERUSER_PASSWORD 均已设置，且该用户名尚不存在，则创建一个超级管理员账号。
|
||||
- 若用户名已存在，跳过创建（不会重置密码，避免每次部署都覆盖已被人工修改过的密码）。
|
||||
- 若三个环境变量未完整设置，跳过此步骤并给出提示（适用于已手动创建过超管、
|
||||
或本地开发环境无需自动建号的场景）。
|
||||
2. 调用 `setup_rbac_groups` 命令初始化/重置 RBAC 权限组（详见该命令的说明）。
|
||||
|
||||
安全说明：
|
||||
- 密码只应通过环境变量/密钥管理服务注入，不会被写入日志。
|
||||
- 本命令不会修改已存在用户的密码或权限，避免误覆盖人工调整过的账号状态。
|
||||
"""
|
||||
import os
|
||||
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.core.management import call_command
|
||||
from django.core.management.base import BaseCommand
|
||||
|
||||
|
||||
class Command(BaseCommand):
|
||||
help = "部署引导：按需创建首个超级管理员账号 + 初始化 RBAC 权限组，可重复执行。"
|
||||
|
||||
def handle(self, *args, **options):
|
||||
self._bootstrap_superuser()
|
||||
call_command("setup_rbac_groups")
|
||||
|
||||
def _bootstrap_superuser(self):
|
||||
username = os.environ.get("DJANGO_SUPERUSER_USERNAME")
|
||||
email = os.environ.get("DJANGO_SUPERUSER_EMAIL")
|
||||
password = os.environ.get("DJANGO_SUPERUSER_PASSWORD")
|
||||
|
||||
if not (username and email and password):
|
||||
self.stdout.write(
|
||||
self.style.WARNING(
|
||||
"未完整设置 DJANGO_SUPERUSER_USERNAME / DJANGO_SUPERUSER_EMAIL / "
|
||||
"DJANGO_SUPERUSER_PASSWORD，跳过自动创建超级管理员账号。"
|
||||
)
|
||||
)
|
||||
return
|
||||
|
||||
User = get_user_model()
|
||||
if User.objects.filter(username=username).exists():
|
||||
self.stdout.write(
|
||||
self.style.SUCCESS(f"超级管理员账号 '{username}' 已存在，跳过创建。")
|
||||
)
|
||||
return
|
||||
|
||||
User.objects.create_superuser(username=username, email=email, password=password)
|
||||
self.stdout.write(self.style.SUCCESS(f"已创建超级管理员账号 '{username}'。"))
|
||||
+170
-1
@@ -269,7 +269,6 @@ def test_setup_rbac_groups_page_permissions_match_role_matrix():
|
||||
# 编辑可以创建/编辑但不能发布
|
||||
editor_perms = page_codenames("编辑")
|
||||
assert editor_perms == {"add_page", "change_page", "view_page"}
|
||||
assert "publish_page" not in editor_perms
|
||||
# 审核员可以编辑并发布（用于审批发布场景）
|
||||
assert page_codenames("审核员") == {"change_page", "publish_page", "view_page"}
|
||||
# 作者只有 add + view：只能创建页面，仅能编辑自己拥有的页面（Wagtail 内置 ownership 策略）
|
||||
@@ -279,6 +278,105 @@ def test_setup_rbac_groups_page_permissions_match_role_matrix():
|
||||
assert page_codenames("查看者") == {"view_page"}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 健康检查端点（apps/core/views.py，供 K8s liveness/readiness 探针使用）
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_healthz_returns_ok(client):
|
||||
response = client.get("/healthz")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"status": "ok"}
|
||||
|
||||
|
||||
def test_readyz_returns_ok_when_database_available(client):
|
||||
response = client.get("/readyz")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"status": "ok"}
|
||||
|
||||
|
||||
def test_readyz_returns_503_when_database_unavailable(client, monkeypatch):
|
||||
from django.db import connection
|
||||
|
||||
def broken_cursor(*args, **kwargs):
|
||||
raise Exception("simulated db outage")
|
||||
|
||||
monkeypatch.setattr(connection, "cursor", broken_cursor)
|
||||
|
||||
response = client.get("/readyz")
|
||||
|
||||
assert response.status_code == 503
|
||||
assert response.json()["status"] == "error"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 发布后前端 revalidate webhook（apps/core/signals.py）
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_notify_frontend_revalidate_skips_when_url_not_configured(
|
||||
settings, monkeypatch, home_page
|
||||
):
|
||||
from wagtail.signals import page_published
|
||||
|
||||
settings.FRONTEND_REVALIDATE_URL = ""
|
||||
calls = []
|
||||
monkeypatch.setattr(
|
||||
"apps.core.signals.requests.post",
|
||||
lambda *args, **kwargs: calls.append((args, kwargs)),
|
||||
)
|
||||
|
||||
page_published.send(sender=type(home_page), instance=home_page)
|
||||
|
||||
assert calls == []
|
||||
|
||||
|
||||
def test_notify_frontend_revalidate_posts_webhook_when_configured(
|
||||
settings, monkeypatch, home_page
|
||||
):
|
||||
from wagtail.signals import page_published
|
||||
|
||||
settings.FRONTEND_REVALIDATE_URL = "https://frontend.example.com/api/revalidate"
|
||||
settings.REVALIDATE_SECRET = "test-secret"
|
||||
calls = []
|
||||
|
||||
def fake_post(url, json=None, headers=None, timeout=None):
|
||||
calls.append({"url": url, "json": json, "headers": headers})
|
||||
|
||||
class FakeResponse:
|
||||
status_code = 200
|
||||
|
||||
return FakeResponse()
|
||||
|
||||
monkeypatch.setattr("apps.core.signals.requests.post", fake_post)
|
||||
|
||||
page_published.send(sender=type(home_page), instance=home_page)
|
||||
|
||||
assert len(calls) == 1
|
||||
assert calls[0]["url"] == "https://frontend.example.com/api/revalidate"
|
||||
assert calls[0]["headers"] == {"Authorization": "Bearer test-secret"}
|
||||
assert f"page:{home_page.id}" in calls[0]["json"]["tags"]
|
||||
|
||||
|
||||
def test_notify_frontend_revalidate_swallows_request_exception(
|
||||
settings, monkeypatch, home_page
|
||||
):
|
||||
import requests as requests_module
|
||||
from wagtail.signals import page_published
|
||||
|
||||
settings.FRONTEND_REVALIDATE_URL = "https://frontend.example.com/api/revalidate"
|
||||
|
||||
def raise_request_exception(*args, **kwargs):
|
||||
raise requests_module.RequestException("network error")
|
||||
|
||||
monkeypatch.setattr("apps.core.signals.requests.post", raise_request_exception)
|
||||
|
||||
# 不应向外抛出异常——webhook 失败不能影响页面发布流程。
|
||||
page_published.send(sender=type(home_page), instance=home_page)
|
||||
|
||||
|
||||
def test_setup_rbac_groups_author_can_only_edit_own_pages(home_page):
|
||||
"""验证“Author 仅能编辑自己的页面”这一矩阵要求，由 Wagtail 内置的
|
||||
OwnershipPermissionPolicy（仅 add 权限）自然实现，无需自定义 hook。"""
|
||||
@@ -363,3 +461,74 @@ def test_setup_rbac_groups_is_idempotent():
|
||||
group=group, page=root_page, permission__codename="add_page"
|
||||
).count() == 1
|
||||
assert group.permissions.filter(codename="access_admin").count() == 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 部署引导（bootstrap_deployment 管理命令）
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_bootstrap_deployment_creates_superuser_from_env_vars(monkeypatch):
|
||||
from django.core.management import call_command
|
||||
|
||||
monkeypatch.setenv("DJANGO_SUPERUSER_USERNAME", "deploy-admin")
|
||||
monkeypatch.setenv("DJANGO_SUPERUSER_EMAIL", "deploy-admin@example.com")
|
||||
monkeypatch.setenv("DJANGO_SUPERUSER_PASSWORD", "s3cret-pass")
|
||||
|
||||
call_command("bootstrap_deployment")
|
||||
|
||||
User = get_user_model()
|
||||
user = User.objects.get(username="deploy-admin")
|
||||
assert user.is_superuser is True
|
||||
assert user.is_staff is True
|
||||
assert user.email == "deploy-admin@example.com"
|
||||
assert user.check_password("s3cret-pass") is True
|
||||
|
||||
|
||||
def test_bootstrap_deployment_skips_superuser_creation_when_env_vars_missing(monkeypatch):
|
||||
from django.core.management import call_command
|
||||
|
||||
monkeypatch.delenv("DJANGO_SUPERUSER_USERNAME", raising=False)
|
||||
monkeypatch.delenv("DJANGO_SUPERUSER_EMAIL", raising=False)
|
||||
monkeypatch.delenv("DJANGO_SUPERUSER_PASSWORD", raising=False)
|
||||
|
||||
User = get_user_model()
|
||||
count_before = User.objects.count()
|
||||
|
||||
call_command("bootstrap_deployment")
|
||||
|
||||
assert User.objects.count() == count_before
|
||||
|
||||
|
||||
def test_bootstrap_deployment_does_not_overwrite_existing_superuser_password(monkeypatch):
|
||||
from django.core.management import call_command
|
||||
|
||||
User = get_user_model()
|
||||
User.objects.create_superuser(
|
||||
username="deploy-admin", email="old@example.com", password="original-pass"
|
||||
)
|
||||
|
||||
monkeypatch.setenv("DJANGO_SUPERUSER_USERNAME", "deploy-admin")
|
||||
monkeypatch.setenv("DJANGO_SUPERUSER_EMAIL", "new@example.com")
|
||||
monkeypatch.setenv("DJANGO_SUPERUSER_PASSWORD", "new-pass")
|
||||
|
||||
call_command("bootstrap_deployment")
|
||||
|
||||
user = User.objects.get(username="deploy-admin")
|
||||
assert user.email == "old@example.com"
|
||||
assert user.check_password("original-pass") is True
|
||||
assert user.check_password("new-pass") is False
|
||||
|
||||
|
||||
def test_bootstrap_deployment_also_initializes_rbac_groups(monkeypatch):
|
||||
from django.contrib.auth.models import Group
|
||||
from django.core.management import call_command
|
||||
|
||||
monkeypatch.delenv("DJANGO_SUPERUSER_USERNAME", raising=False)
|
||||
monkeypatch.delenv("DJANGO_SUPERUSER_EMAIL", raising=False)
|
||||
monkeypatch.delenv("DJANGO_SUPERUSER_PASSWORD", raising=False)
|
||||
|
||||
call_command("bootstrap_deployment")
|
||||
|
||||
expected_names = {"站点管理员", "编辑", "审核员", "作者", "查看者"}
|
||||
assert expected_names <= set(Group.objects.values_list("name", flat=True))
|
||||
|
||||
@@ -0,0 +1,53 @@
|
||||
"""apps.home 单元测试：HomePage 页面树结构。
|
||||
|
||||
test_home_page_subpage_types_allows_all_top_level_sections 是一条回归测试：
|
||||
此前 HomePage.subpage_types 曾经只包含 blog.BlogIndexPage，导致 products/cases/
|
||||
solutions 的 IndexPage 无法通过 Wagtail 管理后台"添加子页面"界面创建（subpage_types/
|
||||
parent_page_types 只影响后台创建校验，不影响 ORM add_child()，问题一度被脚本化测试掩盖）。
|
||||
"""
|
||||
import pytest
|
||||
|
||||
from apps.blog.models import BlogIndexPage
|
||||
from apps.cases.models import CaseStudyIndexPage
|
||||
from apps.core.models import SimpleContentPage
|
||||
from apps.home.models import HomePage
|
||||
from apps.products.models import ProductIndexPage
|
||||
from apps.solutions.models import SolutionIndexPage
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
|
||||
def test_home_page_subpage_types_allows_all_top_level_sections():
|
||||
expected_types = {
|
||||
"blog.BlogIndexPage",
|
||||
"products.ProductIndexPage",
|
||||
"solutions.SolutionIndexPage",
|
||||
"cases.CaseStudyIndexPage",
|
||||
"core.SimpleContentPage",
|
||||
}
|
||||
|
||||
assert expected_types.issubset(set(HomePage.subpage_types))
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"index_model,kwargs",
|
||||
[
|
||||
(BlogIndexPage, {"title": "博客", "slug": "blog"}),
|
||||
(ProductIndexPage, {"title": "产品", "slug": "products"}),
|
||||
(SolutionIndexPage, {"title": "解决方案", "slug": "solutions"}),
|
||||
(CaseStudyIndexPage, {"title": "案例", "slug": "cases"}),
|
||||
(SimpleContentPage, {"title": "隐私政策", "slug": "privacy-policy"}),
|
||||
],
|
||||
)
|
||||
def test_each_top_level_section_can_be_created_under_home_in_admin(
|
||||
home_page, index_model, kwargs
|
||||
):
|
||||
"""`can_create_at` 是 Wagtail 管理后台"添加子页面"按钮实际使用的判断（综合
|
||||
parent.subpage_types 与 child.parent_page_types 两个方向），此前的 bug 正是
|
||||
仅靠 ORM `add_child()` 无法暴露的——它只影响管理后台，不影响直接调用 ORM。"""
|
||||
assert index_model.can_create_at(home_page) is True
|
||||
|
||||
child = index_model(**kwargs)
|
||||
home_page.add_child(instance=child)
|
||||
|
||||
assert index_model.objects.child_of(home_page).count() == 1
|
||||
@@ -0,0 +1,39 @@
|
||||
"""apps.solutions 单元测试：页面树结构与列表上下文（与 apps.products/apps.cases 同构）。"""
|
||||
import pytest
|
||||
|
||||
from apps.solutions.models import SolutionIndexPage, SolutionPage
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def solution_index(home_page):
|
||||
index = SolutionIndexPage(title="解决方案", slug="solutions", intro="行业解决方案")
|
||||
home_page.add_child(instance=index)
|
||||
return index
|
||||
|
||||
|
||||
def test_solution_page_can_be_created_under_index(solution_index):
|
||||
solution = SolutionPage(
|
||||
title="制造业解决方案",
|
||||
slug="manufacturing",
|
||||
industry="制造业",
|
||||
summary="面向制造业的数字化转型方案",
|
||||
)
|
||||
solution_index.add_child(instance=solution)
|
||||
|
||||
assert SolutionPage.objects.live().descendant_of(solution_index).count() == 1
|
||||
|
||||
|
||||
def test_solution_index_context_excludes_draft_solutions(solution_index):
|
||||
live_solution = SolutionPage(title="方案 A", slug="solution-a")
|
||||
solution_index.add_child(instance=live_solution)
|
||||
|
||||
draft_solution = SolutionPage(title="方案 B", slug="solution-b", live=False)
|
||||
solution_index.add_child(instance=draft_solution)
|
||||
|
||||
context = solution_index.get_context(request=None)
|
||||
titles = [s.title for s in context["solutions"]]
|
||||
|
||||
assert "方案 A" in titles
|
||||
assert "方案 B" not in titles
|
||||
Reference in New Issue
Block a user