wagtailcms/apps/core/tests.py

"""apps.core 基础单元测试：SEOablePage 抽象基类字段与公共 Block。"""
import pytest
from django.contrib.auth import get_user_model
from rest_framework.test import APIClient

from apps.core.blocks import FormBlock, TeamMemberChooserBlock
from apps.core.models import (
    NavigationMenu,
    NavigationMenuItem,
    Partner,
    SimpleContentPage,
    SiteSettings,
    TeamMember,
    Testimonial,
)
from apps.forms.models import FormDefinition, FormDefinitionField

pytestmark = pytest.mark.django_db


def test_seoable_page_fields_available_on_subclass(home_page):
    """SEOablePage 是抽象基类，其 SEO 字段应出现在具体子类（HomePage）上。"""
    home_page.seo_title_override = "自定义标题"
    home_page.seo_description_override = "自定义描述"
    home_page.canonical_url = "https://example.com/"
    home_page.schema_json = {"@type": "Organization"}
    home_page.save()

    home_page.refresh_from_db()
    assert home_page.seo_title_override == "自定义标题"
    assert home_page.seo_description_override == "自定义描述"
    assert home_page.canonical_url == "https://example.com/"
    assert home_page.schema_json == {"@type": "Organization"}


def test_form_block_api_representation_expands_fields():
    """FormBlock.get_api_representation 应展开 FormDefinition 及其子字段，
    而不是仅返回 SnippetChooserBlock 默认的主键。"""
    form_def = FormDefinition.objects.create(
        name="联系我们",
        submit_button_text="立即提交",
        success_message="已收到您的信息",
    )
    FormDefinitionField.objects.create(
        form=form_def,
        label="姓名",
        field_key="name",
        field_type="text",
        required=True,
    )
    FormDefinitionField.objects.create(
        form=form_def,
        label="意向",
        field_key="interest",
        field_type="select",
        required=False,
        choices="产品咨询, 合作洽谈",
    )

    block = FormBlock()
    result = block.get_api_representation({"form": form_def})

    assert result["form"]["id"] == form_def.pk
    assert result["form"]["name"] == "联系我们"
    assert result["form"]["submit_button_text"] == "立即提交"
    assert len(result["form"]["fields"]) == 2
    assert result["form"]["fields"][1]["choices"] == ["产品咨询", "合作洽谈"]


def test_form_block_api_representation_handles_empty_form():
    block = FormBlock()
    result = block.get_api_representation({"form": None})
    assert result == {"form": None}


def test_simple_content_page_can_be_created_under_home(home_page):
    """隐私政策等法务页面应可作为 HomePage 的子页面创建，并保留富文本内容。"""
    page = SimpleContentPage(
        title="隐私政策",
        slug="privacy-policy",
        body="<p>我们重视您的个人信息保护。</p>",
    )
    home_page.add_child(instance=page)

    page.refresh_from_db()
    assert SimpleContentPage.objects.live().descendant_of(home_page).count() == 1
    assert "个人信息保护" in page.body


def test_team_member_default_ordering():
    """团队成员应按 order 字段升序排列。"""
    TeamMember.objects.create(name="李四", order=2)
    TeamMember.objects.create(name="张三", order=1)
    names = list(TeamMember.objects.values_list("name", flat=True))
    assert names == ["张三", "李四"]


def test_testimonial_str_includes_author_and_quote():
    testimonial = Testimonial.objects.create(
        quote="非常专业的服务，效果超出预期。", author_name="王经理"
    )
    assert "王经理" in str(testimonial)


def test_partner_str_returns_name():
    partner = Partner.objects.create(name="示例合作伙伴")
    assert str(partner) == "示例合作伙伴"


def test_navigation_menu_item_url_prefers_external_link_over_page(home_page):
    menu = NavigationMenu.objects.create(name="main")
    item = NavigationMenuItem.objects.create(
        menu=menu, label="首页", link_page=home_page, link_url="https://example.com"
    )
    assert item.url == "https://example.com"


def test_navigation_menu_item_url_falls_back_to_page_url(home_page):
    menu = NavigationMenu.objects.create(name="main")
    item = NavigationMenuItem.objects.create(menu=menu, label="首页", link_page=home_page)
    assert item.url == home_page.url


