test case for testing old knownnodes pickle

-test case for testing old knownnodes pickle

-fixed CQ for tests.core module

-implemented shutil.copyfile feature and use small size knownnodes.dat file in test_pattern dir

-remove unnecessary code

-Fixed CQ of tests.core module
This commit is contained in:
navjot 2020-12-19 14:18:58 +05:30
parent d2a340d012
commit 9fe4ad0489
No known key found for this signature in database
GPG Key ID: 9EE70AFD71357F1C
3 changed files with 122 additions and 1 deletions

View File

@ -17,6 +17,8 @@ import state
from bmconfigparser import BMConfigParser from bmconfigparser import BMConfigParser
from network.node import Peer from network.node import Peer
state.Peer = Peer
knownNodesLock = threading.RLock() knownNodesLock = threading.RLock()
"""Thread lock for knownnodes modification""" """Thread lock for knownnodes modification"""
knownNodes = {stream: {} for stream in range(1, 4)} knownNodes = {stream: {} for stream in range(1, 4)}

View File

@ -7,6 +7,7 @@ import os
import pickle # nosec import pickle # nosec
import Queue import Queue
import random # nosec import random # nosec
import shutil
import string import string
import sys import sys
import time import time
@ -231,7 +232,8 @@ class TestCore(unittest.TestCase):
msg = protocol.assembleVersionMessage('127.0.0.1', 8444, [1]) msg = protocol.assembleVersionMessage('127.0.0.1', 8444, [1])
decoded = self._decode_msg(msg, "IQQiiQlsLv") decoded = self._decode_msg(msg, "IQQiiQlsLv")
peer, _, ua, streams = self._decode_msg(msg, "IQQiiQlsLv")[4:] peer, _, ua, streams = self._decode_msg(msg, "IQQiiQlsLv")[4:]
self.assertEqual(peer, Node(3, '127.0.0.1', 8444)) self.assertEqual(
peer, Node(11 if state.dandelion else 3, '127.0.0.1', 8444))
self.assertEqual(ua, '/PyBitmessage:' + softwareVersion + '/') self.assertEqual(ua, '/PyBitmessage:' + softwareVersion + '/')
self.assertEqual(streams, [1]) self.assertEqual(streams, [1])
# with multiple streams # with multiple streams
@ -259,6 +261,19 @@ class TestCore(unittest.TestCase):
'''select typeof(msgid) from sent where ackdata=?''', result) '''select typeof(msgid) from sent where ackdata=?''', result)
self.assertEqual(column_type[0][0] if column_type else '', 'text') self.assertEqual(column_type[0][0] if column_type else '', 'text')
def test_old_knownnodes_pickle(self):
"""Testing old(v.0.6.2) version knownnodes.dat file"""
try:
old_source_file = os.path.join(
os.path.abspath(os.path.dirname(__file__)), 'test_pattern', 'knownnodes.dat')
new_destination_file = os.path.join(state.appdata, 'knownnodes.dat')
shutil.copyfile(old_source_file, new_destination_file)
knownnodes.readKnownNodes()
except AttributeError as e:
self.fail('Failed to load knownnodes: %s' % e)
finally:
cleanup(files=('knownnodes.dat',))
def run(): def run():
"""Starts all tests defined in this module""" """Starts all tests defined in this module"""

View File

@ -0,0 +1,104 @@
(dp0
I1
(dp1
ccopy_reg
_reconstructor
p2
(cstate
Peer
p3
c__builtin__
tuple
p4
(S'85.180.139.241'
p5
I8444
tp6
tp7
Rp8
I1608398841
sg2
(g3
g4
(S'158.222.211.81'
p9
I8080
tp10
tp11
Rp12
I1608398841
sg2
(g3
g4
(S'178.62.12.187'
p13
I8448
tp14
tp15
Rp16
I1608398841
sg2
(g3
g4
(S'109.147.204.113'
p17
I1195
tp18
tp19
Rp20
I1608398841
sg2
(g3
g4
(S'5.45.99.75'
p21
I8444
tp22
tp23
Rp24
I1608398841
sg2
(g3
g4
(S'178.11.46.221'
p25
I8444
tp26
tp27
Rp28
I1608398841
sg2
(g3
g4
(S'95.165.168.168'
p29
I8444
tp30
tp31
Rp32
I1608398841
sg2
(g3
g4
(S'24.188.198.204'
p33
I8111
tp34
tp35
Rp36
I1608398841
sg2
(g3
g4
(S'75.167.159.54'
p37
I8444
tp38
tp39
Rp40
I1608398841
ssI2
(dp41
sI3
(dp42
s.