diff --git a/hbd/server/http.py b/hbd/server/http.py index b86d945..c972e63 100644 --- a/hbd/server/http.py +++ b/hbd/server/http.py @@ -20,6 +20,7 @@ from . import users as users_mod from . import oauth as oauth_mod from . import ws as ws_mod from . import configio as configio_mod +from . import config_access logger = logging.getLogger(__name__) @@ -1407,19 +1408,13 @@ async def start( # ------------------------------------------------------------------------- def _visible_channels_for_user(user): - """Return {name: cfg} of channels visible to user (public + own private).""" + """Return {name: cfg} of channels visible to user (global + own).""" all_channels = config.get("notification_channels") or {} if user is None: return {} if user.admin: return dict(all_channels) - visible = {} - for name, cfg in all_channels.items(): - if not isinstance(cfg, dict): - continue - if not cfg.get("private") or cfg.get("owner") == user.username: - visible[name] = cfg - return visible + return config_access.user_channels(all_channels, user.username) def _build_channel_response(ch_name, ch_cfg): """Serialize a channel config dict for the API response.""" @@ -1443,7 +1438,7 @@ async def start( "type": ch_type, "type_label": settings_mod._CHANNEL_TYPE_LABELS.get(ch_type, ch_type.title()), "owner": ch_cfg.get("owner"), - "private": bool(ch_cfg.get("private", False)), + "private": not config_access.is_global(ch_cfg), "min_level": ch_cfg.get("min_level", "WARNING"), "fields": fields, } @@ -1508,9 +1503,12 @@ async def start( if body.get("min_level"): channel_cfg["min_level"] = body["min_level"] - channel_cfg["owner"] = user.username - if body.get("private"): - channel_cfg["private"] = True + if user.admin: + owner = (body.get("owner") or "").strip() + if owner: + channel_cfg["owner"] = owner + else: + channel_cfg["owner"] = user.username try: disk_data = configio_mod.read_roundtrip(_config_path) @@ -1575,12 +1573,12 @@ async def start( if body.get("min_level"): channel_cfg["min_level"] = body["min_level"] - if owner is not None: + if user.admin: + new_owner = (body.get("owner") or "").strip() if "owner" in body else (owner or "") + if new_owner: + channel_cfg["owner"] = new_owner + elif owner is not None: channel_cfg["owner"] = owner - if "private" in body: - channel_cfg["private"] = bool(body["private"]) - elif existing_on_disk.get("private"): - channel_cfg["private"] = True configio_mod.apply_channel(disk_data, ch_name, channel_cfg) configio_mod.write_config(_config_path, disk_data) diff --git a/tests/test_notification_channels_api.py b/tests/test_notification_channels_api.py index f5d311e..79fa653 100644 --- a/tests/test_notification_channels_api.py +++ b/tests/test_notification_channels_api.py @@ -86,61 +86,53 @@ def test_delete_channel_persisted_after_write(tmp_path): # --------------------------------------------------------------------------- -# Visibility logic (mirrors http.py _visible_channels_for_user) +# Visibility logic (owner-presence rule, hbd.server.config_access) # --------------------------------------------------------------------------- +from hbd.server import config_access as ca # noqa: E402 + + 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) - } + return set(ca.user_channels(all_channels, user["username"])) CONFIG_VISIBILITY = { "notification_channels": { "pub_ch": {"type": "pushover", "token": "t", "user": "u"}, - "alice_priv": {"type": "email", "owner": "alice", "private": True, + "alice_priv": {"type": "email", "owner": "alice", "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"}, + "bob_priv": {"type": "signal", "owner": "bob", "user": "+1", "recipient": "+2"}, + "stale_flag": {"type": "pushover", "token": "t2", "user": "u2", "private": True}, } } -def test_public_channel_visible_to_all(): +def test_global_channel_visible_to_all(): for uname in ("alice", "bob", "carol"): - user = {"username": uname, "admin": False} - assert "pub_ch" in _visible(CONFIG_VISIBILITY, user) + assert "pub_ch" in _visible(CONFIG_VISIBILITY, {"username": uname, "admin": False}) -def test_private_channel_visible_only_to_owner(): +def test_owned_channel_visible_only_to_owner(): alice = {"username": "alice", "admin": False} - bob = {"username": "bob", "admin": False} - carol = {"username": "carol", "admin": False} - + bob = {"username": "bob", "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"} + assert _visible(CONFIG_VISIBILITY, admin) == {"pub_ch", "alice_priv", "bob_priv", "stale_flag"} -def test_admin_owned_channel_is_public_by_default(): +def test_stale_private_flag_without_owner_is_global(): + """Owner-presence is the single signal; a leftover private flag is ignored.""" alice = {"username": "alice", "admin": False} - assert "admin_owned" in _visible(CONFIG_VISIBILITY, alice) + assert "stale_flag" in _visible(CONFIG_VISIBILITY, alice) # ---------------------------------------------------------------------------