eryao/backend/tests/unit/test_static_notification_sync.py

from __future__ import annotations

from pathlib import Path
import textwrap

import pytest

from core.config.notification.static_schema import load_static_notification_file
from core.config.notification.static_sync import (
    build_static_notification_content_hash,
    load_static_notification_documents,
)
from schemas.enums import NotificationTargetMode


def _write_yaml(path: Path, content: str) -> None:
    path.write_text(textwrap.dedent(content).strip() + "\n", encoding="utf-8")


def test_load_static_notification_file_parses_valid_yaml(tmp_path: Path) -> None:
    file_path = tmp_path / "welcome.yaml"
    _write_yaml(
        file_path,
        """
        notification:
          source_key: welcome_bonus
          version: 1
          type: system
          status: published
          title:
            zh: 欢迎
            en: Welcome
          body:
            zh: 欢迎使用
            en: Welcome to the app.
          payload:
            action: open_route
            route: /points
            tab: balance
        targets:
          mode: user_ids
          user_ids:
            - 11111111-1111-1111-1111-111111111111
        """,
    )

    loaded = load_static_notification_file(file_path)

    assert loaded.notification.source_key == "welcome_bonus"
    assert loaded.notification.title == {"zh": "欢迎", "en": "Welcome"}
    assert loaded.notification.body == {"zh": "欢迎使用", "en": "Welcome to the app."}
    assert loaded.notification.payload.action == "open_route"
    assert loaded.targets.mode == NotificationTargetMode.USER_IDS
    assert len(loaded.targets.user_ids or []) == 1


def test_load_static_notification_file_parses_new_users(tmp_path: Path) -> None:
    file_path = tmp_path / "welcome_points.yaml"
    _write_yaml(
        file_path,
        """
        notification:
          source_key: welcome_points
          version: 1
          type: system
          status: published
          title:
            zh: 欢迎
          body:
            zh: 你好
          payload:
            action: open_route
            route: /points
            tab: balance
        targets:
          mode: new_users
        """,
    )

    loaded = load_static_notification_file(file_path)

    assert loaded.targets.mode == NotificationTargetMode.NEW_USERS
    assert loaded.targets.user_ids is None


def test_load_static_notification_file_parses_exist_users(tmp_path: Path) -> None:
    file_path = tmp_path / "promo.yaml"
    _write_yaml(
        file_path,
        """
        notification:
          source_key: promo_return
          version: 1
          type: system
          status: published
          title:
            zh: 回来吧
          body:
            zh: 想你
          payload:
            action: none
        targets:
          mode: exist_users
        """,
    )

    loaded = load_static_notification_file(file_path)

    assert loaded.targets.mode == NotificationTargetMode.EXIST_USERS
    assert loaded.targets.user_ids is None


def test_load_static_notification_file_parses_all_users(tmp_path: Path) -> None:
    file_path = tmp_path / "announce.yaml"
    _write_yaml(
        file_path,
        """
        notification:
          source_key: system_announce
          version: 1
          type: system
          status: published
          title:
            zh: 公告
          body:
            zh: 午夜维护
          payload:
            action: none
        targets:
          mode: all_users
        """,
    )

    loaded = load_static_notification_file(file_path)

    assert loaded.targets.mode == NotificationTargetMode.ALL_USERS


def test_load_static_notification_file_rejects_invalid_targets(tmp_path: Path) -> None:
    file_path = tmp_path / "invalid.yaml"
    _write_yaml(
        file_path,
        """
        notification:
          source_key: invalid_targets
          version: 1
          type: system
          status: published
          title:
            zh: 无效
          body:
            zh: 无效
          payload:
            action: none
        targets:
          mode: all_users
          user_ids:
            - 11111111-1111-1111-1111-111111111111
        """,
    )

    with pytest.raises(ValueError, match="Invalid static notification data"):
        load_static_notification_file(file_path)


def test_load_static_notification_file_rejects_i18n_without_zh(
    tmp_path: Path,
) -> None:
    file_path = tmp_path / "missing_zh.yaml"
    _write_yaml(
        file_path,
        """
        notification:
          source_key: missing_zh
          version: 1
          type: system
          status: published
          title:
            en: Welcome
          body:
            zh: 正文
          payload:
            action: none
        targets:
          mode: all_users
        """,
    )

    with pytest.raises(ValueError, match="Invalid static notification data"):
        load_static_notification_file(file_path)


def test_load_static_notification_file_rejects_empty_i18n_text(
    tmp_path: Path,
) -> None:
    file_path = tmp_path / "empty_i18n.yaml"
    _write_yaml(
        file_path,
        """
        notification:
          source_key: empty_i18n
          version: 1
          type: system
          status: published
          title:
            zh: ""
          body:
            zh: 正文
          payload:
            action: none
        targets:
          mode: all_users
        """,
    )

    with pytest.raises(ValueError, match="Invalid static notification data"):
        load_static_notification_file(file_path)


