From 06cc8f2fdaf04dd6cad38c98700477b4ff9c7442 Mon Sep 17 00:00:00 2001 From: Lev Gorodetskiy Date: Tue, 21 Apr 2020 18:33:33 +0300 Subject: [PATCH] More tests for slskmessages --- pynicotine/slskmessages.py | 19 ++- tests/test_pynicotine/test_slskmessages.py | 151 +++++++++++++++++++++ 2 files changed, 164 insertions(+), 6 deletions(-) diff --git a/pynicotine/slskmessages.py b/pynicotine/slskmessages.py index d1864543b64c..74f85585416a 100644 --- a/pynicotine/slskmessages.py +++ b/pynicotine/slskmessages.py @@ -357,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__,)) @@ -367,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) @@ -430,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): @@ -440,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: @@ -493,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) @@ -568,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): @@ -580,7 +587,7 @@ 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): @@ -588,7 +595,7 @@ def parseNetworkMessage(self, message): 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): diff --git a/tests/test_pynicotine/test_slskmessages.py b/tests/test_pynicotine/test_slskmessages.py index 4a494d0411c5..7176c1619ed5 100644 --- a/tests/test_pynicotine/test_slskmessages.py +++ b/tests/test_pynicotine/test_slskmessages.py @@ -1,11 +1,21 @@ import unittest +from pynicotine.slskmessages import AddUser +from pynicotine.slskmessages import ChangePassword +from pynicotine.slskmessages import GetPeerAddress +from pynicotine.slskmessages import GetUserStatus from pynicotine.slskmessages import JustDecoded +from pynicotine.slskmessages import Login from pynicotine.slskmessages import NetworkIntType from pynicotine.slskmessages import NetworkLongLongType from pynicotine.slskmessages import NetworkSignedIntType +from pynicotine.slskmessages import NotifyPrivileges +from pynicotine.slskmessages import RemoveUser +from pynicotine.slskmessages import SetStatus +from pynicotine.slskmessages import SetWaitPort from pynicotine.slskmessages import SlskMessage from pynicotine.slskmessages import ToBeEncoded +from pynicotine.slskmessages import Unknown6 class ToBeEncodedTest(unittest.TestCase): @@ -78,3 +88,144 @@ def test_packObject(self): self.assertEqual(b'{\x00\x00\x00', net_int_message) self.assertEqual(b'{\x00\x00\x00', net_signed_int_message) self.assertEqual(b'{\x00\x00\x00\x00\x00\x00\x00', net_long_long_message) + + +class LoginMessageTest(unittest.TestCase): + def test_makeNetworkMessage(self): + # Arrange + obj = Login(username='test', passwd='s33cr3t', version=157) + + # Act + message = obj.makeNetworkMessage() + + # Assert + self.assertEqual( + b'\x04\x00\x00\x00test\x07\x00\x00\x00s33cr3t\x9d\x00\x00\x00 \x00\x00\x00d' + b'bc93f24d8f3f109deed23c3e2f8b74c\x11\x00\x00\x00', + message) + + +class ChangePasswordMessageTest(unittest.TestCase): + def test_makeNetworkMessage(self): + # Arrange + obj = ChangePassword(password='s33cr3t') + + # Act + message = obj.makeNetworkMessage() + + # Assert + self.assertEqual( + b'\x07\x00\x00\x00s33cr3t', + message) + + +class SetWaitPortMessageTest(unittest.TestCase): + def test_makeNetworkMessage(self): + # Arrange + obj = SetWaitPort(port=1337) + + # Act + message = obj.makeNetworkMessage() + + # Assert + self.assertEqual( + b'9\x05\x00\x00', + message) + + +class GetPeerAddressMessageTest(unittest.TestCase): + def test_makeNetworkMessage(self): + # Arrange + obj = GetPeerAddress(user='test') + + # Act + message = obj.makeNetworkMessage() + + # Assert + self.assertEqual( + b'\x04\x00\x00\x00test', + message) + + +class AddUserMessageTest(unittest.TestCase): + def test_makeNetworkMessage(self): + # Arrange + obj = AddUser(user='test') + + # Act + message = obj.makeNetworkMessage() + + # Assert + self.assertEqual( + b'\x04\x00\x00\x00test', + message) + + +class Unknown6MessageTest(unittest.TestCase): + def test_makeNetworkMessage(self): + # Arrange + obj = Unknown6(user='test') + + # Act + message = obj.makeNetworkMessage() + + # Assert + self.assertEqual( + b'\x04\x00\x00\x00test', + message) + + +class RemoveUserMessageTest(unittest.TestCase): + def test_makeNetworkMessage(self): + # Arrange + obj = RemoveUser(user='test') + + # Act + message = obj.makeNetworkMessage() + + # Assert + self.assertEqual( + b'\x04\x00\x00\x00test', + message) + + +class GetUserStatusMessageTest(unittest.TestCase): + def test_makeNetworkMessage(self): + # Arrange + obj = GetUserStatus(user='test') + + # Act + message = obj.makeNetworkMessage() + + # Assert + self.assertEqual( + b'\x04\x00\x00\x00test', + message) + + +class SetStatusMessageTest(unittest.TestCase): + def test_makeNetworkMessage(self): + # Arrange + obj = SetStatus(status=1) + + # Act + message = obj.makeNetworkMessage() + + # Assert + self.assertEqual( + b'\x01\x00\x00\x00', + message) + + +class NotifyPrivilegesMessageTest(unittest.TestCase): + def test_makeNetworkMessage(self): + # Arrange + obj = NotifyPrivileges(token=123456, user='test') + + # Act + message = obj.makeNetworkMessage() + + # Assert + self.assertEqual( + b'@\xe2\x01\x00\x04\x00\x00\x00test', + message)