def test_navigation_menu_item_url_defaults_to_hash_when_empty():
    menu = NavigationMenu.objects.create(name="main")
    item = NavigationMenuItem.objects.create(menu=menu, label="占位")
    assert item.url == "#"


def test_site_settings_for_site_creates_singleton_with_defaults():
    from wagtail.models import Site

    site = Site.objects.get(is_default_site=True)
    settings_obj = SiteSettings.for_site(site)
    assert settings_obj.icp_url == "https://beian.miit.gov.cn/"
    # 再次获取应复用同一条记录而非重复创建
    assert SiteSettings.for_site(site).pk == settings_obj.pk


def test_team_api_returns_ordered_members():
    TeamMember.objects.create(name="李四", role="工程师", order=2)
    TeamMember.objects.create(name="张三", role="总经理", order=1)

    client = APIClient()
    response = client.get("/api/v1/custom/core/team/")

    assert response.status_code == 200
    items = response.data["items"]
    assert [item["name"] for item in items] == ["张三", "李四"]
    assert items[0]["role"] == "总经理"


def test_testimonial_api_returns_list():
    Testimonial.objects.create(quote="效果很好", author_name="客户A")

    client = APIClient()
    response = client.get("/api/v1/custom/core/testimonials/")

    assert response.status_code == 200
    assert response.data["items"][0]["author_name"] == "客户A"


def test_team_member_chooser_block_api_representation_expands_fields():
    """TeamMemberChooserBlock.get_api_representation 应展开 TeamMember 的完整字段，
    而不是仅返回 SnippetChooserBlock 默认的主键。"""
    member = TeamMember.objects.create(name="王五", role="产品经理", bio="十年行业经验")

    block = TeamMemberChooserBlock()
    result = block.get_api_representation(member)

    assert result["id"] == member.pk
    assert result["name"] == "王五"
    assert result["role"] == "产品经理"
    assert result["bio"] == "十年行业经验"
    assert result["photo"] is None


def test_team_member_chooser_block_api_representation_handles_none():
    block = TeamMemberChooserBlock()
    assert block.get_api_representation(None) is None


def test_partner_api_returns_list():
    Partner.objects.create(name="伙伴A", website_url="https://partner-a.example.com")

    client = APIClient()
    response = client.get("/api/v1/custom/core/partners/")

    assert response.status_code == 200
    assert response.data["items"][0]["name"] == "伙伴A"


def test_navigation_menu_api_returns_items_by_name(home_page):
    menu = NavigationMenu.objects.create(name="main")
    NavigationMenuItem.objects.create(menu=menu, label="首页", link_page=home_page, sort_order=0)
    NavigationMenuItem.objects.create(menu=menu, label="外部链接", link_url="https://example.com", sort_order=1)

    client = APIClient()
    response = client.get("/api/v1/custom/core/navigation/?name=main")

    assert response.status_code == 200
    assert response.data["name"] == "main"
    assert len(response.data["items"]) == 2
    assert response.data["items"][1]["url"] == "https://example.com"


def test_navigation_menu_api_404_when_missing():
    client = APIClient()
    response = client.get("/api/v1/custom/core/navigation/?name=does-not-exist")
    assert response.status_code == 404


def test_site_settings_api_returns_defaults():
    client = APIClient()
    response = client.get("/api/v1/custom/core/site-settings/")

    assert response.status_code == 200
    assert response.data["icp_url"] == "https://beian.miit.gov.cn/"


# ---------------------------------------------------------------------------
# RBAC 权限组（setup_rbac_groups 管理命令）
# ---------------------------------------------------------------------------


def test_setup_rbac_groups_creates_all_expected_groups():
    from django.contrib.auth.models import Group
    from django.core.management import call_command

    call_command("setup_rbac_groups")

    expected_names = {"站点管理员", "编辑", "审核员", "作者", "查看者"}
    assert expected_names <= set(Group.objects.values_list("name", flat=True))


def test_setup_rbac_groups_grants_access_admin_to_all_groups():
    from django.contrib.auth.models import Group
    from django.core.management import call_command

    call_command("setup_rbac_groups")

    for name in ["站点管理员", "编辑", "审核员", "作者", "查看者"]:
        group = Group.objects.get(name=name)
        assert group.permissions.filter(codename="access_admin").exists()


