[Mod] Mute/unmute bugfixes (#2230)

- Helper methods mute_user and unmute_user now take GuildChannel instead of solely TextChannel.
- The bot will not attempt to mute a member with the Administrator permission, as that permission bypasses channel overwrites anyway. The bot will still unmute members with the Administrator permission (see #2076).
- Audit reasons are now specified for mass mutes / unmutes.
- A few typos and missing keyword specifiers were corrected.
- Streamlined some logic and used some already-existing functions.

redbot/cogs/mod/mod.py

@@ -893,34 +893,33 @@ class Mod(commands.Cog):
         author = ctx.author
         if user_voice_state:
             channel = user_voice_state.channel
-            if channel and channel.permissions_for(user).speak:
+            if channel:
-                overwrites = channel.overwrites_for(user)
+                audit_reason = get_audit_reason(author, reason)
-                overwrites.speak = False
+
-                audit_reason = get_audit_reason(ctx.author, reason)
+                success, issue = await self.mute_user(guild, channel, author, user, audit_reason)
-                await channel.set_permissions(user, overwrite=overwrites, reason=audit_reason)
+
-                await ctx.send(
+                if success:
-                    _("Muted {user} in channel {channel.name}").format(user, channel=channel)
+                    await ctx.send(
-                )
+                        _("Muted {user} in channel {channel.name}").format(
-                try:
+                            user=user, channel=channel
-                    await modlog.create_case(
+                        )
-                        self.bot,
-                        guild,
-                        ctx.message.created_at,
-                        "boicemute",
-                        user,
-                        author,
-                        reason,
-                        until=None,
-                        channel=channel,
                     )
-                except RuntimeError as e:
+                    try:
-                    await ctx.send(e)
+                        await modlog.create_case(
-                return
+                            self.bot,
-            elif channel.permissions_for(user).speak is False:
+                            guild,
-                await ctx.send(
+                            ctx.message.created_at,
-                    _("That user is already muted in {channel}!").format(channel=channel.name)
+                            "vmute",
-                )
+                            user,
-                return
+                            author,
+                            reason,
+                            until=None,
+                            channel=channel,
+                        )
+                    except RuntimeError as e:
+                        await ctx.send(e)
+                else:
+                    await channel.send(issue)
             else:
                 await ctx.send(_("That user is not in a voice channel right now!"))
         else:
@@ -938,13 +937,7 @@ class Mod(commands.Cog):
         author = ctx.message.author
         channel = ctx.message.channel
         guild = ctx.guild
-
+        audit_reason = get_audit_reason(author, reason)
-        if reason is None:
-            audit_reason = "Channel mute requested by {a} (ID {a.id})".format(a=author)
-        else:
-            audit_reason = "Channel mute requested by {a} (ID {a.id}). Reason: {r}".format(
-                a=author, r=reason
-            )
 
         success, issue = await self.mute_user(guild, channel, author, user, audit_reason)
 
@@ -975,26 +968,12 @@ class Mod(commands.Cog):
         """Mutes user in the server"""
         author = ctx.message.author
         guild = ctx.guild
-        if reason is None:
+        audit_reason = get_audit_reason(author, reason)
-            audit_reason = "server mute requested by {author} (ID {author.id})".format(
-                author=author
-            )
-        else:
-            audit_reason = (
-                "server mute requested by {author} (ID {author.id}). Reason: {reason}"
-            ).format(author=author, reason=reason)
 
         mute_success = []
         for channel in guild.channels:
-            if not isinstance(channel, discord.TextChannel):
+            success, issue = await self.mute_user(guild, channel, author, user, audit_reason)
-                if channel.permissions_for(user).speak:
+            mute_success.append((success, issue))
-                    overwrites = channel.overwrites_for(user)
-                    overwrites.speak = False
-                    audit_reason = get_audit_reason(ctx.author, reason)
-                    await channel.set_permissions(user, overwrite=overwrites, reason=audit_reason)
-            else:
-                success, issue = await self.mute_user(guild, channel, author, user, audit_reason)
-                mute_success.append((success, issue))
             await asyncio.sleep(0.1)
         await ctx.send(_("User has been muted in this server."))
         try:
@@ -1015,7 +994,7 @@ class Mod(commands.Cog):
     async def mute_user(
         self,
         guild: discord.Guild,
-        channel: discord.TextChannel,
+        channel: discord.abc.GuildChannel,
         author: discord.Member,
         user: discord.Member,
         reason: str,
@@ -1023,25 +1002,32 @@ class Mod(commands.Cog):
         """Mutes the specified user in the specified channel"""
         overwrites = channel.overwrites_for(user)
         permissions = channel.permissions_for(user)
-        perms_cache = await self.settings.member(user).perms_cache()
 
-        if overwrites.send_messages is False or permissions.send_messages is False:
+        if permissions.administrator:
+            return False, T_(mute_unmute_issues["is_admin"])
+
+        new_overs = {}
+        if not isinstance(channel, discord.TextChannel):
+            new_overs.update(speak=False)
+        if not isinstance(channel, discord.VoiceChannel):
+            new_overs.update(send_messages=False, add_reactions=False)
+
+        if all(getattr(permissions, p) is False for p in new_overs.keys()):
             return False, T_(mute_unmute_issues["already_muted"])
 
         elif not await is_allowed_by_hierarchy(self.bot, self.settings, guild, author, user):
             return False, T_(mute_unmute_issues["hierarchy_problem"])
 
-        perms_cache[str(channel.id)] = {
+        old_overs = {k: getattr(overwrites, k) for k in new_overs}
-            "send_messages": overwrites.send_messages,
+        overwrites.update(**new_overs)
-            "add_reactions": overwrites.add_reactions,
-        }
-        overwrites.update(send_messages=False, add_reactions=False)
         try:
             await channel.set_permissions(user, overwrite=overwrites, reason=reason)
         except discord.Forbidden:
             return False, T_(mute_unmute_issues["permissions_issue"])
         else:
-            await self.settings.member(user).perms_cache.set(perms_cache)
+            await self.settings.member(user).set_raw(
+                "perms_cache", str(channel.id), value=old_overs
+            )
             return True, None
 
     @commands.group()
@@ -1061,37 +1047,39 @@ class Mod(commands.Cog):
     ):
         """Unmute a user in their current voice channel."""
         user_voice_state = user.voice
+        guild = ctx.guild
+        author = ctx.author
         if user_voice_state:
             channel = user_voice_state.channel
-            if channel and channel.permissions_for(user).speak is False:
+            if channel:
-                overwrites = channel.overwrites_for(user)
+                audit_reason = get_audit_reason(author, reason)
-                overwrites.speak = None
+
-                audit_reason = get_audit_reason(ctx.author, reason)
+                success, message = await self.unmute_user(
-                await channel.set_permissions(user, overwrite=overwrites, reason=audit_reason)
+                    guild, channel, author, user, audit_reason
-                author = ctx.author
-                guild = ctx.guild
-                await ctx.send(
-                    _("Unmuted {}#{} in channel {}").format(
-                        user.name, user.discriminator, channel.name
-                    )
                 )
-                try:
+
-                    await modlog.create_case(
+                if success:
-                        self.bot,
+                    await ctx.send(
-                        guild,
+                        _("Unmuted {user} in channel {channel.name}").format(
-                        ctx.message.created_at,
+                            user=user, channel=channel
-                        "voiceunmute",
+                        )
-                        user,
-                        author,
-                        reason,
-                        until=None,
-                        channel=channel,
                     )
-                except RuntimeError as e:
+                    try:
-                    await ctx.send(e)
+                        await modlog.create_case(
-            elif channel.permissions_for(user).speak:
+                            self.bot,
-                await ctx.send(_("That user is already unmuted in {}!").format(channel.name))
+                            guild,
-                return
+                            ctx.message.created_at,
+                            "vunmute",
+                            user,
+                            author,
+                            reason,
+                            until=None,
+                            channel=channel,
+                        )
+                    except RuntimeError as e:
+                        await ctx.send(e)
+                else:
+                    await ctx.send(_("Unmute failed. Reason: {}").format(message))
             else:
                 await ctx.send(_("That user is not in a voice channel right now!"))
         else:
@@ -1109,8 +1097,9 @@ class Mod(commands.Cog):
         channel = ctx.channel
         author = ctx.author
         guild = ctx.guild
+        audit_reason = get_audit_reason(author, reason)
 
-        success, message = await self.unmute_user(guild, channel, author, user)
+        success, message = await self.unmute_user(guild, channel, author, user, audit_reason)
 
         if success:
             await ctx.send(_("User unmuted in this channel."))
@@ -1141,16 +1130,11 @@ class Mod(commands.Cog):
         """Unmute a user in this server."""
         guild = ctx.guild
         author = ctx.author
+        audit_reason = get_audit_reason(author, reason)
 
         unmute_success = []
         for channel in guild.channels:
-            if not isinstance(channel, discord.TextChannel):
+            success, message = await self.unmute_user(guild, channel, author, user, audit_reason)
-                if channel.permissions_for(user).speak is False:
-                    overwrites = channel.overwrites_for(user)
-                    overwrites.speak = None
-                    audit_reason = get_audit_reason(author, reason)
-                    await channel.set_permissions(user, overwrite=overwrites, reason=audit_reason)
-            success, message = await self.unmute_user(guild, channel, author, user)
             unmute_success.append((success, message))
             await asyncio.sleep(0.1)
         await ctx.send(_("User has been unmuted in this server."))
@@ -1171,45 +1155,37 @@ class Mod(commands.Cog):
     async def unmute_user(
         self,
         guild: discord.Guild,
-        channel: discord.TextChannel,
+        channel: discord.abc.GuildChannel,
         author: discord.Member,
         user: discord.Member,
+        reason: str,
     ) -> (bool, str):
         overwrites = channel.overwrites_for(user)
-        permissions = channel.permissions_for(user)
         perms_cache = await self.settings.member(user).perms_cache()
 
-        if overwrites.send_messages or permissions.send_messages:
+        if channel.id in perms_cache:
+            old_values = perms_cache[channel.id]
+        else:
+            old_values = {"send_messages": None, "add_reactions": None, "speak": None}
+
+        if all(getattr(overwrites, k) == v for k, v in old_values.items()):
             return False, T_(mute_unmute_issues["already_unmuted"])
 
         elif not await is_allowed_by_hierarchy(self.bot, self.settings, guild, author, user):
             return False, T_(mute_unmute_issues["hierarchy_problem"])
 
-        if channel.id in perms_cache:
+        overwrites.update(**old_values)
-            old_values = perms_cache[channel.id]
-        else:
-            old_values = {"send_messages": None, "add_reactions": None}
-        overwrites.update(
-            send_messages=old_values["send_messages"], add_reactions=old_values["add_reactions"]
-        )
-        is_empty = self.are_overwrites_empty(overwrites)
-
         try:
-            if not is_empty:
+            if overwrites.is_empty():
-                await channel.set_permissions(user, overwrite=overwrites)
-            else:
                 await channel.set_permissions(
-                    user, overwrite=cast(discord.PermissionOverwrite, None)
+                    user, overwrite=cast(discord.PermissionOverwrite, None), reason=reason
                 )
+            else:
+                await channel.set_permissions(user, overwrite=overwrites, reason=reason)
         except discord.Forbidden:
             return False, T_(mute_unmute_issues["permissions_issue"])
         else:
-            try:
+            await self.settings.member(user).clear_raw("perms_cache", str(channel.id))
-                del perms_cache[channel.id]
-            except KeyError:
-                pass
-            else:
-                await self.settings.member(user).perms_cache.set(perms_cache)
             return True, None
 
     @commands.group()
@@ -1695,20 +1671,15 @@ class Mod(commands.Cog):
                 while len(nick_list) > 20:
                     nick_list.pop(0)
 
-    @staticmethod
-    def are_overwrites_empty(overwrites):
-        """There is currently no cleaner way to check if a
-        PermissionOverwrite object is empty"""
-        return [p for p in iter(overwrites)] == [p for p in iter(discord.PermissionOverwrite())]
-
 
 _ = lambda s: s
 mute_unmute_issues = {
     "already_muted": _("That user can't send messages in this channel."),
-    "already_unmuted": _("That user isn't muted in this channel!"),
+    "already_unmuted": _("That user isn't muted in this channel."),
     "hierarchy_problem": _(
-        "I cannot let you do that. You are not higher than " "the user in the role hierarchy."
+        "I cannot let you do that. You are not higher than the user in the role hierarchy."
     ),
+    "is_admin": _("That user cannot be muted, as they have the Administrator permission."),
     "permissions_issue": _(
         "Failed to mute user. I need the manage roles "
         "permission and the user I'm muting must be "