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

Code Duplication Length = 72-73 lines in 3 locations

tests/test_ws_subscribe_recent_trades.py 1 location



class TestWSSubscribeRecentTrades(APITestCase):

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

        expect: dict = {
            'channel': str,
            'sequence_number': 812,
            'result': [
                {
                    'id': str,
                    'block_created_at': str,
                    'taker_id': str,
                    'taker_address': str,
                    'taker_fee_amount': str,
                    'taker_fee_denom': str,
                    'taker_side': str,
                    'maker_id': str,
                    'maker_address': str,
                    'maker_fee_amount': str,
                    'maker_fee_denom': str,
                    'maker_side': str,
                    'market': str,
                    'price': str,
                    'quantity': str,
                    'liquidation': str,
                    'taker_username': str,
                    'maker_username': str,
                    'block_height': 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():
            # use AMM to be sure deterministic of which tokens the wallet holds
            await client.subscribe_recent_trades('orders', "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 asyncio.TimeoutError:
            loop = asyncio.get_event_loop()
            loop.run_until_complete(client.disconnect())

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

        if len(self.response) < 2:
            self.skipTest(f"Did not receive orders within time, test can not finish.")

        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_account_trades.py 2 locations


            self.assertDictStructure(expect, message)


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

        expect: dict = {
            "id": str,
            "sequence_number": int,
            "result": [
                {
                    "id": str,
                    "block_created_at": str,
                    "taker_id": str,
                    "taker_address": str,
                    "taker_fee_amount": str,
                    "taker_fee_denom": str,
                    "taker_side": str,
                    "maker_id": str,
                    "maker_address": str,
                    "maker_fee_amount": str,
                    "maker_fee_denom": str,
                    "maker_side": str,
                    "market": str,
                    "price": str,
                    "quantity": str,
                    "liquidation": str,
                    "taker_username": str,
                    "maker_username": str,
                    "block_height": 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():
            # use AMM to be sure deterministic of which tokens the wallet holds
            await client.subscribe_account_trades('balance', WALLET_SWTH_ETH1_AMM, "swth_eth1")

        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 asyncio.TimeoutError:
            loop = asyncio.get_event_loop()
            loop.run_until_complete(client.disconnect())

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

        if len(self.response) < 2:
            self.skipTest(f"Did not receive orders within time, test can not finish.")

        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)





class TestWSSubscribeAccountTrades(APITestCase):

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

        expect: dict = {
            "id": str,
            "sequence_number": int,
            "result": [
                {
                    "id": str,
                    "block_created_at": str,
                    "taker_id": str,
                    "taker_address": str,
                    "taker_fee_amount": str,
                    "taker_fee_denom": str,
                    "taker_side": str,
                    "maker_id": str,
                    "maker_address": str,
                    "maker_fee_amount": str,
                    "maker_fee_denom": str,
                    "maker_side": str,
                    "market": str,
                    "price": str,
                    "quantity": str,
                    "liquidation": str,
                    "taker_username": str,
                    "maker_username": str,
                    "block_height": 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():
            # use AMM to be sure deterministic of which tokens the wallet holds
            await client.subscribe_account_trades('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 asyncio.TimeoutError:
            loop = asyncio.get_event_loop()
            loop.run_until_complete(client.disconnect())

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

        if len(self.response) < 2:
            self.skipTest(f"Did not receive orders within time, test can not finish.")

        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)


    def test_subscribe_account_market_trades_structure(self):

		@@ 11-83 (lines=73) @@
8
9		class TestWSSubscribeRecentTrades(APITestCase):
10
11		def test_subscribe_recent_trades_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': 812,
24		'result': [
25		{
26		'id': str,
27		'block_created_at': str,
28		'taker_id': str,
29		'taker_address': str,
30		'taker_fee_amount': str,
31		'taker_fee_denom': str,
32		'taker_side': str,
33		'maker_id': str,
34		'maker_address': str,
35		'maker_fee_amount': str,
36		'maker_fee_denom': str,
37		'maker_side': str,
38		'market': str,
39		'price': str,
40		'quantity': str,
41		'liquidation': str,
42		'taker_username': str,
43		'maker_username': str,
44		'block_height': str,
45		}
46		]
47		}
48
49
50		# connect to websocket
51		client = DemexWebsocket(f"ws://{DEVEL_AND_CO_SENTRY}:5000/ws")
52		# little work around to save the response
53		self.response: List[dict] = []
54
55		async def on_connect():
56		# use AMM to be sure deterministic of which tokens the wallet holds
57		await client.subscribe_recent_trades('orders', "eth1_usdc1")
58
59		async def on_message(message: dict):
60		# save response into self
61		self.response.append(message)
62
63		try:
64		loop = asyncio.get_event_loop()
65		loop.run_until_complete(asyncio.wait_for(client.connect(on_connect_callback=on_connect,
66		on_receive_message_callback=on_message),
67		WEBSOCKET_TIMEOUT_SUBSCRIPTION))
68		except asyncio.TimeoutError:
69		loop = asyncio.get_event_loop()
70		loop.run_until_complete(client.disconnect())
71
72		if not self.response:
73		raise RuntimeError("Did not receive a response.")
74
75		if len(self.response) < 2:
76		self.skipTest(f"Did not receive orders within time, test can not finish.")
77
78		channel_subscription: dict = self.response[0]
79		self.assertDictStructure(expect_subscription, channel_subscription)
80
81		for message in self.response[1:]:
82		# if this fails, check if the AMM wallet own other tokens as expected
83		self.assertDictStructure(expect, message)

		@@ 85-156 (lines=72) @@
82		self.assertDictStructure(expect, message)
83
84
85		def test_subscribe_account_market_trades_structure(self):
86		"""
87		Check if response match expected dict structure.
88		:return:
89		"""
90		expect_subscription: dict = {
91		'id': str,
92		'result': [str]
93		}
94
95		expect: dict = {
96		"id": str,
97		"sequence_number": int,
98		"result": [
99		{
100		"id": str,
101		"block_created_at": str,
102		"taker_id": str,
103		"taker_address": str,
104		"taker_fee_amount": str,
105		"taker_fee_denom": str,
106		"taker_side": str,
107		"maker_id": str,
108		"maker_address": str,
109		"maker_fee_amount": str,
110		"maker_fee_denom": str,
111		"maker_side": str,
112		"market": str,
113		"price": str,
114		"quantity": str,
115		"liquidation": str,
116		"taker_username": str,
117		"maker_username": str,
118		"block_height": str
119		},
120		]
121		}
122
123		# connect to websocket
124		client = DemexWebsocket(f"ws://{DEVEL_AND_CO_SENTRY}:5000/ws")
125		# little work around to save the response
126		self.response: List[dict] = []
127
128		async def on_connect():
129		# use AMM to be sure deterministic of which tokens the wallet holds
130		await client.subscribe_account_trades('balance', WALLET_SWTH_ETH1_AMM, "swth_eth1")
131
132		async def on_message(message: dict):
133		# save response into self
134		self.response.append(message)
135
136		try:
137		loop = asyncio.get_event_loop()
138		loop.run_until_complete(asyncio.wait_for(client.connect(on_connect_callback=on_connect,
139		on_receive_message_callback=on_message),
140		WEBSOCKET_TIMEOUT_SUBSCRIPTION))
141		except asyncio.TimeoutError:
142		loop = asyncio.get_event_loop()
143		loop.run_until_complete(client.disconnect())
144
145		if not self.response:
146		raise RuntimeError("Did not receive a response.")
147
148		if len(self.response) < 2:
149		self.skipTest(f"Did not receive orders within time, test can not finish.")
150
151		channel_subscription: dict = self.response[0]
152		self.assertDictStructure(expect_subscription, channel_subscription)
153
154		for message in self.response[1:]:
155		# if this fails, check if the AMM wallet own other tokens as expected
156		self.assertDictStructure(expect, message)
157
158
159
		@@ 11-82 (lines=72) @@
8
9		class TestWSSubscribeAccountTrades(APITestCase):
10
11		def test_subscribe_account_trades_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		"id": str,
23		"sequence_number": int,
24		"result": [
25		{
26		"id": str,
27		"block_created_at": str,
28		"taker_id": str,
29		"taker_address": str,
30		"taker_fee_amount": str,
31		"taker_fee_denom": str,
32		"taker_side": str,
33		"maker_id": str,
34		"maker_address": str,
35		"maker_fee_amount": str,
36		"maker_fee_denom": str,
37		"maker_side": str,
38		"market": str,
39		"price": str,
40		"quantity": str,
41		"liquidation": str,
42		"taker_username": str,
43		"maker_username": str,
44		"block_height": 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: List[dict] = []
53
54		async def on_connect():
55		# use AMM to be sure deterministic of which tokens the wallet holds
56		await client.subscribe_account_trades('balance', WALLET_SWTH_ETH1_AMM)
57
58		async def on_message(message: dict):
59		# save response into self
60		self.response.append(message)
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_SUBSCRIPTION))
67		except asyncio.TimeoutError:
68		loop = asyncio.get_event_loop()
69		loop.run_until_complete(client.disconnect())
70
71		if not self.response:
72		raise RuntimeError("Did not receive a response.")
73
74		if len(self.response) < 2:
75		self.skipTest(f"Did not receive orders within time, test can not finish.")
76
77		channel_subscription: dict = self.response[0]
78		self.assertDictStructure(expect_subscription, channel_subscription)
79
80		for message in self.response[1:]:
81		# if this fails, check if the AMM wallet own other tokens as expected
82		self.assertDictStructure(expect, message)
83
84
85		def test_subscribe_account_market_trades_structure(self):

Mai-Te-Pora / tradehub-python

Code Duplication Length = 72-73 lines in 3 locations

tests/test_ws_subscribe_recent_trades.py 1 location

tests/test_ws_subscribe_account_trades.py 2 locations