Fix type errors and documentation issues

pyproject.toml

@@ -145,6 +145,8 @@ reportUnknownParameterType = "error"
 reportUnknownReturnType = "error"
 # reportUnknownArgumentType = "error"
 reportUnknownMemberType = "error"
+reportUnnecessaryComparison = true
+reportUnnecessaryContains = true
 
 [tool.pytest.ini_options]
 testpaths = ["tests"]

steam/enums.py

@@ -76,7 +76,7 @@ def __new__(mcs, name: str, bases: tuple[type, ...], namespace: dict[str, Any])
         value_map: dict[Any, Enum] = {}
         member_map: dict[str, Enum] = {}
 
-        new_mcs: type[EnumType] = type(
+        new_mcs: type[Self] = type(
             f"{name}Type",
             tuple(
                 dict.fromkeys([base.__class__ for base in bases if base.__class__ is not type] + [EnumType, type])
@@ -86,8 +86,9 @@ def __new__(mcs, name: str, bases: tuple[type, ...], namespace: dict[str, Any])
 
         members = {name: value for name, value in namespace.items() if not _is_descriptor(value) and name[0] != "_"}
 
-        cls = type.__new__(
-            new_mcs, name, bases, {key: value for key, value in namespace.items() if key not in members}
+        cls = cast(
+            "type[Enum]",
+            type.__new__(new_mcs, name, bases, {key: value for key, value in namespace.items() if key not in members}),
         )  # this allows us to disallow member access from other members as members become proper class variables
 
         for name, value in members.items():

steam/gateway.py

@@ -643,18 +643,14 @@ def unpack_multi(msg: CMsgMulti) -> bytearray | None:
 
         if flag := data[3]:  # this isn't ever hit, might as well save a few nanos
             if flag & FEXTRA:
-                extra_len = int.from_bytes(data[position:2], byteorder="little")
+                extra_len = int.from_bytes(data[position : position + 2], byteorder="little")
                 position += 2 + extra_len
             if flag & FNAME:
-                for terminator in data[position:]:
+                while data[position:]:
                     position += 1
-                    if not terminator or terminator == b"\000":
-                        break
             if flag & FCOMMENT:
-                for terminator in data[position:]:
+                while data[position:]:
                     position += 1
-                    if not terminator or terminator == b"\000":
-                        break
             if flag & FHCRC:
                 position += 2
 

steam/manifest.py

@@ -66,7 +66,7 @@ def unzip(data: bytes) -> bytes:
     if data[:2] == b"VZ":
         if data[-2:] != b"zv":
             raise RuntimeError(f"VZ: Invalid footer: {data[-2:]!r}")
-        if data[2] != b"a":
+        if data[2] != 97:
             raise RuntimeError(f"VZ: Invalid version: {data[2]!r}")
 
         filters = (lzma._decode_filter_properties(lzma.FILTER_LZMA1, data[7:12]),)  # type: ignore
@@ -211,7 +211,7 @@ def readlink(self) -> Self:
         Raises
         ------
         OSError
-            this path isn't a symlink.
+            This path isn't a symlink.
         """
         if not self.is_symlink():
             raise OSError(errno.EINVAL, f"Invalid argument: {str(self)!r}")

steam/models.py

@@ -286,7 +286,7 @@ def __init__(self, state: ConnectionState, sha: bytes):
         sha = bytes(sha)
         self.sha = (
             sha
-            if sha != "\x00" * 20
+            if sha != b"\x00" * 20
             else b"\xfe\xf4\x9e\x7f\xa7\xe1\x99s\x10\xd7\x05\xb2\xa6\x15\x8f\xf8\xdc\x1c\xdf\xeb"
         )
         self._state = state

steam/state.py

@@ -1141,18 +1141,17 @@ async def process_friends(self, msg: friends.CMsgClientFriendsList) -> None:
         client_user_friends: list[ID64] = []
         is_load = not msg.bincremental
         for friend in msg.friends:
-            steam_id = ID(friend.ulfriendid)
+            id = ID(friend.ulfriendid)
             match relationship := FriendRelationship.try_value(friend.efriendrelationship):
                 case FriendRelationship.Friend:
                     try:
-                        invite = self.invites.pop(steam_id.id64)
+                        invite = self.invites.pop(id.id64)
                     except KeyError:
-                        print("Not in invites", steam_id)
-                        if steam_id.type == Type.Individual:
+                        if id.type == Type.Individual:
                             if is_load:
-                                client_user_friends.append(steam_id.id64)
+                                client_user_friends.append(id.id64)
                             else:
-                                user = await self.fetch_user(steam_id.id64)
+                                user = await self.fetch_user(id.id64)
                                 assert user is not None
                                 self._add_friend(user)
                     else:
@@ -1167,9 +1166,9 @@ async def process_friends(self, msg: friends.CMsgClientFriendsList) -> None:
                                 self._clans[invite.clan.id] = invite.clan
 
                 case FriendRelationship.RequestInitiator | FriendRelationship.RequestRecipient:
-                    match steam_id.type:
+                    match id.type:
                         case Type.Individual:
-                            invitee = await self._maybe_user(steam_id.id64)
+                            invitee = await self._maybe_user(id.id64)
                             invite = UserInvite(state=self, invitee=invitee, relationship=relationship)
                             self.invites[invitee.id64] = invite
                             self.dispatch("user_invite", invite)
@@ -1183,40 +1182,42 @@ async def process_friends(self, msg: friends.CMsgClientFriendsList) -> None:
                                 (
                                     utils.parse_id64(elements[idx + 1]["data-miniprofile"])
                                     for idx, element in enumerate(elements)
-                                    if str(steam_id.id64) in str(element)
+                                    if str(id.id64) in str(element)
                                 ),
                                 0,
                             )
 
                             invitee = await self._maybe_user(invitee_id64)
                             try:
-                                clan = await self.fetch_clan(steam_id.id64) or steam_id
+                                clan = await self.fetch_clan(id.id64) or id
                             except WSException:
-                                clan = steam_id
+                                clan = id
                             invite = ClanInvite(state=self, invitee=invitee, clan=clan, relationship=relationship)
                             self.invites[clan.id64] = invite
                             self.dispatch("clan_invite", invite)
 
                 case FriendRelationship.NONE:
-                    match steam_id.type:
+                    match id.type:
                         case Type.Individual:
                             try:
-                                invite = self.invites.pop(steam_id.id64)
+                                invite = self.invites.pop(id.id64)
                             except KeyError:
-                                friend = self.user._friends.pop(steam_id.id64, None)
+                                friend = self.user._friends.pop(id.id64, None)
                                 if friend is None:
-                                    return log.debug("Unknown friend %s removed", steam_id)
+                                    return log.debug("Unknown friend %s removed", id)
                                 self.dispatch("friend_remove", friend)
                             else:
-                                self.dispatch("user_invite_decline", invite)
+                                self.dispatch(
+                                    f"{'user' if id.type == Type.Individual else 'clan'}_invite_decline", invite
+                                )
 
                         case Type.Clan:
                             try:
-                                invite = self.invites.pop(steam_id.id64)
+                                invite = self.invites.pop(id.id64)
                             except KeyError:
-                                clan = self._clans.pop(steam_id.id, None)
+                                clan = self._clans.pop(id.id, None)
                                 if clan is None:
-                                    return log.debug("Unknown clan %s removed", steam_id)
+                                    return log.debug("Unknown clan %s removed", id)
                                 self.dispatch("clean_leave", clan)
                             else:
                                 self.dispatch("clan_invite_decline", invite)