Coverage for tests/admin/test_rsk_client.py: 100%

1# The MIT License (MIT) 

2# 

3# Copyright (c) 2021 RSK Labs Ltd 

4# 

5# Permission is hereby granted, free of charge, to any person obtaining a copy of 

6# this software and associated documentation files (the "Software"), to deal in 

7# the Software without restriction, including without limitation the rights to 

8# use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies 

9# of the Software, and to permit persons to whom the Software is furnished to do 

10# so, subject to the following conditions: 

11# 

12# The above copyright notice and this permission notice shall be included in all 

13# copies or substantial portions of the Software. 

14# 

15# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR 

16# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, 

17# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE 

18# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER 

19# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, 

20# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE 

21# SOFTWARE. 

22 

23from types import SimpleNamespace 

24from unittest import TestCase 

25from unittest.mock import call, patch 

26from admin.rsk_client import RskClient, RskClientError 

27 

28import json 

29 

30 

31@patch("requests.post") 

32class TestRskClient(TestCase): 

33 def setUp(self): 

34 self.force_wrong_id = False 

35 

36 def generate_post_response(self, url, headers, data): 

37 self.request_id = json.loads(data)['id'] 

38 if self.force_wrong_id: 

39 self.request_id = self.request_id + 1 

40 response = SimpleNamespace() 

41 response.status_code = self.status_code 

42 response.text = json.dumps({ 

43 'id': self.request_id, 

44 'result': 'aa' * 65 

45 }) 

46 

47 return response 

48 

49 def test_get_best_block_number(self, post_mock): 

50 post_mock.side_effect = self.generate_post_response 

51 self.status_code = 200 

52 client = RskClient('an-url') 

53 best_block = client.get_best_block_number() 

54 self.assertEqual([call('an-url', 

55 data=json.dumps({ 

56 "jsonrpc": "2.0", 

57 "id": self.request_id, 

58 "method": "eth_blockNumber", 

59 "params": [] 

60 }), 

61 headers={ 

62 'content-type': 'application/json' 

63 })], post_mock.call_args_list) 

64 self.assertEqual(int('aa' * 65, 16), best_block) 

65 

66 def test_get_best_block_number_server_error(self, post_mock): 

67 post_mock.side_effect = self.generate_post_response 

68 self.status_code = 400 

69 client = RskClient('an-url') 

70 with self.assertRaises(RskClientError): 

71 client.get_best_block_number() 

72 self.assertEqual([call('an-url', 

73 data=json.dumps({ 

74 "jsonrpc": "2.0", 

75 "id": self.request_id, 

76 "method": "eth_blockNumber", 

77 "params": [] 

78 }), 

79 headers={ 

80 'content-type': 'application/json' 

81 })], post_mock.call_args_list) 

82 

83 def test_get_best_block_number_id_error(self, post_mock): 

84 post_mock.side_effect = self.generate_post_response 

85 self.status_code = 200 

86 self.force_wrong_id = True 

87 client = RskClient('an-url') 

88 with self.assertRaises(RskClientError): 

89 client.get_best_block_number() 

90 self.assertEqual([call('an-url', 

91 data=json.dumps({ 

92 "jsonrpc": "2.0", 

93 "id": self.request_id - 1, 

94 "method": "eth_blockNumber", 

95 "params": [] 

96 }), 

97 headers={ 

98 'content-type': 'application/json' 

99 })], post_mock.call_args_list) 

100 

101 def test_get_block_by_number(self, post_mock): 

102 post_mock.side_effect = self.generate_post_response 

103 self.status_code = 200 

104 block_number = 123456789 

105 client = RskClient('an-url') 

106 best_block = client.get_block_by_number(block_number) 

107 

108 self.assertEqual([call('an-url', 

109 data=json.dumps({ 

110 "jsonrpc": "2.0", 

111 "id": self.request_id, 

112 "method": "eth_getBlockByNumber", 

113 "params": [hex(block_number), False], 

114 }), 

115 headers={"content-type": "application/json"})], 

116 post_mock.call_args_list) 

117 self.assertEqual('aa' * 65, best_block) 

118 

119 def test_get_block_by_number_server_error(self, post_mock): 

120 post_mock.side_effect = self.generate_post_response 

121 self.status_code = 400 

122 block_number = 123456789 

123 client = RskClient('an-url') 

124 with self.assertRaises(RskClientError): 

125 client.get_block_by_number(block_number) 

126 

127 self.assertEqual([call('an-url', 

128 data=json.dumps({ 

129 "jsonrpc": "2.0", 

130 "id": self.request_id, 

131 "method": "eth_getBlockByNumber", 

132 "params": [hex(block_number), False], 

133 }), 

134 headers={"content-type": "application/json"})], 

135 post_mock.call_args_list) 

136 

137 def test_get_block_by_number_id_error(self, post_mock): 

138 post_mock.side_effect = self.generate_post_response 

139 self.status_code = 200 

140 self.force_wrong_id = True 

141 block_number = 123456789 

142 client = RskClient('an-url') 

143 

144 with self.assertRaises(RskClientError): 

145 client.get_block_by_number(block_number) 

146 

147 self.assertEqual([call('an-url', 

148 data=json.dumps({ 

149 "jsonrpc": "2.0", 

150 "id": self.request_id - 1, 

151 "method": "eth_getBlockByNumber", 

152 "params": [hex(block_number), False], 

153 }), 

154 headers={"content-type": "application/json"})], 

155 post_mock.call_args_list)