Code Duplication - Mai-Te-Pora/tradehub-python - Measure and Improve Code Quality continuously with Scrutinizer

Code Duplication Length = 57-61 lines in 2 locations

tests/test_ws_subscribe_balance.py 1 location


from tradehub.websocket_client import DemexWebsocket


class TestWSSubscribeBalance(APITestCase):

    def test_subscribe_balance_structure(self):
        """
        Check if response match expected dict structure.
        :return:
        """
        expect_subscription: dict = {
            'id': str,
            'result': [str]
        }

        balance: dict = {
            'available': str,
            'order': str,
            'position': str,
            'denom': str
        }

        expect: dict = {
            'channel': str,
            'result': {
                'eth1': balance,
                'eth1-80-swth-20-lp1': balance,
                'swth': balance
            }
        }

        # connect to websocket
        client = DemexWebsocket(f"ws://{DEVEL_AND_CO_SENTRY}:5000/ws")
        # little work around to save the response
        self.response: List[dict] = []

        async def on_connect():
            # use AMM to be sure deterministic of which tokens the wallet holds
            await client.subscribe_balances('balance', WALLET_SWTH_ETH1_AMM)

        async def on_message(message: dict):
            # save response into self
            self.response.append(message)

        try:
            loop = asyncio.get_event_loop()
            loop.run_until_complete(asyncio.wait_for(client.connect(on_connect_callback=on_connect,
                                                                    on_receive_message_callback=on_message),
                                                     WEBSOCKET_TIMEOUT_SUBSCRIPTION))
        except concurrent.futures._base.TimeoutError:
            loop = asyncio.get_event_loop()
            loop.run_until_complete(client.disconnect())

        if not self.response:
            raise RuntimeError("Did not receive a response.")

        self.assertTrue(len(self.response) >= 2, msg=f"Expected at least 2 messages: channel subscription and and first update message")

        channel_subscription: dict = self.response[0]
        self.assertDictStructure(expect_subscription, channel_subscription)

        for message in self.response[1:]:
            # if this fails, check if the AMM wallet own other tokens as expected
            self.assertDictStructure(expect, message)



tests/test_ws_subscribe_books.py 1 location


from tradehub.websocket_client import DemexWebsocket


class TestWSSubscribeBooks(APITestCase):

    def test_subscribe_books_structure(self):
        """
        Check if response match expected dict structure.
        :return:
        """
        expect_subscription: dict = {
            'id': str,
            'result': [str]
        }

        expect: dict = {
            'channel': str,
            'sequence_number': int,
            'result': [
                {
                    'market': str,
                    'price': str,
                    'quantity': str,
                    'side': str,
                    'type': str
                }
            ]
        }

        # connect to websocket
        client = DemexWebsocket(f"ws://{DEVEL_AND_CO_SENTRY}:5000/ws")
        # little work around to save the response
        self.response: List[dict] = []

        async def on_connect():
            await client.subscribe_books('book', "eth1_usdc1")

        async def on_message(message: dict):
            # save response into self
            self.response.append(message)

        try:
            loop = asyncio.get_event_loop()
            loop.run_until_complete(asyncio.wait_for(client.connect(on_connect_callback=on_connect,
                                                                    on_receive_message_callback=on_message),
                                                     WEBSOCKET_TIMEOUT_SUBSCRIPTION))
        except concurrent.futures._base.TimeoutError:
            loop = asyncio.get_event_loop()
            loop.run_until_complete(client.disconnect())

        if not self.response:
            raise RuntimeError("Did not receive a response.")

        self.assertTrue(len(self.response) >= 2, msg=f"Expected at least 2 messages: channel subscription and an update message")

        channel_subscription: dict = self.response[0]
        self.assertDictStructure(expect_subscription, channel_subscription)

        for message in self.response[1:]:
            self.assertDictStructure(expect, message)



		@@ 9-69 (lines=61) @@
