(self, type, target_exchange=None, test_name=None)
| 1219 | return True |
| 1220 | |
| 1221 | async def run_static_tests(self, type, target_exchange=None, test_name=None): |
| 1222 | folder = get_root_dir() + './ts/src/test/static/' + type + '/' |
| 1223 | static_data = self.load_static_data(folder, target_exchange) |
| 1224 | if static_data is None: |
| 1225 | return True |
| 1226 | exchanges = list(static_data.keys()) |
| 1227 | exchange = init_exchange('Exchange', {}) # tmp to do the calculations until we have the ast-transpiler transpiling this code |
| 1228 | promises = [] |
| 1229 | sum = 0 |
| 1230 | if target_exchange: |
| 1231 | dump('[INFO:MAIN] Exchange to test: ' + target_exchange) |
| 1232 | if test_name: |
| 1233 | dump('[INFO:MAIN] Testing only: ' + test_name) |
| 1234 | for i in range(0, len(exchanges)): |
| 1235 | exchange_name = exchanges[i] |
| 1236 | exchange_data = static_data[exchange_name] |
| 1237 | disabled = self.check_if_exchange_is_disabled(exchange_name, exchange_data) |
| 1238 | if disabled: |
| 1239 | continue |
| 1240 | number_of_tests = self.get_number_of_tests_from_exchange(exchange, exchange_data, test_name) |
| 1241 | sum = exchange.sum(sum, number_of_tests) |
| 1242 | if type == 'request': |
| 1243 | promises.append(self.test_exchange_request_statically(exchange_name, exchange_data, test_name)) |
| 1244 | else: |
| 1245 | promises.append(self.test_exchange_response_statically(exchange_name, exchange_data, test_name)) |
| 1246 | try: |
| 1247 | await asyncio.gather(*promises) |
| 1248 | except Exception as e: |
| 1249 | if type == 'request': |
| 1250 | self.request_tests_failed = True |
| 1251 | else: |
| 1252 | self.response_tests_failed = True |
| 1253 | error_message = '[' + self.lang + '][STATIC_REQUEST]' + exception_message(e) |
| 1254 | dump('[TEST_FAILURE]' + error_message) |
| 1255 | if self.request_tests_failed or self.response_tests_failed: |
| 1256 | exit_script(1) |
| 1257 | else: |
| 1258 | prefix = '[SYNC]' if (is_sync()) else '' |
| 1259 | success_message = '[' + self.lang + ']' + prefix + '[TEST_SUCCESS] ' + str(sum) + ' static ' + type + ' tests passed.' |
| 1260 | dump('[INFO]' + success_message) |
| 1261 | |
| 1262 | async def run_static_response_tests(self, exchange_name=None, test=None): |
| 1263 | # ----------------------------------------------------------------------------- |
no test coverage detected