Merge pull request #125 from jagerman/global-mod-tests

Add tests for global mods/admins

Author: jagerman

sogs/__main__.py

@@ -152,7 +152,7 @@
 from . import web
 from .model.room import Room, get_rooms
 from .model.user import User, SystemUser, get_all_global_moderators
-from .model.exc import AlreadyExists, NoSuchRoom
+from .model.exc import AlreadyExists, NoSuchRoom, NoSuchUser
 
 web.appdb = db.get_conn()
 
@@ -322,7 +322,20 @@ def print_room(room: Room):
                 print(f"{u.session_id} was not a global moderator")
             else:
                 u.remove_moderator(removed_by=sysadmin)
-                print(f"Removed {sid} as global {'admin' if was_admin else 'moderator'}")
+                print(f"Removed {u.session_id} as global {'admin' if was_admin else 'moderator'}")
+
+            if u.is_blinded and sid.startswith('05'):
+                try:
+                    u2 = User(session_id=sid, try_blinding=False, autovivify=False)
+                    if u2.global_admin or u2.global_moderator:
+                        was_admin = u2.global_admin
+                        u2.remove_moderator(removed_by=sysadmin)
+                        print(
+                            f"Removed {u2.session_id} as global "
+                            f"{'admin' if was_admin else 'moderator'}"
+                        )
+                except NoSuchUser:
+                    pass
     else:
         if args.rooms == ['*']:
             rooms = get_rooms()
@@ -334,9 +347,22 @@ def print_room(room: Room):
 
         for sid in args.delete_moderators:
             u = User(session_id=sid, try_blinding=True)
+            u2 = None
+            if u.is_blinded and sid.startswith('05'):
+                try:
+                    u2 = User(session_id=sid, try_blinding=False, autovivify=False)
+                except NoSuchUser:
+                    pass
+
             for room in rooms:
                 room.remove_moderator(u, removed_by=sysadmin)
                 print(f"Removed {u.session_id} as moderator/admin of {room.name} ({room.token})")
+                if u2 is not None:
+                    room.remove_moderator(u2, removed_by=sysadmin)
+                    print(
+                        f"Removed {u2.session_id} as moderator/admin of {room.name} ({room.token})"
+                    )
+
 elif args.list_rooms:
     rooms = get_rooms()
     if rooms:

sogs/model/room.py

