(self, exchange, method, data, type, skip_keys)
| 951 | return new_input |
| 952 | |
| 953 | async def test_request_statically(self, exchange, method, data, type, skip_keys): |
| 954 | output = None |
| 955 | request_url = None |
| 956 | if self.info: |
| 957 | dump('[INFO] STATIC REQUEST TEST:', method, ':', data['description']) |
| 958 | try: |
| 959 | if not is_sync(): |
| 960 | await call_exchange_method_dynamically(exchange, method, self.sanitize_data_input(data['input'])) |
| 961 | else: |
| 962 | call_exchange_method_dynamically_sync(exchange, method, self.sanitize_data_input(data['input'])) |
| 963 | except Exception as e: |
| 964 | if not (isinstance(e, InvalidProxySettings)): |
| 965 | raise e |
| 966 | output = exchange.last_request_body |
| 967 | request_url = exchange.last_request_url |
| 968 | try: |
| 969 | call_output = exchange.safe_value(data, 'output') |
| 970 | self.assert_static_request_output(exchange, type, skip_keys, data['url'], request_url, call_output, output) |
| 971 | except Exception as e: |
| 972 | self.request_tests_failed = True |
| 973 | error_message = '[' + self.lang + '][STATIC_REQUEST]' + '[' + exchange.id + ']' + '[' + method + ']' + '[' + data['description'] + ']' + exception_message(e) |
| 974 | dump('[TEST_FAILURE]' + error_message) |
| 975 | return True |
| 976 | |
| 977 | async def test_response_statically(self, exchange, method, skip_keys, data): |
| 978 | expected_result = exchange.safe_value(data, 'parsedResponse') |
no test coverage detected