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

Code Duplication Length = 65-65 lines in 2 locations

tests/test_ws_get_open_orders.py 1 location


from tradehub.websocket_client import DemexWebsocket


class TestWSGetOpenOrders(APITestCase):

    def test_get_open_orders_structure(self):
        """
        Check if response match expected dict structure.
        :return:
        """
        expect: dict = {
            "id": str,
            "result": [
                {
                    "order_id": str,
                    "block_height": int,
                    "triggered_block_height": int,
                    "address": str,
                    "market": str,
                    "side": str,
                    "price": str,
                    "quantity": str,
                    "available": str,
                    "filled": str,
                    "order_status": str,
                    "order_type": str,
                    "initiator": str,
                    "time_in_force": str,
                    "stop_price": str,
                    "trigger_type": str,
                    "allocated_margin_denom": str,
                    "allocated_margin_amount": str,
                    "is_liquidation": bool,
                    "is_post_only": bool,
                    "is_reduce_only": bool,
                    "type": str,
                    "block_created_at": str,
                    "username": str,
                    "id": str,
                }
            ]
        }

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

        async def on_connect():
            await client.get_open_orders('order_history', WALLET_SWTH_ETH1_AMM)

        async def on_message(message: dict):
            # save response into self
            self.response = message
            await client.disconnect()

        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_GET_REQUEST))
        except asyncio.TimeoutError:
            raise TimeoutError("Test did not complete in time.")

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

        self.assertDictStructure(expect, self.response)



tests/test_ws_get_order_history.py 1 location


from tradehub.websocket_client import DemexWebsocket


class TestWSGetOrderHistory(APITestCase):

    def test_get_order_history_structure(self):
        """
        Check if response match expected dict structure.
        :return:
        """
        expect: dict = {
            "id": str,
            "result": [
                {
                    "order_id": str,
                    "block_height": int,
                    "triggered_block_height": int,
                    "address": str,
                    "market": str,
                    "side": str,
                    "price": str,
                    "quantity": str,
                    "available": str,
                    "filled": str,
                    "order_status": str,
                    "order_type": str,
                    "initiator": str,
                    "time_in_force": str,
                    "stop_price": str,
                    "trigger_type": str,
                    "allocated_margin_denom": str,
                    "allocated_margin_amount": str,
                    "is_liquidation": bool,
                    "is_post_only": bool,
                    "is_reduce_only": bool,
                    "type": str,
                    "block_created_at": str,
                    "username": str,
                    "id": str,
                }
            ]
        }

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

        async def on_connect():
            await client.get_order_history('order_history', WALLET_DEVEL)

        async def on_message(message: dict):
            # save response into self
            self.response = message
            await client.disconnect()

        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_GET_REQUEST))
        except asyncio.TimeoutError:
            raise TimeoutError("Test did not complete in time.")

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

        self.assertDictStructure(expect, self.response)



		@@ 9-73 (lines=65) @@
6		from tradehub.websocket_client import DemexWebsocket
7
8
9		class TestWSGetOpenOrders(APITestCase):
10
11		def test_get_open_orders_structure(self):
12		"""
13		Check if response match expected dict structure.
14		:return:
15		"""
16		expect: dict = {
17		"id": str,
18		"result": [
19		{
20		"order_id": str,
21		"block_height": int,
22		"triggered_block_height": int,
23		"address": str,
24		"market": str,
25		"side": str,
26		"price": str,
27		"quantity": str,
28		"available": str,
29		"filled": str,
30		"order_status": str,
31		"order_type": str,
32		"initiator": str,
33		"time_in_force": str,
34		"stop_price": str,
35		"trigger_type": str,
36		"allocated_margin_denom": str,
37		"allocated_margin_amount": str,
38		"is_liquidation": bool,
39		"is_post_only": bool,
40		"is_reduce_only": bool,
41		"type": str,
42		"block_created_at": str,
43		"username": str,
44		"id": str,
45		}
46		]
47		}
48
49		# connect to websocket
50		client = DemexWebsocket(f"ws://{DEVEL_AND_CO_SENTRY}:5000/ws")
51		# little work around to save the response
52		self.response: Optional[dict] = None
53
54		async def on_connect():
55		await client.get_open_orders('order_history', WALLET_SWTH_ETH1_AMM)
56
57		async def on_message(message: dict):
58		# save response into self
59		self.response = message
60		await client.disconnect()
61
62		try:
63		loop = asyncio.get_event_loop()
64		loop.run_until_complete(asyncio.wait_for(client.connect(on_connect_callback=on_connect,
65		on_receive_message_callback=on_message),
66		WEBSOCKET_TIMEOUT_GET_REQUEST))
67		except asyncio.TimeoutError:
68		raise TimeoutError("Test did not complete in time.")
69
70		if not self.response:
71		raise RuntimeError("Did not receive a response.")
72
73		self.assertDictStructure(expect, self.response)
74
75

		@@ 9-73 (lines=65) @@
6		from tradehub.websocket_client import DemexWebsocket
7
8
9		class TestWSGetOrderHistory(APITestCase):
10
11		def test_get_order_history_structure(self):
12		"""
13		Check if response match expected dict structure.
14		:return:
15		"""
16		expect: dict = {
17		"id": str,
18		"result": [
19		{
20		"order_id": str,
21		"block_height": int,
22		"triggered_block_height": int,
23		"address": str,
24		"market": str,
25		"side": str,
26		"price": str,
27		"quantity": str,
28		"available": str,
29		"filled": str,
30		"order_status": str,
31		"order_type": str,
32		"initiator": str,
33		"time_in_force": str,
34		"stop_price": str,
35		"trigger_type": str,
36		"allocated_margin_denom": str,
37		"allocated_margin_amount": str,
38		"is_liquidation": bool,
39		"is_post_only": bool,
40		"is_reduce_only": bool,
41		"type": str,
42		"block_created_at": str,
43		"username": str,
44		"id": str,
45		}
46		]
47		}
48
49		# connect to websocket
50		client = DemexWebsocket(f"ws://{DEVEL_AND_CO_SENTRY}:5000/ws")
51		# little work around to save the response
52		self.response: Optional[dict] = None
53
54		async def on_connect():
55		await client.get_order_history('order_history', WALLET_DEVEL)
56
57		async def on_message(message: dict):
58		# save response into self
59		self.response = message
60		await client.disconnect()
61
62		try:
63		loop = asyncio.get_event_loop()
64		loop.run_until_complete(asyncio.wait_for(client.connect(on_connect_callback=on_connect,
65		on_receive_message_callback=on_message),
66		WEBSOCKET_TIMEOUT_GET_REQUEST))
67		except asyncio.TimeoutError:
68		raise TimeoutError("Test did not complete in time.")
69
70		if not self.response:
71		raise RuntimeError("Did not receive a response.")
72
73		self.assertDictStructure(expect, self.response)
74
75

Mai-Te-Pora / tradehub-python

Code Duplication Length = 65-65 lines in 2 locations

tests/test_ws_get_open_orders.py 1 location

tests/test_ws_get_order_history.py 1 location