tests.test_ws_subscribe_account_trades - Code Metrics - Inspection of "Merge pull request #21 from Mai-Te-Pora/tests/webs..." - Mai-Te-Pora/tradehub-python - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — main ( 30c9e1...6da681 )

by Switcheolytics

created 2021-02-18 05:28 UTC

tests.test_ws_subscribe_account_trades A

↳ Parent: Project

Complexity

Total Complexity

Size/Duplication

Total Lines	157
Duplicated Lines	91.72 %

Importance

Changes

Metric	Value
eloc	105
dl	144
loc	157
rs	10
c	0
b	0
f	0
wmc	10

2 Methods

Rating	Name	Duplication	Size	Complexity
B	TestWSSubscribeAccountTrades.test_subscribe_account_market_trades_structure()	72	72	5
B	TestWSSubscribeAccountTrades.test_subscribe_account_trades_structure()	72	72	5

How to fix Duplicated Code

import asyncio
import concurrent
from typing import Optional, List

from tests import APITestCase, DEVEL_AND_CO_SENTRY, WALLET_SWTH_ETH1_AMM, WEBSOCKET_TIMEOUT_SUBSCRIPTION
from tradehub.websocket_client import DemexWebsocket


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):

        """
        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)




1		import asyncio
2		import concurrent
3		from typing import Optional, List
4
5		from tests import APITestCase, DEVEL_AND_CO_SENTRY, WALLET_SWTH_ETH1_AMM, WEBSOCKET_TIMEOUT_SUBSCRIPTION
6		from tradehub.websocket_client import DemexWebsocket
7
8
9		class TestWSSubscribeAccountTrades(APITestCase):
10
11	View Code Duplication	def test_subscribe_account_trades_structure(self):
		0 ignored issues – show Duplication introduced 2021-02-17 15:47 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
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	View Code Duplication	def test_subscribe_account_market_trades_structure(self):
		0 ignored issues – show Duplication introduced 2021-02-17 15:47 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
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

Mai-Te-Pora / tradehub-python

Push — main ( 30c9e1...6da681 )

tests.test_ws_subscribe_account_trades A

Complexity

Size/Duplication

Importance

2 Methods

How to fix Duplicated Code

Duplicated Code

Duplication Side-by-Side

Filter issues like