mirror of
https://github.com/jokob-sk/NetAlertX.git
synced 2026-03-30 23:03:03 -07:00
Keep all local changes while resolving conflicts
This commit is contained in:
Jokob @NetAlertX
100
docs/NOTIFICATION_TEMPLATES.md
Normal file
100
docs/NOTIFICATION_TEMPLATES.md
Normal file
@@ -0,0 +1,100 @@
|
||||
# Notification Text Templates
|
||||
|
||||
> Customize how devices and events appear in **text** notifications (email previews, push notifications, Apprise messages).
|
||||
|
||||
By default, NetAlertX formats each device as a vertical list of `Header: Value` pairs. Text templates let you define a **single-line format per device** using `{FieldName}` placeholders — ideal for mobile notification previews and high-volume alerts.
|
||||
|
||||
HTML email tables are **not affected** by these templates.
|
||||
|
||||
## Quick Start
|
||||
|
||||
1. Go to **Settings → Notification Processing**.
|
||||
2. Set a template string for the section you want to customize, e.g.:
|
||||
- **Text Template: New Devices** → `{devName} ({eve_MAC}) - {eve_IP}`
|
||||
3. Save. The next notification will use your format.
|
||||
|
||||
**Before (default):**
|
||||
```
|
||||
🆕 New devices
|
||||
---------
|
||||
devName: MyPhone
|
||||
eve_MAC: aa:bb:cc:dd:ee:ff
|
||||
devVendor: Apple
|
||||
eve_IP: 192.168.1.42
|
||||
eve_DateTime: 2025-01-15 10:30:00
|
||||
eve_EventType: New Device
|
||||
devComments:
|
||||
```
|
||||
|
||||
**After (with template `{devName} ({eve_MAC}) - {eve_IP}`):**
|
||||
```
|
||||
🆕 New devices
|
||||
---------
|
||||
MyPhone (aa:bb:cc:dd:ee:ff) - 192.168.1.42
|
||||
```
|
||||
|
||||
## Settings Reference
|
||||
|
||||
| Setting | Type | Default | Description |
|
||||
|---------|------|---------|-------------|
|
||||
| `NTFPRCS_TEXT_SECTION_HEADERS` | Boolean | `true` | Show/hide section titles (e.g. `🆕 New devices \n---------`). |
|
||||
| `NTFPRCS_TEXT_TEMPLATE_new_devices` | String | *(empty)* | Template for new device rows. |
|
||||
| `NTFPRCS_TEXT_TEMPLATE_down_devices` | String | *(empty)* | Template for down device rows. |
|
||||
| `NTFPRCS_TEXT_TEMPLATE_down_reconnected` | String | *(empty)* | Template for reconnected device rows. |
|
||||
| `NTFPRCS_TEXT_TEMPLATE_events` | String | *(empty)* | Template for event rows. |
|
||||
| `NTFPRCS_TEXT_TEMPLATE_plugins` | String | *(empty)* | Template for plugin event rows. |
|
||||
|
||||
When a template is **empty**, the section uses the original vertical `Header: Value` format (full backward compatibility).
|
||||
|
||||
## Template Syntax
|
||||
|
||||
Use `{FieldName}` to insert a value from the notification data. Field names are **case-sensitive** and must match the column names exactly.
|
||||
|
||||
```
|
||||
{devName} ({eve_MAC}) connected at {eve_DateTime}
|
||||
```
|
||||
|
||||
- No loops, conditionals, or nesting — just simple string replacement.
|
||||
- If a `{FieldName}` does not exist in the data, it is left as-is in the output (safe failure). For example, `{NonExistent}` renders literally as `{NonExistent}`.
|
||||
|
||||
## Variable Availability by Section
|
||||
|
||||
All four device sections (`new_devices`, `down_devices`, `down_reconnected`, `events`) share the same unified field names.
|
||||
|
||||
### `new_devices`, `down_devices`, `down_reconnected`, and `events`
|
||||
|
||||
| Variable | Description |
|
||||
|----------|-------------|
|
||||
| `{devName}` | Device display name |
|
||||
| `{eve_MAC}` | Device MAC address |
|
||||
| `{devVendor}` | Device vendor/manufacturer |
|
||||
| `{eve_IP}` | Device IP address |
|
||||
| `{eve_DateTime}` | Event timestamp |
|
||||
| `{eve_EventType}` | Type of event (e.g. `New Device`, `Connected`, `Device Down`) |
|
||||
| `{devComments}` | Device comments |
|
||||
|
||||
**Example (new_devices/events):** `{devName} ({eve_MAC}) - {eve_IP} [{eve_EventType}]`
|
||||
|
||||
**Example (down_devices):** `{devName} ({eve_MAC}) {devVendor} - went down at {eve_DateTime}`
|
||||
|
||||
**Example (down_reconnected):** `{devName} ({eve_MAC}) reconnected at {eve_DateTime}`
|
||||
|
||||
### `plugins`
|
||||
|
||||
| Variable | Description |
|
||||
|----------|-------------|
|
||||
| `{Plugin}` | Plugin code name |
|
||||
| `{Object_PrimaryId}` | Primary identifier of the object |
|
||||
| `{Object_SecondaryId}` | Secondary identifier |
|
||||
| `{DateTimeChanged}` | Timestamp of change |
|
||||
| `{Watched_Value1}` | First watched value |
|
||||
| `{Watched_Value2}` | Second watched value |
|
||||
| `{Watched_Value3}` | Third watched value |
|
||||
| `{Watched_Value4}` | Fourth watched value |
|
||||
| `{Status}` | Plugin event status |
|
||||
|
||||
**Example:** `{Plugin}: {Object_PrimaryId} - {Status}`
|
||||
|
||||
## Section Headers Toggle
|
||||
|
||||
Set **Text Section Headers** (`NTFPRCS_TEXT_SECTION_HEADERS`) to `false` to remove the section title separators from text notifications. This is useful when you want compact output without the `🆕 New devices \n---------` banners.
|
||||
@@ -152,6 +152,153 @@
|
||||
"string": "You can specify a SQL where condition to filter out Events from notifications. For example <code>AND devLastIP NOT LIKE '192.168.3.%'</code> will always exclude any Event notifications for all devices with the IP starting with <code>192.168.3.%</code>."
|
||||
}
|
||||
]
|
||||
<<<<<<< Updated upstream
|
||||
=======
|
||||
},
|
||||
{
|
||||
"function": "TEXT_SECTION_HEADERS",
|
||||
"type": {
|
||||
"dataType": "boolean",
|
||||
"elements": [
|
||||
{ "elementType": "input", "elementOptions": [{ "type": "checkbox" }], "transformers": [] }
|
||||
]
|
||||
},
|
||||
"default_value": true,
|
||||
"options": [],
|
||||
"localized": ["name", "description"],
|
||||
"name": [
|
||||
{
|
||||
"language_code": "en_us",
|
||||
"string": "Text Section Headers"
|
||||
}
|
||||
],
|
||||
"description": [
|
||||
{
|
||||
"language_code": "en_us",
|
||||
"string": "Enable or disable section titles (e.g. <code>🆕 New devices \\n---------</code>) in text notifications. Enabled by default for backward compatibility."
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"function": "TEXT_TEMPLATE_new_devices",
|
||||
"type": {
|
||||
"dataType": "string",
|
||||
"elements": [
|
||||
{ "elementType": "input", "elementOptions": [], "transformers": [] }
|
||||
]
|
||||
},
|
||||
"default_value": "",
|
||||
"options": [],
|
||||
"localized": ["name", "description"],
|
||||
"name": [
|
||||
{
|
||||
"language_code": "en_us",
|
||||
"string": "Text Template: New Devices"
|
||||
}
|
||||
],
|
||||
"description": [
|
||||
{
|
||||
"language_code": "en_us",
|
||||
"string": "Custom text template for new device notifications. Use <code>{FieldName}</code> placeholders, e.g. <code>{devName} ({eve_MAC}) - {eve_IP}</code>. Leave empty for default formatting. Available fields: <code>{devName}</code>, <code>{eve_MAC}</code>, <code>{devVendor}</code>, <code>{eve_IP}</code>, <code>{eve_DateTime}</code>, <code>{eve_EventType}</code>, <code>{devComments}</code>."
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"function": "TEXT_TEMPLATE_down_devices",
|
||||
"type": {
|
||||
"dataType": "string",
|
||||
"elements": [
|
||||
{ "elementType": "input", "elementOptions": [], "transformers": [] }
|
||||
]
|
||||
},
|
||||
"default_value": "",
|
||||
"options": [],
|
||||
"localized": ["name", "description"],
|
||||
"name": [
|
||||
{
|
||||
"language_code": "en_us",
|
||||
"string": "Text Template: Down Devices"
|
||||
}
|
||||
],
|
||||
"description": [
|
||||
{
|
||||
"language_code": "en_us",
|
||||
"string": "Custom text template for down device notifications. Use <code>{FieldName}</code> placeholders, e.g. <code>{devName} ({eve_MAC}) - {eve_IP}</code>. Leave empty for default formatting. Available fields: <code>{devName}</code>, <code>{eve_MAC}</code>, <code>{devVendor}</code>, <code>{eve_IP}</code>, <code>{eve_DateTime}</code>, <code>{eve_EventType}</code>, <code>{devComments}</code>."
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"function": "TEXT_TEMPLATE_down_reconnected",
|
||||
"type": {
|
||||
"dataType": "string",
|
||||
"elements": [
|
||||
{ "elementType": "input", "elementOptions": [], "transformers": [] }
|
||||
]
|
||||
},
|
||||
"default_value": "",
|
||||
"options": [],
|
||||
"localized": ["name", "description"],
|
||||
"name": [
|
||||
{
|
||||
"language_code": "en_us",
|
||||
"string": "Text Template: Reconnected"
|
||||
}
|
||||
],
|
||||
"description": [
|
||||
{
|
||||
"language_code": "en_us",
|
||||
"string": "Custom text template for reconnected device notifications. Use <code>{FieldName}</code> placeholders, e.g. <code>{devName} ({eve_MAC}) reconnected at {eve_DateTime}</code>. Leave empty for default formatting. Available fields: <code>{devName}</code>, <code>{eve_MAC}</code>, <code>{devVendor}</code>, <code>{eve_IP}</code>, <code>{eve_DateTime}</code>, <code>{eve_EventType}</code>, <code>{devComments}</code>."
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"function": "TEXT_TEMPLATE_events",
|
||||
"type": {
|
||||
"dataType": "string",
|
||||
"elements": [
|
||||
{ "elementType": "input", "elementOptions": [], "transformers": [] }
|
||||
]
|
||||
},
|
||||
"default_value": "",
|
||||
"options": [],
|
||||
"localized": ["name", "description"],
|
||||
"name": [
|
||||
{
|
||||
"language_code": "en_us",
|
||||
"string": "Text Template: Events"
|
||||
}
|
||||
],
|
||||
"description": [
|
||||
{
|
||||
"language_code": "en_us",
|
||||
"string": "Custom text template for event notifications. Use <code>{FieldName}</code> placeholders, e.g. <code>{devName} ({eve_MAC}) {eve_EventType} at {eve_DateTime}</code>. Leave empty for default formatting. Available fields: <code>{devName}</code>, <code>{eve_MAC}</code>, <code>{devVendor}</code>, <code>{eve_IP}</code>, <code>{eve_DateTime}</code>, <code>{eve_EventType}</code>, <code>{devComments}</code>."
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"function": "TEXT_TEMPLATE_plugins",
|
||||
"type": {
|
||||
"dataType": "string",
|
||||
"elements": [
|
||||
{ "elementType": "input", "elementOptions": [], "transformers": [] }
|
||||
]
|
||||
},
|
||||
"default_value": "",
|
||||
"options": [],
|
||||
"localized": ["name", "description"],
|
||||
"name": [
|
||||
{
|
||||
"language_code": "en_us",
|
||||
"string": "Text Template: Plugins"
|
||||
}
|
||||
],
|
||||
"description": [
|
||||
{
|
||||
"language_code": "en_us",
|
||||
"string": "Custom text template for plugin event notifications. Use <code>{FieldName}</code> placeholders, e.g. <code>{Plugin}: {Object_PrimaryId} - {Status}</code>. Leave empty for default formatting. Available fields: <code>{Plugin}</code>, <code>{Object_PrimaryId}</code>, <code>{Object_SecondaryId}</code>, <code>{DateTimeChanged}</code>, <code>{Watched_Value1}</code>, <code>{Watched_Value2}</code>, <code>{Watched_Value3}</code>, <code>{Watched_Value4}</code>, <code>{Status}</code>."
|
||||
}
|
||||
]
|
||||
>>>>>>> Stashed changes
|
||||
}
|
||||
],
|
||||
|
||||
|
||||
133
server/messaging/notification_sections.py
Normal file
133
server/messaging/notification_sections.py
Normal file
@@ -0,0 +1,133 @@
|
||||
# -------------------------------------------------------------------------------
|
||||
# notification_sections.py — Single source of truth for notification section
|
||||
# metadata: titles, SQL templates, datetime fields, and section ordering.
|
||||
#
|
||||
# Both reporting.py and notification_instance.py import from here.
|
||||
# -------------------------------------------------------------------------------
|
||||
|
||||
# Canonical processing order
|
||||
SECTION_ORDER = [
|
||||
"new_devices",
|
||||
"down_devices",
|
||||
"down_reconnected",
|
||||
"events",
|
||||
"plugins",
|
||||
]
|
||||
|
||||
# Section display titles (used in text + HTML notifications)
|
||||
SECTION_TITLES = {
|
||||
"new_devices": "🆕 New devices",
|
||||
"down_devices": "🔴 Down devices",
|
||||
"down_reconnected": "🔁 Reconnected down devices",
|
||||
"events": "⚡ Events",
|
||||
"plugins": "🔌 Plugins",
|
||||
}
|
||||
|
||||
# Which column(s) contain datetime values per section (for timezone conversion)
|
||||
DATETIME_FIELDS = {
|
||||
"new_devices": ["eve_DateTime"],
|
||||
"down_devices": ["eve_DateTime"],
|
||||
"down_reconnected": ["eve_DateTime"],
|
||||
"events": ["eve_DateTime"],
|
||||
"plugins": ["DateTimeChanged"],
|
||||
}
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# SQL templates
|
||||
#
|
||||
# All device sections use unified DB column names so the JSON output
|
||||
# has consistent field names across new_devices, down_devices,
|
||||
# down_reconnected, and events.
|
||||
#
|
||||
# Placeholders:
|
||||
# {condition} — optional WHERE clause appended by condition builder
|
||||
# {alert_down_minutes} — runtime value, only used by down_devices
|
||||
# ---------------------------------------------------------------------------
|
||||
SQL_TEMPLATES = {
|
||||
"new_devices": """
|
||||
SELECT
|
||||
devName,
|
||||
eve_MAC,
|
||||
devVendor,
|
||||
devLastIP as eve_IP,
|
||||
eve_DateTime,
|
||||
eve_EventType,
|
||||
devComments
|
||||
FROM Events_Devices
|
||||
WHERE eve_PendingAlertEmail = 1
|
||||
AND eve_EventType = 'New Device' {condition}
|
||||
ORDER BY eve_DateTime
|
||||
""",
|
||||
"down_devices": """
|
||||
SELECT
|
||||
devName,
|
||||
eve_MAC,
|
||||
devVendor,
|
||||
eve_IP,
|
||||
eve_DateTime,
|
||||
eve_EventType,
|
||||
devComments
|
||||
FROM Events_Devices AS down_events
|
||||
WHERE eve_PendingAlertEmail = 1
|
||||
AND down_events.eve_EventType = 'Device Down'
|
||||
AND eve_DateTime < datetime('now', '-{alert_down_minutes} minutes')
|
||||
AND NOT EXISTS (
|
||||
SELECT 1
|
||||
FROM Events AS connected_events
|
||||
WHERE connected_events.eve_MAC = down_events.eve_MAC
|
||||
AND connected_events.eve_EventType = 'Connected'
|
||||
AND connected_events.eve_DateTime > down_events.eve_DateTime
|
||||
)
|
||||
ORDER BY down_events.eve_DateTime
|
||||
""",
|
||||
"down_reconnected": """
|
||||
SELECT
|
||||
devName,
|
||||
eve_MAC,
|
||||
devVendor,
|
||||
eve_IP,
|
||||
eve_DateTime,
|
||||
eve_EventType,
|
||||
devComments
|
||||
FROM Events_Devices AS reconnected_devices
|
||||
WHERE reconnected_devices.eve_EventType = 'Down Reconnected'
|
||||
AND reconnected_devices.eve_PendingAlertEmail = 1
|
||||
ORDER BY reconnected_devices.eve_DateTime
|
||||
""",
|
||||
"events": """
|
||||
SELECT
|
||||
devName,
|
||||
eve_MAC,
|
||||
devVendor,
|
||||
devLastIP as eve_IP,
|
||||
eve_DateTime,
|
||||
eve_EventType,
|
||||
devComments
|
||||
FROM Events_Devices
|
||||
WHERE eve_PendingAlertEmail = 1
|
||||
AND eve_EventType IN ('Connected', 'Down Reconnected', 'Disconnected','IP Changed') {condition}
|
||||
ORDER BY eve_DateTime
|
||||
""",
|
||||
"plugins": """
|
||||
SELECT
|
||||
Plugin,
|
||||
Object_PrimaryId,
|
||||
Object_SecondaryId,
|
||||
DateTimeChanged,
|
||||
Watched_Value1,
|
||||
Watched_Value2,
|
||||
Watched_Value3,
|
||||
Watched_Value4,
|
||||
Status
|
||||
FROM Plugins_Events
|
||||
""",
|
||||
}
|
||||
|
||||
# Sections that support user-defined condition filters
|
||||
SECTIONS_WITH_CONDITIONS = {"new_devices", "events"}
|
||||
|
||||
# Legacy setting key mapping for condition filters
|
||||
SECTION_CONDITION_MAP = {
|
||||
"new_devices": "NTFPRCS_new_dev_condition",
|
||||
"events": "NTFPRCS_event_condition",
|
||||
}
|
||||
@@ -25,20 +25,20 @@ from helper import ( # noqa: E402 [flake8 lint suppression]
|
||||
from logger import mylog # noqa: E402 [flake8 lint suppression]
|
||||
from db.sql_safe_builder import create_safe_condition_builder # noqa: E402 [flake8 lint suppression]
|
||||
from utils.datetime_utils import format_date_iso # noqa: E402 [flake8 lint suppression]
|
||||
from messaging.notification_sections import ( # noqa: E402 [flake8 lint suppression]
|
||||
SECTION_ORDER,
|
||||
SECTION_TITLES,
|
||||
DATETIME_FIELDS,
|
||||
SQL_TEMPLATES,
|
||||
SECTIONS_WITH_CONDITIONS,
|
||||
SECTION_CONDITION_MAP,
|
||||
)
|
||||
import conf # noqa: E402 [flake8 lint suppression]
|
||||
|
||||
# ===============================================================================
|
||||
# Timezone conversion
|
||||
# ===============================================================================
|
||||
|
||||
DATETIME_FIELDS = {
|
||||
"new_devices": ["Datetime"],
|
||||
"down_devices": ["eve_DateTime"],
|
||||
"down_reconnected": ["eve_DateTime"],
|
||||
"events": ["Datetime"],
|
||||
"plugins": ["DateTimeChanged"],
|
||||
}
|
||||
|
||||
|
||||
def get_datetime_fields_from_columns(column_names):
|
||||
return [
|
||||
@@ -155,6 +155,7 @@ def get_notifications(db):
|
||||
|
||||
return ""
|
||||
|
||||
<<<<<<< Updated upstream
|
||||
# -------------------------
|
||||
# SQL templates
|
||||
# -------------------------
|
||||
@@ -245,13 +246,17 @@ def get_notifications(db):
|
||||
|
||||
# Sections that support dynamic conditions
|
||||
sections_with_conditions = {"new_devices", "events"}
|
||||
=======
|
||||
# SQL templates with placeholders for runtime values
|
||||
# {condition} and {alert_down_minutes} are formatted at query time
|
||||
>>>>>>> Stashed changes
|
||||
|
||||
# Initialize final structure
|
||||
final_json = {}
|
||||
for section in ["new_devices", "down_devices", "down_reconnected", "events", "plugins"]:
|
||||
for section in SECTION_ORDER:
|
||||
final_json[section] = []
|
||||
final_json[f"{section}_meta"] = {
|
||||
"title": section_titles.get(section, section),
|
||||
"title": SECTION_TITLES.get(section, section),
|
||||
"columnNames": []
|
||||
}
|
||||
|
||||
@@ -260,17 +265,8 @@ def get_notifications(db):
|
||||
# -------------------------
|
||||
# Main loop
|
||||
# -------------------------
|
||||
condition_builder = create_safe_condition_builder()
|
||||
|
||||
SECTION_CONDITION_MAP = {
|
||||
"new_devices": "NTFPRCS_new_dev_condition",
|
||||
"events": "NTFPRCS_event_condition",
|
||||
}
|
||||
|
||||
sections_with_conditions = set(SECTION_CONDITION_MAP.keys())
|
||||
|
||||
for section in sections:
|
||||
template = sql_templates.get(section)
|
||||
template = SQL_TEMPLATES.get(section)
|
||||
|
||||
if not template:
|
||||
mylog("verbose", ["[Notification] Unknown section: ", section])
|
||||
@@ -280,7 +276,7 @@ def get_notifications(db):
|
||||
parameters = {}
|
||||
|
||||
try:
|
||||
if section in sections_with_conditions:
|
||||
if section in SECTIONS_WITH_CONDITIONS:
|
||||
condition_key = SECTION_CONDITION_MAP.get(section)
|
||||
condition_setting = get_setting_value(condition_key)
|
||||
|
||||
@@ -289,11 +285,18 @@ def get_notifications(db):
|
||||
condition_setting
|
||||
)
|
||||
|
||||
sqlQuery = template.format(condition=safe_condition)
|
||||
# Format template with runtime placeholders
|
||||
format_vars = {"condition": safe_condition}
|
||||
if section == "down_devices":
|
||||
format_vars["alert_down_minutes"] = alert_down_minutes
|
||||
sqlQuery = template.format(**format_vars)
|
||||
|
||||
except Exception as e:
|
||||
mylog("verbose", [f"[Notification] Error building condition for {section}: ", e])
|
||||
sqlQuery = template.format(condition="")
|
||||
fallback_vars = {"condition": ""}
|
||||
if section == "down_devices":
|
||||
fallback_vars["alert_down_minutes"] = alert_down_minutes
|
||||
sqlQuery = template.format(**fallback_vars)
|
||||
parameters = {}
|
||||
|
||||
mylog("debug", [f"[Notification] {section} SQL query: ", sqlQuery])
|
||||
@@ -307,7 +310,7 @@ def get_notifications(db):
|
||||
|
||||
final_json[section] = json_obj.json.get("data", [])
|
||||
final_json[f"{section}_meta"] = {
|
||||
"title": section_titles.get(section, section),
|
||||
"title": SECTION_TITLES.get(section, section),
|
||||
"columnNames": getattr(json_obj, "columnNames", [])
|
||||
}
|
||||
|
||||
|
||||
@@ -16,6 +16,7 @@ from helper import (
|
||||
getBuildTimeStampAndVersion,
|
||||
)
|
||||
from messaging.in_app import write_notification
|
||||
from messaging.notification_sections import SECTION_ORDER
|
||||
from utils.datetime_utils import timeNowUTC, timeNowTZ, get_timezone_offset
|
||||
|
||||
|
||||
@@ -60,12 +61,7 @@ class NotificationInstance:
|
||||
write_file(logPath + "/report_output.json", json.dumps(JSON))
|
||||
|
||||
# Check if nothing to report, end
|
||||
if (
|
||||
JSON["new_devices"] == [] and JSON["down_devices"] == [] and JSON["events"] == [] and JSON["plugins"] == [] and JSON["down_reconnected"] == []
|
||||
):
|
||||
self.HasNotifications = False
|
||||
else:
|
||||
self.HasNotifications = True
|
||||
self.HasNotifications = any(JSON.get(s, []) for s in SECTION_ORDER)
|
||||
|
||||
self.GUID = str(uuid.uuid4())
|
||||
self.DateTimeCreated = timeNowUTC()
|
||||
@@ -129,47 +125,13 @@ class NotificationInstance:
|
||||
mail_text = mail_text.replace("REPORT_DASHBOARD_URL", self.serverUrl)
|
||||
mail_html = mail_html.replace("REPORT_DASHBOARD_URL", self.serverUrl)
|
||||
|
||||
# Start generating the TEXT & HTML notification messages
|
||||
# new_devices
|
||||
# ---
|
||||
html, text = construct_notifications(self.JSON, "new_devices")
|
||||
|
||||
mail_text = mail_text.replace("NEW_DEVICES_TABLE", text + "\n")
|
||||
mail_html = mail_html.replace("NEW_DEVICES_TABLE", html)
|
||||
mylog("verbose", ["[Notification] New Devices sections done."])
|
||||
|
||||
# down_devices
|
||||
# ---
|
||||
html, text = construct_notifications(self.JSON, "down_devices")
|
||||
|
||||
mail_text = mail_text.replace("DOWN_DEVICES_TABLE", text + "\n")
|
||||
mail_html = mail_html.replace("DOWN_DEVICES_TABLE", html)
|
||||
mylog("verbose", ["[Notification] Down Devices sections done."])
|
||||
|
||||
# down_reconnected
|
||||
# ---
|
||||
html, text = construct_notifications(self.JSON, "down_reconnected")
|
||||
|
||||
mail_text = mail_text.replace("DOWN_RECONNECTED_TABLE", text + "\n")
|
||||
mail_html = mail_html.replace("DOWN_RECONNECTED_TABLE", html)
|
||||
mylog("verbose", ["[Notification] Reconnected Down Devices sections done."])
|
||||
|
||||
# events
|
||||
# ---
|
||||
html, text = construct_notifications(self.JSON, "events")
|
||||
|
||||
mail_text = mail_text.replace("EVENTS_TABLE", text + "\n")
|
||||
mail_html = mail_html.replace("EVENTS_TABLE", html)
|
||||
mylog("verbose", ["[Notification] Events sections done."])
|
||||
|
||||
# plugins
|
||||
# ---
|
||||
html, text = construct_notifications(self.JSON, "plugins")
|
||||
|
||||
mail_text = mail_text.replace("PLUGINS_TABLE", text + "\n")
|
||||
mail_html = mail_html.replace("PLUGINS_TABLE", html)
|
||||
|
||||
mylog("verbose", ["[Notification] Plugins sections done."])
|
||||
# Generate TEXT & HTML for each notification section
|
||||
for section in SECTION_ORDER:
|
||||
html, text = construct_notifications(self.JSON, section)
|
||||
placeholder = f"{section.upper()}_TABLE"
|
||||
mail_text = mail_text.replace(placeholder, text + "\n")
|
||||
mail_html = mail_html.replace(placeholder, html)
|
||||
mylog("verbose", [f"[Notification] {section} section done."])
|
||||
|
||||
final_text = removeDuplicateNewLines(mail_text)
|
||||
|
||||
|
||||
299
test/backend/test_notification_templates.py
Normal file
299
test/backend/test_notification_templates.py
Normal file
@@ -0,0 +1,299 @@
|
||||
"""
|
||||
NetAlertX Notification Text Template Tests
|
||||
|
||||
Tests the template substitution and section header toggle in
|
||||
construct_notifications(). All tests mock get_setting_value to avoid
|
||||
database/config dependencies.
|
||||
|
||||
License: GNU GPLv3
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
# Add the server directory to the path for imports
|
||||
INSTALL_PATH = os.getenv("NETALERTX_APP", "/app")
|
||||
sys.path.extend([f"{INSTALL_PATH}/server"])
|
||||
|
||||
|
||||
def _make_json(section, devices, column_names, title="Test Section"):
|
||||
"""Helper to build the JSON structure expected by construct_notifications."""
|
||||
return {
|
||||
section: devices,
|
||||
f"{section}_meta": {
|
||||
"title": title,
|
||||
"columnNames": column_names,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
SAMPLE_NEW_DEVICES = [
|
||||
{
|
||||
"devName": "MyPhone",
|
||||
"eve_MAC": "aa:bb:cc:dd:ee:ff",
|
||||
"devVendor": "",
|
||||
"eve_IP": "192.168.1.42",
|
||||
"eve_DateTime": "2025-01-15 10:30:00",
|
||||
"eve_EventType": "New Device",
|
||||
"devComments": "",
|
||||
},
|
||||
{
|
||||
"devName": "Laptop",
|
||||
"eve_MAC": "11:22:33:44:55:66",
|
||||
"devVendor": "Dell",
|
||||
"eve_IP": "192.168.1.99",
|
||||
"eve_DateTime": "2025-01-15 11:00:00",
|
||||
"eve_EventType": "New Device",
|
||||
"devComments": "Office",
|
||||
},
|
||||
]
|
||||
|
||||
NEW_DEVICE_COLUMNS = ["devName", "eve_MAC", "devVendor", "eve_IP", "eve_DateTime", "eve_EventType", "devComments"]
|
||||
|
||||
|
||||
class TestConstructNotificationsTemplates(unittest.TestCase):
|
||||
"""Tests for template substitution in construct_notifications."""
|
||||
|
||||
def _setting_factory(self, overrides=None):
|
||||
"""Return a mock get_setting_value that resolves from overrides dict."""
|
||||
settings = overrides or {}
|
||||
|
||||
def mock_get(key):
|
||||
return settings.get(key, "")
|
||||
|
||||
return mock_get
|
||||
|
||||
# -----------------------------------------------------------------
|
||||
# Empty section should always return ("", "") regardless of settings
|
||||
# -----------------------------------------------------------------
|
||||
@patch("models.notification_instance.get_setting_value")
|
||||
def test_empty_section_returns_empty(self, mock_setting):
|
||||
from models.notification_instance import construct_notifications
|
||||
|
||||
mock_setting.return_value = ""
|
||||
json_data = _make_json("new_devices", [], [])
|
||||
html, text = construct_notifications(json_data, "new_devices")
|
||||
self.assertEqual(html, "")
|
||||
self.assertEqual(text, "")
|
||||
|
||||
# -----------------------------------------------------------------
|
||||
# Legacy fallback: no template → vertical Header: Value per device
|
||||
# -----------------------------------------------------------------
|
||||
@patch("models.notification_instance.get_setting_value")
|
||||
def test_legacy_fallback_no_template(self, mock_setting):
|
||||
from models.notification_instance import construct_notifications
|
||||
|
||||
mock_setting.side_effect = self._setting_factory({
|
||||
"NTFPRCS_TEXT_SECTION_HEADERS": True,
|
||||
"NTFPRCS_TEXT_TEMPLATE_new_devices": "",
|
||||
})
|
||||
|
||||
json_data = _make_json(
|
||||
"new_devices", SAMPLE_NEW_DEVICES, NEW_DEVICE_COLUMNS, "🆕 New devices"
|
||||
)
|
||||
html, text = construct_notifications(json_data, "new_devices")
|
||||
|
||||
# Section header must be present
|
||||
self.assertIn("🆕 New devices", text)
|
||||
self.assertIn("---------", text)
|
||||
|
||||
# Legacy format: each header appears as "Header: \tValue"
|
||||
self.assertIn("eve_MAC:", text)
|
||||
self.assertIn("aa:bb:cc:dd:ee:ff", text)
|
||||
self.assertIn("devName:", text)
|
||||
self.assertIn("MyPhone", text)
|
||||
|
||||
# HTML must still be generated
|
||||
self.assertNotEqual(html, "")
|
||||
|
||||
# -----------------------------------------------------------------
|
||||
# Template substitution: single-line format per device
|
||||
# -----------------------------------------------------------------
|
||||
@patch("models.notification_instance.get_setting_value")
|
||||
def test_template_substitution(self, mock_setting):
|
||||
from models.notification_instance import construct_notifications
|
||||
|
||||
mock_setting.side_effect = self._setting_factory({
|
||||
"NTFPRCS_TEXT_SECTION_HEADERS": True,
|
||||
"NTFPRCS_TEXT_TEMPLATE_new_devices": "{devName} ({eve_MAC}) - {eve_IP}",
|
||||
})
|
||||
|
||||
json_data = _make_json(
|
||||
"new_devices", SAMPLE_NEW_DEVICES, NEW_DEVICE_COLUMNS, "🆕 New devices"
|
||||
)
|
||||
_, text = construct_notifications(json_data, "new_devices")
|
||||
|
||||
self.assertIn("MyPhone (aa:bb:cc:dd:ee:ff) - 192.168.1.42", text)
|
||||
self.assertIn("Laptop (11:22:33:44:55:66) - 192.168.1.99", text)
|
||||
|
||||
# -----------------------------------------------------------------
|
||||
# Missing field: {NonExistent} left as-is (safe failure)
|
||||
# -----------------------------------------------------------------
|
||||
@patch("models.notification_instance.get_setting_value")
|
||||
def test_missing_field_safe_failure(self, mock_setting):
|
||||
from models.notification_instance import construct_notifications
|
||||
|
||||
mock_setting.side_effect = self._setting_factory({
|
||||
"NTFPRCS_TEXT_SECTION_HEADERS": True,
|
||||
"NTFPRCS_TEXT_TEMPLATE_new_devices": "{devName} - {NonExistent}",
|
||||
})
|
||||
|
||||
json_data = _make_json(
|
||||
"new_devices", SAMPLE_NEW_DEVICES, NEW_DEVICE_COLUMNS, "🆕 New devices"
|
||||
)
|
||||
_, text = construct_notifications(json_data, "new_devices")
|
||||
|
||||
self.assertIn("MyPhone - {NonExistent}", text)
|
||||
self.assertIn("Laptop - {NonExistent}", text)
|
||||
|
||||
# -----------------------------------------------------------------
|
||||
# Section headers disabled: no title/separator in text output
|
||||
# -----------------------------------------------------------------
|
||||
@patch("models.notification_instance.get_setting_value")
|
||||
def test_section_headers_disabled(self, mock_setting):
|
||||
from models.notification_instance import construct_notifications
|
||||
|
||||
mock_setting.side_effect = self._setting_factory({
|
||||
"NTFPRCS_TEXT_SECTION_HEADERS": False,
|
||||
"NTFPRCS_TEXT_TEMPLATE_new_devices": "{devName} ({eve_MAC})",
|
||||
})
|
||||
|
||||
json_data = _make_json(
|
||||
"new_devices", SAMPLE_NEW_DEVICES, NEW_DEVICE_COLUMNS, "🆕 New devices"
|
||||
)
|
||||
_, text = construct_notifications(json_data, "new_devices")
|
||||
|
||||
self.assertNotIn("🆕 New devices", text)
|
||||
self.assertNotIn("---------", text)
|
||||
# Template output still present
|
||||
self.assertIn("MyPhone (aa:bb:cc:dd:ee:ff)", text)
|
||||
|
||||
# -----------------------------------------------------------------
|
||||
# Section headers enabled (default when setting absent/empty)
|
||||
# -----------------------------------------------------------------
|
||||
@patch("models.notification_instance.get_setting_value")
|
||||
def test_section_headers_default_enabled(self, mock_setting):
|
||||
from models.notification_instance import construct_notifications
|
||||
|
||||
# Simulate setting not configured (returns empty string)
|
||||
mock_setting.side_effect = self._setting_factory({})
|
||||
|
||||
json_data = _make_json(
|
||||
"new_devices", SAMPLE_NEW_DEVICES, NEW_DEVICE_COLUMNS, "🆕 New devices"
|
||||
)
|
||||
_, text = construct_notifications(json_data, "new_devices")
|
||||
|
||||
# Headers should be shown by default
|
||||
self.assertIn("🆕 New devices", text)
|
||||
self.assertIn("---------", text)
|
||||
|
||||
# -----------------------------------------------------------------
|
||||
# Mixed valid and invalid fields in same template
|
||||
# -----------------------------------------------------------------
|
||||
@patch("models.notification_instance.get_setting_value")
|
||||
def test_mixed_valid_and_invalid_fields(self, mock_setting):
|
||||
from models.notification_instance import construct_notifications
|
||||
|
||||
mock_setting.side_effect = self._setting_factory({
|
||||
"NTFPRCS_TEXT_SECTION_HEADERS": True,
|
||||
"NTFPRCS_TEXT_TEMPLATE_new_devices": "{devName} ({BadField}) - {eve_IP}",
|
||||
})
|
||||
|
||||
json_data = _make_json(
|
||||
"new_devices", SAMPLE_NEW_DEVICES, NEW_DEVICE_COLUMNS, "🆕 New devices"
|
||||
)
|
||||
_, text = construct_notifications(json_data, "new_devices")
|
||||
|
||||
self.assertIn("MyPhone ({BadField}) - 192.168.1.42", text)
|
||||
|
||||
# -----------------------------------------------------------------
|
||||
# Down devices section uses same column names as all other sections
|
||||
# -----------------------------------------------------------------
|
||||
@patch("models.notification_instance.get_setting_value")
|
||||
def test_down_devices_template(self, mock_setting):
|
||||
from models.notification_instance import construct_notifications
|
||||
|
||||
mock_setting.side_effect = self._setting_factory({
|
||||
"NTFPRCS_TEXT_SECTION_HEADERS": True,
|
||||
"NTFPRCS_TEXT_TEMPLATE_down_devices": "{devName} ({eve_MAC}) down since {eve_DateTime}",
|
||||
})
|
||||
|
||||
down_devices = [
|
||||
{
|
||||
"devName": "Router",
|
||||
"eve_MAC": "ff:ee:dd:cc:bb:aa",
|
||||
"devVendor": "Cisco",
|
||||
"eve_IP": "10.0.0.1",
|
||||
"eve_DateTime": "2025-01-15 08:00:00",
|
||||
"eve_EventType": "Device Down",
|
||||
"devComments": "",
|
||||
}
|
||||
]
|
||||
columns = ["devName", "eve_MAC", "devVendor", "eve_IP", "eve_DateTime", "eve_EventType", "devComments"]
|
||||
|
||||
json_data = _make_json("down_devices", down_devices, columns, "🔴 Down devices")
|
||||
_, text = construct_notifications(json_data, "down_devices")
|
||||
|
||||
self.assertIn("Router (ff:ee:dd:cc:bb:aa) down since 2025-01-15 08:00:00", text)
|
||||
|
||||
# -----------------------------------------------------------------
|
||||
# Down reconnected section uses same unified column names
|
||||
# -----------------------------------------------------------------
|
||||
@patch("models.notification_instance.get_setting_value")
|
||||
def test_down_reconnected_template(self, mock_setting):
|
||||
from models.notification_instance import construct_notifications
|
||||
|
||||
mock_setting.side_effect = self._setting_factory({
|
||||
"NTFPRCS_TEXT_SECTION_HEADERS": True,
|
||||
"NTFPRCS_TEXT_TEMPLATE_down_reconnected": "{devName} ({eve_MAC}) reconnected at {eve_DateTime}",
|
||||
})
|
||||
|
||||
reconnected = [
|
||||
{
|
||||
"devName": "Switch",
|
||||
"eve_MAC": "aa:11:bb:22:cc:33",
|
||||
"devVendor": "Netgear",
|
||||
"eve_IP": "10.0.0.2",
|
||||
"eve_DateTime": "2025-01-15 09:30:00",
|
||||
"eve_EventType": "Down Reconnected",
|
||||
"devComments": "",
|
||||
}
|
||||
]
|
||||
columns = ["devName", "eve_MAC", "devVendor", "eve_IP", "eve_DateTime", "eve_EventType", "devComments"]
|
||||
|
||||
json_data = _make_json("down_reconnected", reconnected, columns, "🔁 Reconnected down devices")
|
||||
_, text = construct_notifications(json_data, "down_reconnected")
|
||||
|
||||
self.assertIn("Switch (aa:11:bb:22:cc:33) reconnected at 2025-01-15 09:30:00", text)
|
||||
|
||||
# -----------------------------------------------------------------
|
||||
# HTML output is unchanged regardless of template config
|
||||
# -----------------------------------------------------------------
|
||||
@patch("models.notification_instance.get_setting_value")
|
||||
def test_html_unchanged_with_template(self, mock_setting):
|
||||
from models.notification_instance import construct_notifications
|
||||
|
||||
# Get HTML without template
|
||||
mock_setting.side_effect = self._setting_factory({
|
||||
"NTFPRCS_TEXT_SECTION_HEADERS": True,
|
||||
"NTFPRCS_TEXT_TEMPLATE_new_devices": "",
|
||||
})
|
||||
json_data = _make_json(
|
||||
"new_devices", SAMPLE_NEW_DEVICES, NEW_DEVICE_COLUMNS, "🆕 New devices"
|
||||
)
|
||||
html_without, _ = construct_notifications(json_data, "new_devices")
|
||||
|
||||
# Get HTML with template
|
||||
mock_setting.side_effect = self._setting_factory({
|
||||
"NTFPRCS_TEXT_SECTION_HEADERS": True,
|
||||
"NTFPRCS_TEXT_TEMPLATE_new_devices": "{devName} ({eve_MAC})",
|
||||
})
|
||||
html_with, _ = construct_notifications(json_data, "new_devices")
|
||||
|
||||
self.assertEqual(html_without, html_with)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -42,8 +42,10 @@ def test_db(test_db_path):
|
||||
eve_MAC TEXT,
|
||||
eve_DateTime TEXT,
|
||||
devLastIP TEXT,
|
||||
eve_IP TEXT,
|
||||
eve_EventType TEXT,
|
||||
devName TEXT,
|
||||
devVendor TEXT,
|
||||
devComments TEXT,
|
||||
eve_PendingAlertEmail INTEGER
|
||||
)
|
||||
@@ -84,13 +86,13 @@ def test_db(test_db_path):
|
||||
|
||||
# Insert test data
|
||||
test_data = [
|
||||
('aa:bb:cc:dd:ee:ff', '2024-01-01 12:00:00', '192.168.1.100', 'New Device', 'Test Device', 'Test Comment', 1),
|
||||
('11:22:33:44:55:66', '2024-01-01 12:01:00', '192.168.1.101', 'Connected', 'Test Device 2', 'Another Comment', 1),
|
||||
('77:88:99:aa:bb:cc', '2024-01-01 12:02:00', '192.168.1.102', 'Disconnected', 'Test Device 3', 'Third Comment', 1),
|
||||
('aa:bb:cc:dd:ee:ff', '2024-01-01 12:00:00', '192.168.1.100', '192.168.1.100', 'New Device', 'Test Device', 'Apple', 'Test Comment', 1),
|
||||
('11:22:33:44:55:66', '2024-01-01 12:01:00', '192.168.1.101', '192.168.1.101', 'Connected', 'Test Device 2', 'Dell', 'Another Comment', 1),
|
||||
('77:88:99:aa:bb:cc', '2024-01-01 12:02:00', '192.168.1.102', '192.168.1.102', 'Disconnected', 'Test Device 3', 'Cisco', 'Third Comment', 1),
|
||||
]
|
||||
cur.executemany('''
|
||||
INSERT INTO Events_Devices (eve_MAC, eve_DateTime, devLastIP, eve_EventType, devName, devComments, eve_PendingAlertEmail)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?)
|
||||
INSERT INTO Events_Devices (eve_MAC, eve_DateTime, devLastIP, eve_IP, eve_EventType, devName, devVendor, devComments, eve_PendingAlertEmail)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
''', test_data)
|
||||
|
||||
conn.commit()
|
||||
@@ -115,7 +117,7 @@ def test_fresh_install_compatibility(builder):
|
||||
def test_existing_db_compatibility():
|
||||
mock_db = Mock()
|
||||
mock_result = Mock()
|
||||
mock_result.columnNames = ['MAC', 'Datetime', 'IP', 'Event Type', 'Device name', 'Comments']
|
||||
mock_result.columnNames = ['devName', 'eve_MAC', 'devVendor', 'eve_IP', 'eve_DateTime', 'eve_EventType', 'devComments']
|
||||
mock_result.json = {'data': []}
|
||||
mock_db.get_table_as_json.return_value = mock_result
|
||||
|
||||
|
||||
Reference in New Issue
Block a user