feat(B-NOTIF): Notification→Message Konsolidierung — System-Channel, post_system_message(), Migration
Check Cross-Plugin Imports / check (push) Has been cancelled

B-NOTIF-SYS: CommConversation um is_system Feld erweitert, get_or_create_system_channel()
B-NOTIF-EVT: post_system_message() in kommunikation/services.py — erstellt CommMessage im System-Channel
- create_notification() als deprecated Wrapper delegiert auf post_system_message()
- Text-Block + action_card Block mit Deep-Link, Metadata: notification_type/severity/entity_ref
B-NOTIF-PREF: NotificationPreference als Routing-Konfiguration, stumm/disabled respektiert
B-NOTIF-MIG: Migration 0120 — Notifications → CommMessages migriert, notifications_legacy View
B-NOTIF-DEPREC: Notification-Routes als deprecated markiert, keine Routes entfernt
B-NOTIF-TEST: 19 Tests in test_notification_migration.py — alle grün
- System-Channel, post_system_message, create_notification Wrapper, Preferences, Unread-Badge, Migration Mapping, Sensitive Fields
This commit is contained in:
Agent Zero
2026-08-13 21:19:41 +02:00
parent 78963f2ca9
commit 1baa9481a2
7 changed files with 1010 additions and 42 deletions
+6 -6
View File
@@ -148,13 +148,13 @@
| Task | Status | Forgejo Issue | Verifiziert | | Task | Status | Forgejo Issue | Verifiziert |
|------|-------|---------------|------------| |------|-------|---------------|------------|
| B-NOTIF-SYS | `not_started` | — | — | | B-NOTIF-SYS | `done` | — | 19 tests pass |
| B-NOTIF-EVT | `not_started` | — | — | | B-NOTIF-EVT | `done` | — | post_system_message + create_notification wrapper |
| B-NOTIF-UI | `not_started` | — | — | | B-NOTIF-UI | `not_started` | — | — |
| B-NOTIF-PREF | `not_started` | — | — | | B-NOTIF-PREF | `done` | — | NotificationPreference routing retained |
| B-NOTIF-MIG | `not_started` | — | — | | B-NOTIF-MIG | `done` | — | Alembic 0120 migration |
| B-NOTIF-DEPREC | `not_started` | — | — | | B-NOTIF-DEPREC | `done` | — | Routes + create_notification deprecated |
| B-NOTIF-TEST | `not_started` | — | — | | B-NOTIF-TEST | `done` | — | tests/test_notification_migration.py 19/19 pass |
--- ---
@@ -0,0 +1,144 @@
"""Add is_system column to comm_conversations and migrate notifications to system channel (B-NOTIF-*).
Adds is_system boolean to comm_conversations for system channel support.
Migrates existing notifications into the system channel as CommMessages.
Creates a view notifications_legacy as a compatibility layer over the old notifications table.
Revision ID: 0120
Revises: 0119
"""
from alembic import op
import sqlalchemy as sa
revision = "0120"
down_revision = "0119"
branch_labels = None
depends_on = None
def upgrade() -> None:
# 1. Add is_system column to comm_conversations
op.add_column(
"comm_conversations",
sa.Column("is_system", sa.Boolean(), nullable=False, server_default=sa.text("false")),
)
op.create_index(
"ix_comm_conversations_tenant_system",
"comm_conversations",
["tenant_id", "is_system"],
)
# 2. Create system channel per tenant (for tenants that have notifications)
op.execute("""
INSERT INTO comm_conversations (id, tenant_id, title, is_pinned, is_locked, is_direct, is_archived, is_system, created_by, created_by_type, metadata, created_at, updated_at)
SELECT
gen_random_uuid(),
n.tenant_id,
'System Channel',
false,
true,
false,
false,
true,
NULL,
'system',
'{}'::jsonb,
NOW(),
NOW()
FROM (
SELECT DISTINCT tenant_id FROM notifications WHERE deleted_at IS NULL
) n
WHERE NOT EXISTS (
SELECT 1 FROM comm_conversations cc
WHERE cc.tenant_id = n.tenant_id AND cc.is_system = true AND cc.deleted_at IS NULL
);
""")
# 3. Insert notifications as CommMessages in the system channel
op.execute("""
INSERT INTO comm_messages (id, tenant_id, conversation_id, sender_id, sender_type, content, content_format, metadata, created_at, updated_at)
SELECT
gen_random_uuid(),
n.tenant_id,
sc.id,
n.user_id,
'system',
COALESCE(n.title, '') || CASE WHEN n.body IS NOT NULL THEN E'\n' || n.body ELSE '' END,
'text',
jsonb_build_object(
'notification_type', n.type,
'severity', 'info',
'entity_ref', CASE WHEN n.entity_type IS NOT NULL THEN jsonb_build_object('entity_type', n.entity_type, 'entity_id', n.entity_id::text) ELSE NULL END,
'migrated_from_notification', true,
'original_notification_id', n.id::text
),
n.created_at,
COALESCE(n.read_at, n.created_at)
FROM notifications n
JOIN comm_conversations sc ON sc.tenant_id = n.tenant_id AND sc.is_system = true AND sc.deleted_at IS NULL
WHERE n.deleted_at IS NULL;
""")
# 4. Insert text blocks for each migrated message
op.execute("""
INSERT INTO comm_message_blocks (id, tenant_id, message_id, block_type, block_data, sort_order)
SELECT
gen_random_uuid(),
cm.tenant_id,
cm.id,
'text',
jsonb_build_object('text', cm.content),
0
FROM comm_messages cm
WHERE cm.metadata->>'migrated_from_notification' = 'true';
""")
# 5. Insert action_card blocks for messages with entity references
op.execute("""
INSERT INTO comm_message_blocks (id, tenant_id, message_id, block_type, block_data, sort_order)
SELECT
gen_random_uuid(),
cm.tenant_id,
cm.id,
'action_card',
jsonb_build_object(
'label', 'Open',
'entity_type', (cm.metadata->'entity_ref'->>'entity_type'),
'entity_id', (cm.metadata->'entity_ref'->>'entity_id')
),
1
FROM comm_messages cm
WHERE cm.metadata->>'migrated_from_notification' = 'true'
AND cm.metadata->'entity_ref' IS NOT NULL;
""")
# 6. For read notifications, create CommMessageRead entries
op.execute("""
INSERT INTO comm_message_reads (id, tenant_id, conversation_id, user_id, last_read_msg_id, last_read_at)
SELECT
gen_random_uuid(),
cm.tenant_id,
cm.conversation_id,
cm.sender_id,
cm.id,
COALESCE(n.read_at, n.created_at)
FROM comm_messages cm
JOIN notifications n ON n.id::text = cm.metadata->>'original_notification_id'
WHERE cm.metadata->>'migrated_from_notification' = 'true'
AND n.read_at IS NOT NULL
AND n.deleted_at IS NULL;
""")
# 7. Create legacy view over notifications table for backward compatibility
op.execute("DROP VIEW IF EXISTS notifications_legacy")
op.execute("CREATE VIEW notifications_legacy AS SELECT * FROM notifications")
def downgrade() -> None:
op.execute("DROP VIEW IF EXISTS notifications_legacy")
op.execute("DELETE FROM comm_message_blocks WHERE message_id IN (SELECT id FROM comm_messages WHERE metadata->>'migrated_from_notification' = 'true')")
op.execute("DELETE FROM comm_messages WHERE metadata->>'migrated_from_notification' = 'true'")
op.execute("DELETE FROM comm_conversations WHERE is_system = true AND title = 'System Channel'")
op.drop_index("ix_comm_conversations_tenant_system", table_name="comm_conversations")
op.drop_column("comm_conversations", "is_system")
+55 -35
View File
@@ -1,7 +1,13 @@
"""Notification service — create and manage user notifications.""" """Notification service — create and manage user notifications.
As of B-NOTIF-EVT, the primary entry point is post_system_message() which posts
to the Communication system channel. create_notification() is retained as a
deprecated backward-compat wrapper that delegates to post_system_message().
"""
from __future__ import annotations from __future__ import annotations
import logging
import uuid import uuid
from datetime import UTC from datetime import UTC
from typing import Any from typing import Any
@@ -15,6 +21,31 @@ from app.models.notification import (
NotificationType, NotificationType,
) )
logger = logging.getLogger(__name__)
# Re-export post_system_message from kommunikation services for convenience
async def post_system_message(
db: AsyncSession,
tenant_id: uuid.UUID,
user_id: uuid.UUID,
message_type: str,
title: str,
body: str | None = None,
entity_type: str | None = None,
entity_id: uuid.UUID | None = None,
severity: str = "info",
):
"""Post a typed system message to the tenant system channel.
Delegates to kommunikation.services.post_system_message.
Returns the created CommMessage, or None if the user has muted this type.
"""
from app.plugins.builtins.kommunikation.services import post_system_message as _post
return await _post(
db, tenant_id, user_id, message_type, title, body,
entity_type, entity_id, severity,
)
async def create_notification( async def create_notification(
db: AsyncSession, db: AsyncSession,
@@ -28,34 +59,24 @@ async def create_notification(
) -> Notification | None: ) -> Notification | None:
"""Create a new notification for a user if they have not disabled this type. """Create a new notification for a user if they have not disabled this type.
.. deprecated:: B-NOTIF-EVT
Use post_system_message() instead. This wrapper delegates to
post_system_message() and also creates a legacy Notification record
for backward compatibility with existing routes and frontend.
Returns None if the user has opted out of this notification type. Returns None if the user has opted out of this notification type.
""" """
# Check user preference # Delegate to post_system_message for the comm channel
pref = await db.execute( comm_msg = await post_system_message(
select(NotificationPreference).where( db, tenant_id, user_id, type, title, body,
and_( entity_type, entity_id, severity="info",
NotificationPreference.user_id == user_id,
NotificationPreference.type_key == type,
NotificationPreference.tenant_id == tenant_id,
) )
)
)
pref_row = pref.scalar_one_or_none()
# If preference exists and is disabled, skip if comm_msg is None:
if pref_row and not pref_row.is_enabled: # User has muted this type — don't create legacy record either
return None return None
# If no preference, check if type is enabled by default # Also create legacy Notification record for backward compat
if not pref_row:
type_def = await db.execute(
select(NotificationType).where(NotificationType.type_key == type)
)
type_row = type_def.scalar_one_or_none()
if type_row and not type_row.is_enabled_by_default:
return None
# Create notification
notif = Notification( notif = Notification(
tenant_id=tenant_id, tenant_id=tenant_id,
user_id=user_id, user_id=user_id,
@@ -68,17 +89,18 @@ async def create_notification(
db.add(notif) db.add(notif)
await db.flush() await db.flush()
# Publish notification.created event # Publish notification.created event (backward compat)
from app.core.event_bus import get_event_bus from app.core.event_bus import get_event_bus
event_bus = get_event_bus() event_bus = get_event_bus()
await event_bus.publish('notification.created', { await event_bus.publish("notification.created", {
'notification_id': str(notif.id), "notification_id": str(notif.id),
'tenant_id': str(tenant_id), "tenant_id": str(tenant_id),
'user_id': str(user_id), "user_id": str(user_id),
'type': type, "type": type,
'title': title, "title": title,
'entity_type': entity_type, "entity_type": entity_type,
'entity_id': str(entity_id) if entity_id else None, "entity_id": str(entity_id) if entity_id else None,
"comm_message_id": str(comm_msg.id),
}) })
return notif return notif
@@ -93,7 +115,6 @@ async def list_notifications(
) -> dict[str, Any]: ) -> dict[str, Any]:
"""List notifications for a user, unread first, then by created_at desc.""" """List notifications for a user, unread first, then by created_at desc."""
offset = (page - 1) * page_size offset = (page - 1) * page_size
# Count total
count_q = ( count_q = (
select(func.count()) select(func.count())
.select_from(Notification) .select_from(Notification)
@@ -104,7 +125,6 @@ async def list_notifications(
) )
total = (await db.execute(count_q)).scalar() or 0 total = (await db.execute(count_q)).scalar() or 0
# Query — unread first (read_at IS NULL), then newest
q = ( q = (
select(Notification) select(Notification)
.where( .where(
@@ -112,7 +132,7 @@ async def list_notifications(
Notification.user_id == user_id, Notification.user_id == user_id,
) )
.order_by( .order_by(
Notification.read_at.isnot(None), # False (unread) sorts first Notification.read_at.isnot(None),
Notification.created_at.desc(), Notification.created_at.desc(),
) )
.offset(offset) .offset(offset)
@@ -44,6 +44,7 @@ class CommConversation(Base, TenantMixin, OwnedMixin):
locked_by: Mapped[str | None] = mapped_column(String(100), nullable=True) locked_by: Mapped[str | None] = mapped_column(String(100), nullable=True)
is_direct: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) is_direct: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
is_archived: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) is_archived: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
is_system: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
created_by: Mapped[uuid.UUID | None] = mapped_column(PGUUID(as_uuid=True), nullable=True) created_by: Mapped[uuid.UUID | None] = mapped_column(PGUUID(as_uuid=True), nullable=True)
created_by_type: Mapped[str] = mapped_column(String(20), nullable=False, default="user") created_by_type: Mapped[str] = mapped_column(String(20), nullable=False, default="user")
last_msg_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) last_msg_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
@@ -1135,3 +1135,205 @@ async def create_plugin_room(
parts = [plugin_p, user_p] parts = [plugin_p, user_p]
return conversation_to_response(conv, parts, is_pinned_by_user=True) return conversation_to_response(conv, parts, is_pinned_by_user=True)
# ─── System Channel ───
async def get_or_create_system_channel(
db: AsyncSession,
tenant_id: uuid.UUID,
) -> CommConversation:
"""Get or create the tenant-wide system channel.
The system channel is a locked, is_system=True conversation that serves as
the central destination for system notifications, user alerts, and agent messages.
All users of the tenant are automatically added as participants.
"""
result = await db.execute(
select(CommConversation).where(
CommConversation.tenant_id == tenant_id,
CommConversation.is_system.is_(True),
CommConversation.deleted_at.is_(None),
)
)
conv = result.scalar_one_or_none()
if conv is not None:
return conv
# Create the system channel
conv = CommConversation(
tenant_id=tenant_id,
title="System Channel",
is_pinned=False,
is_locked=True,
is_direct=False,
is_archived=False,
is_system=True,
created_by=None,
created_by_type="system",
metadata_={},
)
db.add(conv)
await db.flush()
# Add all tenant users as participants
from app.models.user import User, UserTenant
users_result = await db.execute(
select(User.id)
.join(UserTenant, UserTenant.user_id == User.id)
.where(UserTenant.tenant_id == tenant_id)
)
user_ids = [row[0] for row in users_result.all()]
for uid in user_ids:
p = CommParticipant(
tenant_id=tenant_id,
conversation_id=conv.id,
participant_id=uid,
participant_type="user",
role="member",
)
db.add(p)
await db.flush()
return conv
async def post_system_message(
db: AsyncSession,
tenant_id: uuid.UUID,
user_id: uuid.UUID,
message_type: str,
title: str,
body: str | None = None,
entity_type: str | None = None,
entity_id: uuid.UUID | None = None,
severity: str = "info",
) -> CommMessage | None:
"""Post a typed system message to the tenant system channel.
Creates a CommMessage in the system channel with:
- A text block containing title + body
- An action_card block with deep-link if entity_type/entity_id is set
- Block/message metadata: notification_type, severity, entity_ref
Returns the created CommMessage, or None if the user has muted this type.
"""
# Check user preferences — reuse the notification preference system
from app.models.notification import NotificationPreference, NotificationType
pref = await db.execute(
select(NotificationPreference).where(
and_(
NotificationPreference.user_id == user_id,
NotificationPreference.type_key == message_type,
NotificationPreference.tenant_id == tenant_id,
)
)
)
pref_row = pref.scalar_one_or_none()
if pref_row and not pref_row.is_enabled:
return None
if not pref_row:
type_def = await db.execute(
select(NotificationType).where(NotificationType.type_key == message_type)
)
type_row = type_def.scalar_one_or_none()
if type_row and not type_row.is_enabled_by_default:
return None
# Get or create system channel
conv = await get_or_create_system_channel(db, tenant_id)
# Build message content
content = title
if body:
content = f"{title}\n{body}"
# Build metadata
msg_metadata: dict[str, Any] = {
"notification_type": message_type,
"severity": severity,
"target_user_id": str(user_id),
}
if entity_type and entity_id:
msg_metadata["entity_ref"] = {
"entity_type": entity_type,
"entity_id": str(entity_id),
}
# Build blocks
blocks: list[dict[str, Any]] = [
{
"block_type": "text",
"block_data": {"text": content, "title": title, "body": body or ""},
}
]
if entity_type and entity_id:
blocks.append(
{
"block_type": "action_card",
"block_data": {
"label": "Open",
"entity_type": entity_type,
"entity_id": str(entity_id),
},
}
)
# Create message directly (not via send_message to avoid trigger_depth issues)
msg = CommMessage(
tenant_id=tenant_id,
conversation_id=conv.id,
sender_id=None,
sender_type="system",
content=content,
content_format="text",
metadata_=msg_metadata,
)
db.add(msg)
await db.flush()
# Create blocks
for i, block in enumerate(blocks):
b = CommMessageBlock(
tenant_id=tenant_id,
message_id=msg.id,
block_type=block["block_type"],
block_data=block["block_data"],
sort_order=i,
)
db.add(b)
await db.flush()
# Update conversation last_msg
from datetime import datetime, timezone as dt_timezone
await db.execute(
update(CommConversation)
.where(CommConversation.id == conv.id)
.values(
last_msg_at=datetime.now(dt_timezone.utc),
last_msg_preview=content[:200],
last_msg_sender_type="system",
)
)
# Publish event
event_bus = get_event_bus()
await event_bus.publish("system.message.posted", {
"conversation_id": str(conv.id),
"message_id": str(msg.id),
"tenant_id": str(tenant_id),
"user_id": str(user_id),
"message_type": message_type,
"severity": severity,
})
return msg
+5 -1
View File
@@ -1,4 +1,8 @@
"""Notification routes.""" """Notification routes (deprecated — delegates to Communication system channel).
All notification endpoints are deprecated as of B-NOTIF-DEPREC. New code should use
the Communication system channel via post_system_message() instead.
"""
from __future__ import annotations from __future__ import annotations
+597
View File
@@ -0,0 +1,597 @@
"""Tests for notification-to-communication consolidation (B-NOTIF-* + B-NOTIF-TEST)."""
from __future__ import annotations
import uuid
from datetime import UTC, datetime
import pytest
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.notifications import create_notification, post_system_message
from app.models.notification import Notification, NotificationPreference, NotificationType
from app.plugins.builtins.kommunikation.models import (
CommConversation,
CommMessage,
CommMessageBlock,
CommMessageRead,
CommParticipant,
)
from app.plugins.builtins.kommunikation.services import get_or_create_system_channel
from tests.conftest import seed_tenant_and_users
@pytest.mark.asyncio
class TestSystemChannel:
"""B-NOTIF-SYS: System channel creation and properties."""
async def test_get_or_create_system_channel_creates_channel(self, db_session: AsyncSession):
"""get_or_create_system_channel creates a system channel with is_system=True."""
seed = await seed_tenant_and_users(db_session)
tenant_id = seed["tenant_a"].id
conv = await get_or_create_system_channel(db_session, tenant_id)
await db_session.flush()
assert conv is not None
assert conv.is_system is True
assert conv.is_locked is True
assert conv.title == "System Channel"
assert conv.created_by_type == "system"
assert conv.tenant_id == tenant_id
async def test_get_or_create_system_channel_is_idempotent(self, db_session: AsyncSession):
"""Calling get_or_create_system_channel twice returns the same channel."""
seed = await seed_tenant_and_users(db_session)
tenant_id = seed["tenant_a"].id
conv1 = await get_or_create_system_channel(db_session, tenant_id)
await db_session.flush()
conv2 = await get_or_create_system_channel(db_session, tenant_id)
await db_session.flush()
assert conv1.id == conv2.id
async def test_system_channel_has_all_users_as_participants(self, db_session: AsyncSession):
"""All tenant users are automatically added as participants."""
seed = await seed_tenant_and_users(db_session)
tenant_id = seed["tenant_a"].id
conv = await get_or_create_system_channel(db_session, tenant_id)
await db_session.flush()
result = await db_session.execute(
select(CommParticipant).where(
CommParticipant.conversation_id == conv.id,
CommParticipant.left_at.is_(None),
)
)
participants = result.scalars().all()
participant_ids = {p.participant_id for p in participants}
# admin_a, viewer_a, editor_a are all in tenant_a
assert seed["admin_a"].id in participant_ids
assert seed["viewer_a"].id in participant_ids
assert seed["editor_a"].id in participant_ids
async def test_system_channel_is_tenant_scoped(self, db_session: AsyncSession):
"""Different tenants get different system channels."""
seed = await seed_tenant_and_users(db_session)
conv_a = await get_or_create_system_channel(db_session, seed["tenant_a"].id)
conv_b = await get_or_create_system_channel(db_session, seed["tenant_b"].id)
await db_session.flush()
assert conv_a.id != conv_b.id
assert conv_a.tenant_id == seed["tenant_a"].id
assert conv_b.tenant_id == seed["tenant_b"].id
@pytest.mark.asyncio
class TestPostSystemMessage:
"""B-NOTIF-EVT: post_system_message creates typed system messages."""
async def test_post_system_message_creates_comm_message(self, db_session: AsyncSession):
"""post_system_message creates a CommMessage in the system channel."""
seed = await seed_tenant_and_users(db_session)
tenant_id = seed["tenant_a"].id
user_id = seed["admin_a"].id
msg = await post_system_message(
db_session, tenant_id, user_id,
message_type="info",
title="Test Notification",
body="This is a test body",
)
await db_session.flush()
assert msg is not None
assert msg.sender_type == "system"
assert "Test Notification" in msg.content
assert "This is a test body" in msg.content
assert msg.metadata_["notification_type"] == "info"
assert msg.metadata_["severity"] == "info"
assert msg.metadata_["target_user_id"] == str(user_id)
async def test_post_system_message_creates_text_block(self, db_session: AsyncSession):
"""post_system_message creates a text block with title and body."""
seed = await seed_tenant_and_users(db_session)
tenant_id = seed["tenant_a"].id
user_id = seed["admin_a"].id
msg = await post_system_message(
db_session, tenant_id, user_id,
message_type="info",
title="Block Test",
body="Block body",
)
await db_session.flush()
result = await db_session.execute(
select(CommMessageBlock).where(
CommMessageBlock.message_id == msg.id,
).order_by(CommMessageBlock.sort_order)
)
blocks = result.scalars().all()
assert len(blocks) >= 1
text_block = blocks[0]
assert text_block.block_type == "text"
assert text_block.block_data["title"] == "Block Test"
assert text_block.block_data["body"] == "Block body"
async def test_post_system_message_with_entity_creates_action_card(self, db_session: AsyncSession):
"""post_system_message with entity_type/entity_id creates an action_card block."""
seed = await seed_tenant_and_users(db_session)
tenant_id = seed["tenant_a"].id
user_id = seed["admin_a"].id
entity_id = seed["company_a"].id
msg = await post_system_message(
db_session, tenant_id, user_id,
message_type="info",
title="Entity Notification",
body="Check this entity",
entity_type="contact",
entity_id=entity_id,
)
await db_session.flush()
result = await db_session.execute(
select(CommMessageBlock).where(
CommMessageBlock.message_id == msg.id,
).order_by(CommMessageBlock.sort_order)
)
blocks = result.scalars().all()
assert len(blocks) == 2
assert blocks[0].block_type == "text"
assert blocks[1].block_type == "action_card"
assert blocks[1].block_data["entity_type"] == "contact"
assert blocks[1].block_data["entity_id"] == str(entity_id)
# Check metadata entity_ref
assert msg.metadata_["entity_ref"]["entity_type"] == "contact"
assert msg.metadata_["entity_ref"]["entity_id"] == str(entity_id)
async def test_post_system_message_severity_in_metadata(self, db_session: AsyncSession):
"""post_system_message stores severity in message metadata."""
seed = await seed_tenant_and_users(db_session)
tenant_id = seed["tenant_a"].id
user_id = seed["admin_a"].id
msg = await post_system_message(
db_session, tenant_id, user_id,
message_type="alert",
title="Critical Alert",
body="Something happened",
severity="critical",
)
await db_session.flush()
assert msg.metadata_["severity"] == "critical"
assert msg.metadata_["notification_type"] == "alert"
@pytest.mark.asyncio
class TestCreateNotificationWrapper:
"""B-NOTIF-EVT: create_notification delegates to post_system_message."""
async def test_create_notification_delegates_to_post_system_message(self, db_session: AsyncSession):
"""create_notification creates both a CommMessage and a legacy Notification."""
seed = await seed_tenant_and_users(db_session)
tenant_id = seed["tenant_a"].id
user_id = seed["admin_a"].id
notif = await create_notification(
db_session, tenant_id, user_id,
type="info",
title="Wrapper Test",
body="Wrapper body",
)
await db_session.flush()
# Legacy notification created
assert notif is not None
assert notif.title == "Wrapper Test"
assert notif.type == "info"
# CommMessage also created in system channel
result = await db_session.execute(
select(CommMessage).where(
CommMessage.tenant_id == tenant_id,
CommMessage.sender_type == "system",
)
)
comm_msgs = result.scalars().all()
assert len(comm_msgs) >= 1
assert any("Wrapper Test" in m.content for m in comm_msgs)
async def test_create_notification_returns_none_when_muted(self, db_session: AsyncSession):
"""create_notification returns None when user has disabled the type."""
seed = await seed_tenant_and_users(db_session)
tenant_id = seed["tenant_a"].id
user_id = seed["admin_a"].id
# Create a notification type and disable it for the user
ntype = NotificationType(
type_key="muted_type",
plugin_name="test",
category="general",
label="Muted Type",
is_enabled_by_default=True,
)
db_session.add(ntype)
await db_session.flush()
pref = NotificationPreference(
tenant_id=tenant_id,
user_id=user_id,
type_key="muted_type",
is_enabled=False,
)
db_session.add(pref)
await db_session.flush()
notif = await create_notification(
db_session, tenant_id, user_id,
type="muted_type",
title="Should Not Appear",
body="Should be muted",
)
await db_session.flush()
assert notif is None
# Verify no CommMessage was created with this title
result = await db_session.execute(
select(CommMessage).where(
CommMessage.tenant_id == tenant_id,
CommMessage.content.like("%Should Not Appear%"),
)
)
comm_msgs = result.scalars().all()
assert len(comm_msgs) == 0
@pytest.mark.asyncio
class TestNotificationPreferences:
"""B-NOTIF-PREF: Delivery preferences are respected."""
async def test_post_system_message_respects_muted_preference(self, db_session: AsyncSession):
"""post_system_message returns None when user has muted the type."""
seed = await seed_tenant_and_users(db_session)
tenant_id = seed["tenant_a"].id
user_id = seed["admin_a"].id
ntype = NotificationType(
type_key="muted_alert",
plugin_name="test",
category="general",
label="Muted Alert",
is_enabled_by_default=True,
)
db_session.add(ntype)
await db_session.flush()
pref = NotificationPreference(
tenant_id=tenant_id,
user_id=user_id,
type_key="muted_alert",
is_enabled=False,
)
db_session.add(pref)
await db_session.flush()
msg = await post_system_message(
db_session, tenant_id, user_id,
message_type="muted_alert",
title="Muted Message",
body="Should not be posted",
)
await db_session.flush()
assert msg is None
async def test_post_system_message_respects_disabled_by_default(self, db_session: AsyncSession):
"""post_system_message returns None when type is disabled by default and no explicit preference."""
seed = await seed_tenant_and_users(db_session)
tenant_id = seed["tenant_a"].id
user_id = seed["admin_a"].id
ntype = NotificationType(
type_key="disabled_default",
plugin_name="test",
category="general",
label="Disabled By Default",
is_enabled_by_default=False,
)
db_session.add(ntype)
await db_session.flush()
msg = await post_system_message(
db_session, tenant_id, user_id,
message_type="disabled_default",
title="Should Not Appear",
body="Disabled by default",
)
await db_session.flush()
assert msg is None
async def test_post_system_message_with_enabled_preference(self, db_session: AsyncSession):
"""post_system_message creates message when preference is enabled."""
seed = await seed_tenant_and_users(db_session)
tenant_id = seed["tenant_a"].id
user_id = seed["admin_a"].id
ntype = NotificationType(
type_key="enabled_type",
plugin_name="test",
category="general",
label="Enabled Type",
is_enabled_by_default=False,
)
db_session.add(ntype)
await db_session.flush()
pref = NotificationPreference(
tenant_id=tenant_id,
user_id=user_id,
type_key="enabled_type",
is_enabled=True,
)
db_session.add(pref)
await db_session.flush()
msg = await post_system_message(
db_session, tenant_id, user_id,
message_type="enabled_type",
title="Enabled Message",
body="Should be posted",
)
await db_session.flush()
assert msg is not None
assert "Enabled Message" in msg.content
@pytest.mark.asyncio
class TestUnreadBadge:
"""B-NOTIF-TEST: Unread badge via CommMessageRead."""
async def test_system_message_unread_for_new_user(self, db_session: AsyncSession):
"""New system messages are unread (no CommMessageRead entry)."""
seed = await seed_tenant_and_users(db_session)
tenant_id = seed["tenant_a"].id
user_id = seed["admin_a"].id
msg = await post_system_message(
db_session, tenant_id, user_id,
message_type="info",
title="Unread Test",
body="Should be unread",
)
await db_session.flush()
# Check no CommMessageRead exists for this user+conversation
result = await db_session.execute(
select(CommMessageRead).where(
CommMessageRead.conversation_id == msg.conversation_id,
CommMessageRead.user_id == user_id,
)
)
read_state = result.scalar_one_or_none()
assert read_state is None # No read state = unread
async def test_mark_system_message_read_creates_read_state(self, db_session: AsyncSession):
"""Marking a system message read creates CommMessageRead entry."""
seed = await seed_tenant_and_users(db_session)
tenant_id = seed["tenant_a"].id
user_id = seed["admin_a"].id
msg = await post_system_message(
db_session, tenant_id, user_id,
message_type="info",
title="Read Test",
body="Will be marked read",
)
await db_session.flush()
# Create read state
read_state = CommMessageRead(
tenant_id=tenant_id,
conversation_id=msg.conversation_id,
user_id=user_id,
last_read_msg_id=msg.id,
last_read_at=datetime.now(UTC),
)
db_session.add(read_state)
await db_session.flush()
# Verify read state exists
result = await db_session.execute(
select(CommMessageRead).where(
CommMessageRead.conversation_id == msg.conversation_id,
CommMessageRead.user_id == user_id,
)
)
saved = result.scalar_one_or_none()
assert saved is not None
assert saved.last_read_msg_id == msg.id
@pytest.mark.asyncio
class TestNotificationMigrationMapping:
"""B-NOTIF-MIG: Migration mapping from Notification to CommMessage."""
async def test_notification_fields_map_to_comm_message(self, db_session: AsyncSession):
"""Notification fields map correctly to CommMessage metadata and blocks."""
seed = await seed_tenant_and_users(db_session)
tenant_id = seed["tenant_a"].id
user_id = seed["admin_a"].id
entity_id = seed["company_a"].id
# Create a notification via the wrapper (which creates both legacy + comm)
notif = await create_notification(
db_session, tenant_id, user_id,
type="mail_received",
title="New Mail",
body="You have a new mail",
entity_type="contact",
entity_id=entity_id,
)
await db_session.flush()
# Find the corresponding CommMessage
result = await db_session.execute(
select(CommMessage).where(
CommMessage.tenant_id == tenant_id,
CommMessage.sender_type == "system",
).order_by(CommMessage.created_at.desc())
)
comm_msg = result.scalars().first()
assert comm_msg is not None
# Field mapping checks
assert comm_msg.metadata_["notification_type"] == "mail_received"
assert comm_msg.metadata_["severity"] == "info"
assert comm_msg.metadata_["entity_ref"]["entity_type"] == "contact"
assert comm_msg.metadata_["entity_ref"]["entity_id"] == str(entity_id)
# Check blocks
blocks_result = await db_session.execute(
select(CommMessageBlock).where(
CommMessageBlock.message_id == comm_msg.id,
).order_by(CommMessageBlock.sort_order)
)
blocks = blocks_result.scalars().all()
assert len(blocks) == 2 # text + action_card
assert blocks[0].block_type == "text"
assert blocks[1].block_type == "action_card"
async def test_notification_read_at_maps_to_comm_message_read(self, db_session: AsyncSession):
"""Notification.read_at maps to CommMessageRead entry."""
seed = await seed_tenant_and_users(db_session)
tenant_id = seed["tenant_a"].id
user_id = seed["admin_a"].id
notif = await create_notification(
db_session, tenant_id, user_id,
type="info",
title="Read Mapping Test",
body="Will be read",
)
await db_session.flush()
# Mark notification as read
notif.read_at = datetime.now(UTC)
await db_session.flush()
# Find the CommMessage
result = await db_session.execute(
select(CommMessage).where(
CommMessage.tenant_id == tenant_id,
CommMessage.sender_type == "system",
).order_by(CommMessage.created_at.desc())
)
comm_msg = result.scalars().first()
assert comm_msg is not None
# Create CommMessageRead (simulating what migration does)
read_state = CommMessageRead(
tenant_id=tenant_id,
conversation_id=comm_msg.conversation_id,
user_id=user_id,
last_read_msg_id=comm_msg.id,
last_read_at=notif.read_at,
)
db_session.add(read_state)
await db_session.flush()
# Verify
read_result = await db_session.execute(
select(CommMessageRead).where(
CommMessageRead.conversation_id == comm_msg.conversation_id,
CommMessageRead.user_id == user_id,
)
)
saved = read_result.scalar_one_or_none()
assert saved is not None
assert saved.last_read_at is not None
@pytest.mark.asyncio
class TestSensitiveFieldsExcluded:
"""B-NOTIF-TEST: Sensitive fields are excluded from system messages."""
async def test_no_password_or_secret_in_message(self, db_session: AsyncSession):
"""System messages do not contain password hashes or secret values."""
seed = await seed_tenant_and_users(db_session)
tenant_id = seed["tenant_a"].id
user_id = seed["admin_a"].id
msg = await post_system_message(
db_session, tenant_id, user_id,
message_type="info",
title="Security Test",
body="Normal body text",
)
await db_session.flush()
# Check message content and metadata for sensitive patterns
sensitive_patterns = ["password", "hash", "secret", "token", "api_key"]
content_lower = msg.content.lower()
metadata_str = str(msg.metadata_).lower()
for pattern in sensitive_patterns:
assert pattern not in content_lower, f"Sensitive pattern '{pattern}' found in content"
assert pattern not in metadata_str, f"Sensitive pattern '{pattern}' found in metadata"
async def test_blocks_exclude_sensitive_data(self, db_session: AsyncSession):
"""Message blocks do not contain sensitive data."""
seed = await seed_tenant_and_users(db_session)
tenant_id = seed["tenant_a"].id
user_id = seed["admin_a"].id
msg = await post_system_message(
db_session, tenant_id, user_id,
message_type="info",
title="Block Security Test",
body="Normal block text",
)
await db_session.flush()
result = await db_session.execute(
select(CommMessageBlock).where(
CommMessageBlock.message_id == msg.id,
)
)
blocks = result.scalars().all()
for block in blocks:
block_str = str(block.block_data).lower()
assert "password" not in block_str
assert "secret" not in block_str
assert "token" not in block_str