feat: 静态通知同步 + 积分审计 JSONB 序列化修复
This commit is contained in:
@@ -0,0 +1,181 @@
|
||||
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,
|
||||
)
|
||||
|
||||
|
||||
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: Welcome
|
||||
body: 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.payload.action == "open_route"
|
||||
assert loaded.targets.mode == "user_ids"
|
||||
assert len(loaded.targets.user_ids or []) == 1
|
||||
|
||||
|
||||
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: Invalid
|
||||
body: Invalid targets.
|
||||
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_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: First
|
||||
body: First body.
|
||||
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: Second
|
||||
body: Second body.
|
||||
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: Title A
|
||||
body: Body A.
|
||||
payload:
|
||||
action: none
|
||||
targets:
|
||||
mode: all_users
|
||||
""",
|
||||
)
|
||||
_write_yaml(
|
||||
second_path,
|
||||
"""
|
||||
notification:
|
||||
source_key: same_key
|
||||
version: 1
|
||||
type: system
|
||||
status: published
|
||||
title: Title B
|
||||
body: Body 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: Deleted
|
||||
body: Deleted body.
|
||||
payload:
|
||||
action: none
|
||||
targets:
|
||||
mode: all_users
|
||||
""",
|
||||
)
|
||||
|
||||
loaded = load_static_notification_file(file_path)
|
||||
|
||||
assert loaded.notification.deleted is True
|
||||
Reference in New Issue
Block a user