WIP: Initial tests for slskmessages

pynicotine/slskmessages.py

@@ -28,6 +28,7 @@
 import zlib
 from gettext import gettext as _
 from itertools import count
+from typing import Optional
 from typing import Union
 
 from pynicotine.logfacility import log
@@ -65,7 +66,7 @@ def _str(arg: Union[bytes, str]) -> str:
 
 class NetworkBaseType:
     """Base class for other network types."""
-    def __init__(self, value):
+    def __init__(self, value: int) -> None:
         self.value = value
 
 
@@ -91,65 +92,54 @@ class NetworkLongLongType(NetworkBaseType):
 
 class ToBeEncoded:
     """Holds text and the desired eventual encoding"""
-    def __init__(self, uni, encoding):
-        if not isinstance(uni, str):
-            raise ValueError("ZOMG, you really don't know what you're doing! %s is NOT unicode, its a %s: %s" % (uni, type(uni), repr(uni)))
-        self.str = uni
-        self.encoding = encoding
-        self.cached = None
-
-    def getbytes(self):
-        if self.cached:
-            return self.cached
-        self.cached = self.str.encode(self.encoding, "replace")
-        # print "The bytes of %s are %s" % (self.unicode, repr(self.cached))
-        return self.cached
-
-    def dont(self):
-        print("Dont do that")
-    bytes = property(getbytes, dont)
-
-    def __getitem__(self, key):
-        return self.str[key]
+    def __init__(self, string: str, encoding: str) -> None:
+        self._string = string
+        self._encoding = encoding
+        self._bytes: Optional[bytes] = None
 
-    def __str__(self):
-        return "%s" % (self.getbytes(),)
+    @property
+    def bytes(self) -> bytes:
+        if self._bytes is None:
+            self._bytes = self._string.encode(self._encoding, "replace")
+        return self._bytes
 
-    def __repr__(self):
-        return "ToBeEncoded(%s, %s)" % (repr(self.getbytes()), self.encoding)
+    def __getitem__(self, key: int) -> str:
+        return self._string[key]
+
+    def __str__(self) -> str:
+        return str(self.bytes)
+
+    def __repr__(self) -> str:
+        return f"ToBeEncoded({self.bytes}, {self._encoding})"
 
 
 class JustDecoded:
     """Holds text, the original bytes and its supposed encoding"""
-    def __init__(self, bytes, encoding):
-        if not isinstance(bytes, str):
-            raise ValueError("ZOMG, you really don't know what you're doing! %s is NOT string, its a %s: %s" % (bytes, type(bytes), repr(bytes)))
-        self.bytes = bytes
-        self.encoding = encoding
-        self.cached = None
-        self.modified = False
-
-    def getunicode(self):
-        if self.cached:
-            return self.cached
-        self.cached = self.bytes.decode(self.encoding, "replace")
-        return self.cached
-
-    def setunicode(self, uni):
-        if not isinstance(uni, str):
-            print("ZOMG, you really don't know what you're doing! %s is NOT unicode, its a %s: %s" % (uni, type(uni), repr(bytes)))
-            raise Exception
-        self.cached = uni
-    str = property(getunicode, setunicode)
-
-    def __getitem__(self, key):
+    def __init__(self, bytes_: bytes, encoding: str):
+        self._bytes = bytes_
+        self._encoding = encoding
+        self._string: Optional[str] = None
+        self._modified: bool = False
+
+    @property
+    def str(self) -> str:
+        if self._string is None:
+            self._string = self._bytes.decode(self._encoding, "replace")
+        return self._string
+
+    # FIXME: Should we allow to set this property?
+    @str.setter
+    def str(self, string: str) -> None:
+        self._string = string
+
+    def __getitem__(self, key: int) -> str:
         return self.str[key]
 
-    def __str__(self):
-        return "%s" % (self.getbytes(),)
+    def __str__(self) -> str:
+        return str(self._bytes)
 
-    def __repr__(self):
-        return "ToBeEncoded(%s, %s)" % (repr(self.getbytes()), self.encoding)
+    def __repr__(self) -> str:
+        return f"JustDecoded({self._bytes}, {self._encoding})"
 
 
 class InternalMessage:
@@ -367,6 +357,13 @@ def packObject(self, object):
         log.addwarning(_("Warning: unknown object type %s") % type(object) + " " + "in message %(type)s" % {'type': self.__class__})
         return b""
 
+    def packObjects(self, *objects) -> bytes:
+        """Helper method to pack each object in sequence and concatenate resulting bytes"""
+        bytes_ = b''
+        for obj in objects:
+            bytes_ += self.packObject(obj)
+        return bytes_
+
     def makeNetworkMessage(self):
         """ Returns binary array, that can be sent over the network"""
         log.addwarning(_("Empty message made, class %s") % (self.__class__,))
