feat: replace YAML notification channel editor with form-based UI
Notification channels are now managed through a proper web form instead
of a raw YAML textarea. Any authenticated user can create channels; private
channels (owner-scoped) are hidden from other users. The user profile
channel selector becomes a tag/chip picker with a "My Channels" CRUD section.
- settings.py: add CHANNEL_TYPE_SCHEMAS for all 6 notifier types; channel
section switches to section_mode="channels"; cards include owner/private/min_level
- configio.py: add apply_channel() and delete_channel() for per-entry CRUD
- notify.py: strip owner/private metadata before dispatching to drivers
- http.py: add GET/POST /api/0/notification_channels, PUT/DELETE /{name},
GET /api/0/notification_channel_types; visibility helper filters private
channels per user; PUT /api/0/users/me validates against visible channels
- settings.html: card grid with edit/delete per channel; add/edit modal
with type dropdown and dynamically rendered type-specific fields
- profile.html: chip picker replaces checkbox list; My Channels section
for creating/editing/deleting user-owned channels
- tests: update test_settings_sections, test_http_users_me; add
test_notification_channels_api (16 new tests, 46 total passing)
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -83,3 +83,41 @@ def test_put_users_me_notification_channels(tmp_path):
|
||||
configio.write_config(str(cfg), data)
|
||||
result = configio.read_roundtrip(str(cfg))
|
||||
assert result["users"]["alice"]["notification_channels"] == ["pushover_ops"]
|
||||
|
||||
|
||||
def test_visible_channels_excludes_private_from_others():
|
||||
"""Private channels owned by another user must not appear in the visible set."""
|
||||
from hbd.server import settings as settings_mod
|
||||
|
||||
config = {
|
||||
"notification_channels": {
|
||||
"public_ch": {"type": "pushover", "token": "t", "user": "u"},
|
||||
"alice_priv": {"type": "email", "owner": "alice", "private": True,
|
||||
"recipients": ["a@b.com"], "sender": "s@b.com", "smtp_server": "s"},
|
||||
"bob_priv": {"type": "email", "owner": "bob", "private": True,
|
||||
"recipients": ["b@b.com"], "sender": "s@b.com", "smtp_server": "s"},
|
||||
}
|
||||
}
|
||||
|
||||
class FakeUser:
|
||||
def __init__(self, username, admin=False):
|
||||
self.username = username
|
||||
self.admin = admin
|
||||
|
||||
alice = FakeUser("alice")
|
||||
bob = FakeUser("bob")
|
||||
admin = FakeUser("admin", admin=True)
|
||||
|
||||
# Simulate _visible_channels_for_user logic (mirrors http.py implementation)
|
||||
def visible(user):
|
||||
all_channels = config.get("notification_channels") or {}
|
||||
if user.admin:
|
||||
return set(all_channels.keys())
|
||||
return {
|
||||
name for name, cfg in all_channels.items()
|
||||
if not cfg.get("private") or cfg.get("owner") == user.username
|
||||
}
|
||||
|
||||
assert visible(alice) == {"public_ch", "alice_priv"}
|
||||
assert visible(bob) == {"public_ch", "bob_priv"}
|
||||
assert visible(admin) == {"public_ch", "alice_priv", "bob_priv"}
|
||||
|
||||
@@ -0,0 +1,178 @@
|
||||
"""Tests for notification channel CRUD via configio helpers and visibility logic."""
|
||||
import pytest
|
||||
from hbd.server import configio, settings as settings_mod
|
||||
|
||||
|
||||
SAMPLE_YAML = """\
|
||||
hbd_port: 50004
|
||||
notification_channels:
|
||||
pushover_ops:
|
||||
type: pushover
|
||||
token: abc123
|
||||
user: usr456
|
||||
"""
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# configio helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_apply_channel_adds_new_entry(tmp_path):
|
||||
f = tmp_path / ".hb.yaml"
|
||||
f.write_text(SAMPLE_YAML)
|
||||
data = configio.read_roundtrip(str(f))
|
||||
configio.apply_channel(data, "email_ops", {"type": "email", "recipients": ["ops@example.com"]})
|
||||
assert "email_ops" in data["notification_channels"]
|
||||
assert data["notification_channels"]["email_ops"]["type"] == "email"
|
||||
# Existing channel preserved
|
||||
assert "pushover_ops" in data["notification_channels"]
|
||||
|
||||
|
||||
def test_apply_channel_updates_existing(tmp_path):
|
||||
f = tmp_path / ".hb.yaml"
|
||||
f.write_text(SAMPLE_YAML)
|
||||
data = configio.read_roundtrip(str(f))
|
||||
configio.apply_channel(data, "pushover_ops", {"type": "pushover", "token": "new_tok", "user": "new_usr"})
|
||||
assert data["notification_channels"]["pushover_ops"]["token"] == "new_tok"
|
||||
|
||||
|
||||
def test_apply_channel_creates_section_if_absent():
|
||||
data = {"hbd_port": 50004}
|
||||
configio.apply_channel(data, "test_ch", {"type": "pushover", "token": "t", "user": "u"})
|
||||
assert "notification_channels" in data
|
||||
assert "test_ch" in data["notification_channels"]
|
||||
|
||||
|
||||
def test_delete_channel_removes_entry(tmp_path):
|
||||
f = tmp_path / ".hb.yaml"
|
||||
f.write_text(SAMPLE_YAML)
|
||||
data = configio.read_roundtrip(str(f))
|
||||
configio.delete_channel(data, "pushover_ops")
|
||||
assert "pushover_ops" not in data["notification_channels"]
|
||||
|
||||
|
||||
def test_delete_channel_noop_for_missing():
|
||||
data = {"notification_channels": {"ch1": {"type": "pushover"}}}
|
||||
configio.delete_channel(data, "nonexistent") # must not raise
|
||||
assert "ch1" in data["notification_channels"]
|
||||
|
||||
|
||||
def test_delete_channel_noop_when_no_section():
|
||||
data = {}
|
||||
configio.delete_channel(data, "anything") # must not raise
|
||||
|
||||
|
||||
def test_apply_channel_persisted_after_write(tmp_path):
|
||||
f = tmp_path / ".hb.yaml"
|
||||
f.write_text(SAMPLE_YAML)
|
||||
data = configio.read_roundtrip(str(f))
|
||||
configio.apply_channel(data, "signal_ops", {"type": "signal", "user": "+1", "recipient": "+2"})
|
||||
configio.write_config(str(f), data)
|
||||
result = configio.read_roundtrip(str(f))
|
||||
assert "signal_ops" in result["notification_channels"]
|
||||
assert result["notification_channels"]["signal_ops"]["user"] == "+1"
|
||||
# Original channel preserved
|
||||
assert "pushover_ops" in result["notification_channels"]
|
||||
|
||||
|
||||
def test_delete_channel_persisted_after_write(tmp_path):
|
||||
f = tmp_path / ".hb.yaml"
|
||||
f.write_text(SAMPLE_YAML)
|
||||
data = configio.read_roundtrip(str(f))
|
||||
configio.delete_channel(data, "pushover_ops")
|
||||
configio.write_config(str(f), data)
|
||||
result = configio.read_roundtrip(str(f))
|
||||
assert "pushover_ops" not in (result.get("notification_channels") or {})
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Visibility logic (mirrors http.py _visible_channels_for_user)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _visible(config, user):
|
||||
"""Local copy of the visibility helper for unit testing without the HTTP layer."""
|
||||
all_channels = config.get("notification_channels") or {}
|
||||
if user.get("admin"):
|
||||
return set(all_channels.keys())
|
||||
username = user["username"]
|
||||
return {
|
||||
name for name, cfg in all_channels.items()
|
||||
if isinstance(cfg, dict) and (not cfg.get("private") or cfg.get("owner") == username)
|
||||
}
|
||||
|
||||
|
||||
CONFIG_VISIBILITY = {
|
||||
"notification_channels": {
|
||||
"pub_ch": {"type": "pushover", "token": "t", "user": "u"},
|
||||
"alice_priv": {"type": "email", "owner": "alice", "private": True,
|
||||
"recipients": ["a@a.com"], "sender": "s@a.com", "smtp_server": "s"},
|
||||
"bob_priv": {"type": "signal", "owner": "bob", "private": True,
|
||||
"user": "+1", "recipient": "+2"},
|
||||
"admin_owned": {"type": "pushover", "token": "t2", "user": "u2", "owner": "adminuser"},
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
def test_public_channel_visible_to_all():
|
||||
for uname in ("alice", "bob", "carol"):
|
||||
user = {"username": uname, "admin": False}
|
||||
assert "pub_ch" in _visible(CONFIG_VISIBILITY, user)
|
||||
|
||||
|
||||
def test_private_channel_visible_only_to_owner():
|
||||
alice = {"username": "alice", "admin": False}
|
||||
bob = {"username": "bob", "admin": False}
|
||||
carol = {"username": "carol", "admin": False}
|
||||
|
||||
assert "alice_priv" in _visible(CONFIG_VISIBILITY, alice)
|
||||
assert "alice_priv" not in _visible(CONFIG_VISIBILITY, bob)
|
||||
assert "alice_priv" not in _visible(CONFIG_VISIBILITY, carol)
|
||||
|
||||
assert "bob_priv" in _visible(CONFIG_VISIBILITY, bob)
|
||||
assert "bob_priv" not in _visible(CONFIG_VISIBILITY, alice)
|
||||
|
||||
|
||||
def test_admin_sees_all_channels():
|
||||
admin = {"username": "adminuser", "admin": True}
|
||||
visible = _visible(CONFIG_VISIBILITY, admin)
|
||||
assert visible == {"pub_ch", "alice_priv", "bob_priv", "admin_owned"}
|
||||
|
||||
|
||||
def test_admin_owned_channel_is_public_by_default():
|
||||
alice = {"username": "alice", "admin": False}
|
||||
assert "admin_owned" in _visible(CONFIG_VISIBILITY, alice)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Channel type schemas
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_all_required_types_in_schema():
|
||||
for t in ("pushover", "email", "signal", "matrix", "sms_voipms"):
|
||||
assert t in settings_mod.CHANNEL_TYPE_SCHEMAS
|
||||
|
||||
|
||||
def test_schema_fields_have_required_keys():
|
||||
for type_id, schema in settings_mod.CHANNEL_TYPE_SCHEMAS.items():
|
||||
assert "label" in schema, f"{type_id} missing label"
|
||||
assert "fields" in schema, f"{type_id} missing fields"
|
||||
for f in schema["fields"]:
|
||||
for k in ("key", "label", "type", "required"):
|
||||
assert k in f, f"{type_id} field missing {k!r}"
|
||||
|
||||
|
||||
def test_secret_fields_use_secret_type():
|
||||
"""Known secret fields must be typed 'secret' so the UI masks them."""
|
||||
secret_keys = {"token", "user_key", "api_key", "api_password",
|
||||
"smtp_password", "access_token"}
|
||||
for type_id, schema in settings_mod.CHANNEL_TYPE_SCHEMAS.items():
|
||||
for f in schema["fields"]:
|
||||
if f["key"] in secret_keys:
|
||||
assert f["type"] == "secret", (
|
||||
f"{type_id}.{f['key']} should be type 'secret'"
|
||||
)
|
||||
|
||||
|
||||
def test_channel_labels_not_empty():
|
||||
for type_id, schema in settings_mod.CHANNEL_TYPE_SCHEMAS.items():
|
||||
assert schema["label"].strip(), f"{type_id} has empty label"
|
||||
@@ -24,7 +24,7 @@ def test_sections_have_section_mode():
|
||||
sections = settings_mod.get_settings_sections(CFG)
|
||||
for s in sections:
|
||||
assert "section_mode" in s, f"Section {s['id']} missing section_mode"
|
||||
assert s["section_mode"] in ("form", "yaml")
|
||||
assert s["section_mode"] in ("form", "yaml", "channels")
|
||||
|
||||
|
||||
def test_sections_have_api_section():
|
||||
@@ -45,16 +45,41 @@ def test_network_section_has_editable_fields():
|
||||
def test_yaml_sections_have_correct_mode():
|
||||
sections = settings_mod.get_settings_sections(CFG)
|
||||
yaml_sections = {s["id"]: s for s in sections if s["section_mode"] == "yaml"}
|
||||
assert "channels" in yaml_sections
|
||||
assert "channels" not in yaml_sections # now uses "channels" mode
|
||||
assert "hosts" in yaml_sections
|
||||
assert "thresholds" in yaml_sections
|
||||
assert "dns" in yaml_sections
|
||||
assert yaml_sections["channels"]["api_section"] == "notification_channels"
|
||||
assert yaml_sections["hosts"]["api_section"] == "hosts"
|
||||
assert yaml_sections["thresholds"]["api_section"] == "thresholds"
|
||||
assert yaml_sections["dns"]["api_section"] == "dns"
|
||||
|
||||
|
||||
def test_channels_section_uses_channels_mode():
|
||||
sections = settings_mod.get_settings_sections(CFG)
|
||||
ch_sec = next(s for s in sections if s["id"] == "channels")
|
||||
assert ch_sec["section_mode"] == "channels"
|
||||
assert ch_sec["api_section"] == "notification_channels"
|
||||
assert len(ch_sec["channels"]) == 1
|
||||
ch = ch_sec["channels"][0]
|
||||
assert ch["name"] == "pushover_ops"
|
||||
assert ch["type"] == "pushover"
|
||||
assert "owner" in ch
|
||||
assert "private" in ch
|
||||
|
||||
|
||||
def test_channel_type_schemas_exported():
|
||||
assert hasattr(settings_mod, "CHANNEL_TYPE_SCHEMAS")
|
||||
for required_type in ("pushover", "email", "signal", "matrix", "sms_voipms"):
|
||||
assert required_type in settings_mod.CHANNEL_TYPE_SCHEMAS
|
||||
schema = settings_mod.CHANNEL_TYPE_SCHEMAS[required_type]
|
||||
assert "label" in schema
|
||||
assert "fields" in schema
|
||||
for f in schema["fields"]:
|
||||
assert "key" in f
|
||||
assert "type" in f
|
||||
assert "required" in f
|
||||
|
||||
|
||||
def test_oauth_section_exists():
|
||||
sections = settings_mod.get_settings_sections(CFG)
|
||||
oauth = next((s for s in sections if s["id"] == "oauth"), None)
|
||||
|
||||
Reference in New Issue
Block a user