def test_load_static_notification_file_rejects_unknown_i18n_locale(
    tmp_path: Path,
) -> None:
    file_path = tmp_path / "unknown_locale.yaml"
    _write_yaml(
        file_path,
        """
        notification:
          source_key: unknown_locale
          version: 1
          type: system
          status: published
          title:
            zh: 标题
            ja: タイトル
          body:
            zh: 正文
          payload:
            action: none
        targets:
          mode: all_users
        """,
    )

    with pytest.raises(ValueError, match="Invalid static notification data"):
        load_static_notification_file(file_path)


def test_load_static_notification_file_rejects_unknown_mode(tmp_path: Path) -> None:
    file_path = tmp_path / "bad_mode.yaml"
    _write_yaml(
        file_path,
        """
        notification:
          source_key: bad_mode
          version: 1
          type: system
          status: published
          title:
            zh: 坏
          body:
            zh: 坏
          payload:
            action: none
        targets:
          mode: non_existent
        """,
    )

    with pytest.raises(ValueError, match="Invalid static notification data"):
        load_static_notification_file(file_path)


def test_load_static_notification_file_rejects_new_users_with_user_ids(
    tmp_path: Path,
) -> None:
    file_path = tmp_path / "bad_new_users.yaml"
    _write_yaml(
        file_path,
        """
        notification:
          source_key: bad_new
          version: 1
          type: system
          status: published
          title:
            zh: 坏
          body:
            zh: 坏
          payload:
            action: none
        targets:
          mode: new_users
          user_ids:
            - 11111111-1111-1111-1111-111111111111
        """,
    )

    with pytest.raises(ValueError, match="Invalid static notification data"):
        load_static_notification_file(file_path)


def test_load_static_notification_file_rejects_user_ids_without_list(
    tmp_path: Path,
) -> None:
    file_path = tmp_path / "bad_user_ids.yaml"
    _write_yaml(
        file_path,
        """
        notification:
          source_key: bad_uids
          version: 1
          type: system
          status: published
          title:
            zh: 坏
          body:
            zh: 坏
          payload:
            action: none
        targets:
          mode: user_ids
        """,
    )

    with pytest.raises(ValueError, match="Invalid static notification data"):
        load_static_notification_file(file_path)


def test_load_static_notification_documents_rejects_duplicate_source_key(
    tmp_path: Path,
) -> None:
    _write_yaml(
        tmp_path / "first.yaml",
        """
        notification:
          source_key: duplicate_key
          version: 1
          type: system
          status: published
          title:
            zh: 第一
          body:
            zh: 第一
          payload:
            action: none
        targets:
          mode: all_users
        """,
    )
    _write_yaml(
        tmp_path / "second.yaml",
        """
        notification:
          source_key: duplicate_key
          version: 1
          type: system
          status: published
          title:
            zh: 第二
          body:
            zh: 第二
          payload:
            action: none
        targets:
          mode: all_users
        """,
    )

    with pytest.raises(ValueError, match="Duplicate static notification source_key"):
        load_static_notification_documents(path=tmp_path)


def test_content_hash_changes_when_notification_changes(tmp_path: Path) -> None:
    first_path = tmp_path / "first.yaml"
    second_path = tmp_path / "second.yaml"
    _write_yaml(
        first_path,
        """
        notification:
          source_key: same_key
          version: 1
          type: system
          status: published
          title:
            zh: 标题A
          body:
            zh: 正文A
          payload:
            action: none
        targets:
          mode: all_users
        """,
    )
    _write_yaml(
        second_path,
        """
        notification:
          source_key: same_key
          version: 1
          type: system
          status: published
          title:
            zh: 标题B
          body:
            zh: 正文A
          payload:
            action: none
        targets:
          mode: all_users
        """,
    )

    first = load_static_notification_file(first_path)
    second = load_static_notification_file(second_path)

    assert build_static_notification_content_hash(
        first
    ) != build_static_notification_content_hash(second)


def test_load_static_notification_file_supports_deleted_flag(tmp_path: Path) -> None:
    file_path = tmp_path / "deleted.yaml"
    _write_yaml(
        file_path,
        """
        notification:
          source_key: deleted_notice
          version: 1
          type: system
          status: revoked
          deleted: true
          title:
            zh: 已删
          body:
            zh: 已删
          payload:
            action: none
        targets:
          mode: all_users
        """,
    )

    loaded = load_static_notification_file(file_path)

    assert loaded.notification.deleted is True