(
self,
uri: str,
http_method: str = 'GET',
req_type: ReqType = ReqType.JSON,
body: str = '',
status: int = 200,
ret_type: RetType = RetType.JSON,
query_params: typing.Union[dict[str, typing.Any], str, None] = None,
)
| 57 | self.noban_tx_relay = True |
| 58 | |
| 59 | def test_rest_request( |
| 60 | self, |
| 61 | uri: str, |
| 62 | http_method: str = 'GET', |
| 63 | req_type: ReqType = ReqType.JSON, |
| 64 | body: str = '', |
| 65 | status: int = 200, |
| 66 | ret_type: RetType = RetType.JSON, |
| 67 | query_params: typing.Union[dict[str, typing.Any], str, None] = None, |
| 68 | ) -> typing.Union[http.client.HTTPResponse, bytes, str, None]: |
| 69 | rest_uri = '/rest' + uri |
| 70 | if req_type in ReqType: |
| 71 | rest_uri += f'.{req_type.name.lower()}' |
| 72 | if query_params: |
| 73 | if isinstance(query_params, str): |
| 74 | rest_uri += f'?{query_params}' |
| 75 | else: |
| 76 | rest_uri += f'?{urllib.parse.urlencode(query_params)}' |
| 77 | |
| 78 | conn = http.client.HTTPConnection(self.url.hostname, self.url.port) |
| 79 | self.log.debug(f'{http_method} {rest_uri} {body}') |
| 80 | if http_method == 'GET': |
| 81 | conn.request('GET', rest_uri) |
| 82 | elif http_method == 'POST': |
| 83 | conn.request('POST', rest_uri, body) |
| 84 | resp = conn.getresponse() |
| 85 | |
| 86 | assert resp.status == status, f"Expected: {status}, Got: {resp.status} ({resp.reason}) - Response: {str(resp.read())}" |
| 87 | |
| 88 | if ret_type == RetType.OBJ: |
| 89 | return resp |
| 90 | elif ret_type == RetType.BYTES: |
| 91 | return resp.read() |
| 92 | elif ret_type == RetType.JSON: |
| 93 | return json.loads(resp.read().decode('utf-8'), parse_float=Decimal) |
| 94 | |
| 95 | return None |
| 96 | |
| 97 | def run_test(self): |
| 98 | self.url = urllib.parse.urlparse(self.nodes[0].url) |
no test coverage detected