def test_setup_rbac_groups_page_permissions_match_role_matrix():
    from django.contrib.auth.models import Group
    from django.core.management import call_command
    from wagtail.models import GroupPagePermission, Page

    call_command("setup_rbac_groups")

    root_page = Page.objects.get(depth=1)

    def page_codenames(group_name):
        group = Group.objects.get(name=group_name)
        return set(
            GroupPagePermission.objects.filter(group=group, page=root_page).values_list(
                "permission__codename", flat=True
            )
        )

    # 站点管理员拥有全部页面权限
    assert page_codenames("站点管理员") == {
        "add_page", "change_page", "publish_page", "delete_page",
        "bulk_delete_page", "lock_page", "unlock_page", "view_page",
    }
    # 编辑可以创建/编辑但不能发布
    editor_perms = page_codenames("编辑")
    assert editor_perms == {"add_page", "change_page", "view_page"}
    # 审核员可以编辑并发布（用于审批发布场景）
    assert page_codenames("审核员") == {"change_page", "publish_page", "view_page"}
    # 作者只有 add + view：只能创建页面，仅能编辑自己拥有的页面（Wagtail 内置 ownership 策略）
    assert page_codenames("作者") == {"add_page", "view_page"}
    assert "change_page" not in page_codenames("作者")
    # 查看者仅能查看
    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。"""
    from django.contrib.auth.models import Group
    from django.core.management import call_command

    call_command("setup_rbac_groups")

    User = get_user_model()
    author = User.objects.create_user(username="author-user", password="pass1234")
    other_author = User.objects.create_user(username="other-author", password="pass1234")
    author.groups.add(Group.objects.get(name="作者"))
    other_author.groups.add(Group.objects.get(name="作者"))

    own_page = SimpleContentPage(title="自己的页面", slug="own-page", owner=author)
    home_page.add_child(instance=own_page)

    others_page = SimpleContentPage(title="他人的页面", slug="others-page", owner=other_author)
    home_page.add_child(instance=others_page)

    assert own_page.permissions_for_user(author).can_edit() is True
    assert others_page.permissions_for_user(author).can_edit() is False


def test_setup_rbac_groups_snippet_permissions_restrict_lead_to_admin_and_reviewer():
    from django.contrib.auth.models import Group
    from django.core.management import call_command

    call_command("setup_rbac_groups")

    def lead_codenames(group_name):
        group = Group.objects.get(name=group_name)
        return set(
            group.permissions.filter(content_type__app_label="forms", codename__endswith="lead")
            .values_list("codename", flat=True)
        )

    assert lead_codenames("站点管理员") == {"add_lead", "change_lead", "delete_lead", "view_lead"}
    assert lead_codenames("审核员") == {"view_lead"}
    assert lead_codenames("编辑") == set()
    assert lead_codenames("作者") == set()
    assert lead_codenames("查看者") == set()


def test_setup_rbac_groups_only_site_admin_has_user_management_permissions():
    from django.contrib.auth.models import Group
    from django.core.management import call_command

    call_command("setup_rbac_groups")

    User = get_user_model()
    user_ct_app_label = User._meta.app_label

    for name in ["编辑", "审核员", "作者", "查看者"]:
        group = Group.objects.get(name=name)
        assert not group.permissions.filter(
            content_type__app_label=user_ct_app_label, content_type__model="user"
        ).exists()

    site_admin = Group.objects.get(name="站点管理员")
    site_admin_codenames = set(
        site_admin.permissions.filter(
            content_type__app_label=user_ct_app_label, content_type__model="user"
        ).values_list("codename", flat=True)
    )
    assert site_admin_codenames == {"add_user", "change_user", "delete_user", "view_user"}


def test_setup_rbac_groups_is_idempotent():
    from django.contrib.auth.models import Group
    from django.core.management import call_command
    from wagtail.models import GroupPagePermission, Page

    call_command("setup_rbac_groups")
    call_command("setup_rbac_groups")

    assert Group.objects.filter(name="站点管理员").count() == 1
    group = Group.objects.get(name="站点管理员")
    root_page = Page.objects.get(depth=1)
    # 重复运行不应产生重复的页面权限记录
    assert GroupPagePermission.objects.filter(
        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))