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: 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 == 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: Welcome body: You got points. 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: Come back body: We miss you. 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: Announcement body: Maintenance at midnight. 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: 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_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: Bad body: Bad mode. 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: Bad body: Bad. 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: Bad body: Bad. 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: 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