6		from tradehub.websocket_client import DemexWebsocket
7
8
9		class TestWSSubscribeBalance(APITestCase):
10
11		def test_subscribe_balance_structure(self):
12		"""
13		Check if response match expected dict structure.
14		:return:
15		"""
16		expect_subscription: dict = {
17		'id': str,
18		'result': [str]
19		}
20
21		balance: dict = {
22		'available': str,
23		'order': str,
24		'position': str,
25		'denom': str
26		}
27
28		expect: dict = {
29		'channel': str,
30		'result': {
31		'eth1': balance,
32		'eth1-80-swth-20-lp1': balance,
33		'swth': balance
34		}
35		}
36
37		# connect to websocket
38		client = DemexWebsocket(f"ws://{DEVEL_AND_CO_SENTRY}:5000/ws")
39		# little work around to save the response
40		self.response: List[dict] = []
41
42		async def on_connect():
43		# use AMM to be sure deterministic of which tokens the wallet holds
44		await client.subscribe_balances('balance', WALLET_SWTH_ETH1_AMM)
45
46		async def on_message(message: dict):
47		# save response into self
48		self.response.append(message)
49
50		try:
51		loop = asyncio.get_event_loop()
52		loop.run_until_complete(asyncio.wait_for(client.connect(on_connect_callback=on_connect,
53		on_receive_message_callback=on_message),
54		WEBSOCKET_TIMEOUT_SUBSCRIPTION))
55		except concurrent.futures._base.TimeoutError:
56		loop = asyncio.get_event_loop()
57		loop.run_until_complete(client.disconnect())
58
59		if not self.response:
60		raise RuntimeError("Did not receive a response.")
61
62		self.assertTrue(len(self.response) >= 2, msg=f"Expected at least 2 messages: channel subscription and and first update message")
63
64		channel_subscription: dict = self.response[0]
65		self.assertDictStructure(expect_subscription, channel_subscription)
66
67		for message in self.response[1:]:
68		# if this fails, check if the AMM wallet own other tokens as expected
69		self.assertDictStructure(expect, message)
70
71

		@@ 9-65 (lines=57) @@
6		from tradehub.websocket_client import DemexWebsocket
7
8
9		class TestWSSubscribeBooks(APITestCase):
10
11		def test_subscribe_books_structure(self):
12		"""
13		Check if response match expected dict structure.
14		:return:
15		"""
16		expect_subscription: dict = {
17		'id': str,
18		'result': [str]
19		}
20
21		expect: dict = {
22		'channel': str,
23		'sequence_number': int,
24		'result': [
25		{
26		'market': str,
27		'price': str,
28		'quantity': str,
29		'side': str,
30		'type': str
31		}
32		]
33		}
34
35		# connect to websocket
36		client = DemexWebsocket(f"ws://{DEVEL_AND_CO_SENTRY}:5000/ws")
37		# little work around to save the response
38		self.response: List[dict] = []
39
40		async def on_connect():
41		await client.subscribe_books('book', "eth1_usdc1")
42
43		async def on_message(message: dict):
44		# save response into self
45		self.response.append(message)
46
47		try:
48		loop = asyncio.get_event_loop()
49		loop.run_until_complete(asyncio.wait_for(client.connect(on_connect_callback=on_connect,
50		on_receive_message_callback=on_message),
51		WEBSOCKET_TIMEOUT_SUBSCRIPTION))
52		except concurrent.futures._base.TimeoutError:
53		loop = asyncio.get_event_loop()
54		loop.run_until_complete(client.disconnect())
55
56		if not self.response:
57		raise RuntimeError("Did not receive a response.")
58
59		self.assertTrue(len(self.response) >= 2, msg=f"Expected at least 2 messages: channel subscription and an update message")
60
61		channel_subscription: dict = self.response[0]
62		self.assertDictStructure(expect_subscription, channel_subscription)
63
64		for message in self.response[1:]:
65		self.assertDictStructure(expect, message)
66
67

Mai-Te-Pora / tradehub-python

Code Duplication Length = 57-61 lines in 2 locations

tests/test_ws_subscribe_balance.py 1 location

tests/test_ws_subscribe_books.py 1 location