@@ -377,6 +374,7 @@ def parseNetworkMessage(self, message):
         in an object"""
         log.addwarning(_("Can't parse incoming messages, class %s") % (self.__class__,))
 
+    # FIXME: Unresolved attribute reference
     def doubleParseNetworkMessage(self, message):
         """Calls self._parseNetworkMessage first with a NetworkLongLongType, if that fails with NetworkIntType."""
         messagename = str(self)
@@ -440,7 +438,7 @@ def __repr__(self):
     def makeNetworkMessage(self):
         payload = self.username + self.passwd
         md5hash = hashlib.md5(payload.encode()).hexdigest()
-        message = self.packObject(self.username) + self.packObject(self.passwd) + self.packObject(self.version) + self.packObject(md5hash) + self.packObject(17)
+        message = self.packObjects(self.username, self.passwd, self.version, md5hash, 17)
         return message
 
     def parseNetworkMessage(self, message):
@@ -450,9 +448,8 @@ def parseNetworkMessage(self, message):
 
         else:
             pos, self.banner = self.getObject(message, bytes, pos)
-        if len(message[pos:]) > 0:
+        if message[pos:]:
             try:
-                import socket
                 pos, self.ip = pos + 4, socket.inet_ntoa(message[pos:pos + 4][::-1])
                 # Unknown number
             except Exception as error:
@@ -503,7 +500,6 @@ def makeNetworkMessage(self):
 
     def parseNetworkMessage(self, message):
         pos, self.user = self.getObject(message, bytes)
-        import socket
         pos, self.ip = pos + 4, socket.inet_ntoa(message[pos:pos + 4][::-1])
         pos, self.port = self.getObject(message, int, pos, 1)
 
@@ -578,6 +574,7 @@ def parseNetworkMessage(self, message):
             pass
 
 
+# NOTE: int attribute converted to string
 class SetStatus(ServerMessage):
     """ We send our new status to the server """
     def __init__(self, status=None):
@@ -590,15 +587,15 @@ def makeNetworkMessage(self):
 class NotifyPrivileges(ServerMessage):
     """ Server tells us something about privileges"""
     def __init__(self, token=None, user=None):
-        self.token = token
+        self.token: Optional[int] = token
         self.user = _str(user)
 
     def parseNetworkMessage(self, message):
         pos, self.token = self.getObject(message, int)
         pos, self.user = self.getObject(message, bytes, pos)
 
     def makeNetworkMessage(self):
-        return self.packObject(self.token) + self.packObject(self.user)
+        return self.packObjects(self.token, self.user)
 
 
 class AckNotifyPrivileges(ServerMessage):
@@ -645,7 +642,7 @@ def __init__(self, room=None, msg=None):
         self.msg = _str(msg)
 
     def makeNetworkMessage(self):
-        return self.packObject(self.room) + self.packObject(self.msg)
+        return self.packObjects(self.room, self.msg)
 
     def parseNetworkMessage(self, message):
         pos, self.room = self.getObject(message, bytes)
@@ -672,25 +669,26 @@ class JoinRoom(ServerMessage):
     with data on everyone."""
     def __init__(self, room=None, private=None):
         self.room = _str(room)
-        self.private = private
+        self.private: int = private
         self.owner = None
         self.operators = []
 
     def makeNetworkMessage(self):
         if self.private is not None:
-            return self.packObject(self.room) + self.packObject(self.private)
-        return self.packObject(self.room)
+            return self.packObjects(self.room, self.private)
+        return self.packObjects(self.room)
 
     def parseNetworkMessage(self, message):
         pos, self.room = self.getObject(message, bytes)
         pos1 = pos
         pos, self.users = self.getUsers(message[pos:])
         pos = pos1 + pos
 
-        if len(message[pos:]) > 0:
+        if message[pos:]:
+            # FIXME: bool or int?
             self.private = True
             pos, self.owner = self.getObject(message, bytes, pos)
-        if len(message[pos:]) > 0 and self.private:
+        if message[pos:] and self.private:
             pos, numops = self.getObject(message, int, pos)
             for i in range(numops):
                 pos, operator = self.getObject(message, bytes, pos)
@@ -765,7 +763,7 @@ def __init__(self, room=None, user=None):
         self.user = _str(user)
 
     def makeNetworkMessage(self):
-        return self.packObject(self.room) + self.packObject(self.user)
+        return self.packObjects(self.room, self.user)
 
     def parseNetworkMessage(self, message):
         pos, self.room = self.getObject(message, bytes)