@@ -1398,22 +1398,23 @@ def remove_moderator(self, user: User, *, removed_by: User, remove_admin_only: b
             raise BadPermission()
 
         with db.transaction():
-            query(
-                f"""
-                UPDATE user_permission_overrides
-                SET admin = FALSE
-                    {', moderator = FALSE, visible_mod = TRUE' if not remove_admin_only else ''}
-                WHERE room = :r AND "user" = :u
-                """,
-                r=self.id,
-                u=user.id,
-            )
+            with user.check_blinding() as u:
+                query(
+                    f"""
+                    UPDATE user_permission_overrides
+                    SET admin = FALSE
+                        {', moderator = FALSE, visible_mod = TRUE' if not remove_admin_only else ''}
+                    WHERE room = :r AND "user" = :u
+                    """,
+                    r=self.id,
+                    u=user.id,
+                )
 
-            self._refresh()
-            if user.id in self._perm_cache:
-                del self._perm_cache[user.id]
+                self._refresh()
+                if user.id in self._perm_cache:
+                    del self._perm_cache[user.id]
 
-        app.logger.info(f"{removed_by} removed {user} as mod/admin of {self}")
+                app.logger.info(f"{removed_by} removed {u} as mod/admin of {self}")
 
     def ban_user(self, to_ban: User, *, mod: User, timeout: Optional[float] = None):
         """
@@ -1496,24 +1497,28 @@ def unban_user(self, to_unban: User, *, mod: User):
             app.logger.warning(f"Error unbanning {to_unban} from {self} by {mod}: not a moderator")
             raise BadPermission()
 
-        result = query(
-            """
-            UPDATE user_permission_overrides SET banned = FALSE
-            WHERE room = :r AND "user" = :unban AND banned
-            """,
-            r=self.id,
-            unban=to_unban.id,
-        )
-        if result.rowcount > 0:
-            app.logger.debug(f"{mod} unbanned {to_unban} from {self}")
+        with db.transaction():
+            with to_unban.check_blinding() as to_unban:
+                result = query(
+                    """
+                    UPDATE user_permission_overrides SET banned = FALSE
+                    WHERE room = :r AND "user" = :unban AND banned
+                    """,
+                    r=self.id,
+                    unban=to_unban.id,
+                )
+                if result.rowcount > 0:
+                    app.logger.debug(f"{mod} unbanned {to_unban} from {self}")
 
-            if to_unban.id in self._perm_cache:
-                del self._perm_cache[to_unban.id]
+                    if to_unban.id in self._perm_cache:
+                        del self._perm_cache[to_unban.id]
 
-            return True
+                    return True
 
-        app.logger.debug(f"{mod} unbanned {to_unban} from {self} (but user was already unbanned)")
-        return False
+                app.logger.debug(
+                    f"{mod} unbanned {to_unban} from {self} (but user was already unbanned)"
+                )
+                return False
 
     def get_bans(self):
         """

sogs/routes/legacy.py

@@ -331,7 +331,7 @@ def handle_legacy_single_delete(msgid):
 @legacy.post("/block_list")
 def handle_legacy_ban():
     user, room = legacy_check_user_room(moderator=True)
-    ban = User(session_id=request.json['public_key'], autovivify=True)
+    ban = User(session_id=request.json['public_key'], autovivify=True, try_blinding=True)
 
     room.ban_user(to_ban=ban, mod=user)
 
@@ -341,7 +341,7 @@ def handle_legacy_ban():
 @legacy.post("/ban_and_delete_all")
 def handle_legacy_banhammer():
     mod, room = legacy_check_user_room(moderator=True)
-    ban = User(session_id=request.json['public_key'], autovivify=True)
+    ban = User(session_id=request.json['public_key'], autovivify=True, try_blinding=True)
 
     with db.transaction():
         room.ban_user(to_ban=ban, mod=mod)
@@ -353,7 +353,7 @@ def handle_legacy_banhammer():
 @legacy.delete("/block_list/<SessionID:session_id>")
 def handle_legacy_unban(session_id):
     user, room = legacy_check_user_room(moderator=True)
-    to_unban = User(session_id=session_id, autovivify=False)
+    to_unban = User(session_id=session_id, autovivify=False, try_blinding=True)
     if room.unban_user(to_unban, mod=user):
         return jsonify({"status_code": http.OK})
 
@@ -395,7 +395,7 @@ def handle_legacy_add_admin():
     if len(session_id) != 66 or not session_id.startswith("05"):
         abort(http.BAD_REQUEST)
 
-    mod = User(session_id=session_id, autovivify=True)
+    mod = User(session_id=session_id, autovivify=True, try_blinding=True)
     room.set_moderator(mod, admin=True, visible=True, added_by=user)
 
     return jsonify({"status_code": http.OK})
@@ -408,7 +408,7 @@ def handle_legacy_add_admin():
 def handle_legacy_remove_admin(session_id):
     user, room = legacy_check_user_room(admin=True)
 
-    mod = User(session_id=session_id, autovivify=False)
+    mod = User(session_id=session_id, autovivify=False, try_blinding=True)
     room.remove_moderator(mod, removed_by=user)
 
     return jsonify({"status_code": http.OK})

tests/test_room_routes.py

@@ -185,6 +185,73 @@ def test_list(client, room, room2, user, user2, admin, mod, global_mod, global_a
     assert r.json == {**r4_expected, **r4_exp_defs, **exp_gadmin}
 
 
+def test_visible_global_mods(client, room, user, mod, global_mod, global_admin):
+    url_room = '/room/test-room'
+    expect_room = {
+        "token": "test-room",
+        "name": "Test room",
+        "description": "Test suite testing room",
+        "info_updates": 3,
+        "message_sequence": 0,
+        "created": room.created,
+        "active_users": 0,
+        "active_users_cutoff": int(sogs.config.ROOM_DEFAULT_ACTIVE_THRESHOLD),
+        "moderators": [mod.session_id],
+        "admins": [],
+        "read": True,
+        "write": True,
+        "upload": True,
+    }
+    r = sogs_get(client, url_room, user)
+    assert r.status_code == 200
+    assert r.json == expect_room
+
+    expected_for_moderator = {
+        **expect_room,
+        **{'default_' + x: True for x in ('accessible', 'read', 'write', 'upload')},
+        'global_moderator': True,
+        'hidden_admins': [global_admin.session_id],
+        'hidden_moderators': [global_mod.session_id],
+        'moderator': True,
+        'moderators': [mod.session_id],
+    }
+    r = sogs_get(client, "/room/test-room/pollInfo/0", global_mod)
+    assert r.status_code == 200
+    assert r.json == {
+        'token': 'test-room',
+        'active_users': 0,
+        'details': expected_for_moderator,
+        'read': True,
+        'write': True,
+        'upload': True,
+        'moderator': True,
+        'global_moderator': True,
+        'default_accessible': True,
+        'default_read': True,
+        'default_write': True,
+        'default_upload': True,
+    }
+
+    global_mod.set_moderator(added_by=global_admin, admin=False, visible=True)
+    global_admin.set_moderator(added_by=global_admin, admin=True, visible=True)
+
+    for e in (expect_room, expected_for_moderator):
+        e["moderators"] = sorted([mod.session_id, global_mod.session_id])
+        e["admins"] = [global_admin.session_id]
+        e["info_updates"] += 2
+    del expected_for_moderator["hidden_admins"]
+    del expected_for_moderator["hidden_moderators"]
+
+    r = sogs_get(client, url_room, user)
+    assert r.status_code == 200
+    assert r.json == expect_room
+
+    r = sogs_get(client, url_room, mod)
+    assert r.status_code == 200
+    del expected_for_moderator["global_moderator"]
+    assert r.json == expected_for_moderator
+
+
 def test_updates(client, room, user, user2, mod, admin, global_mod, global_admin):
     url_room = '/room/test-room'
     r = sogs_get(client, url_room, user)