Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use _create_pool method in lisa
Best Python code snippet using lisa_python
testBetOnFlyby.py
Source:testBetOnFlyby.py 
...24 self.assertEqual(VMState.HALT, self.engine.vm_state)25 self.assertEqual(VoidType, result)26 event_notifications = self.engine.get_events(event_name=on_change_image_event_name)27 self.assertEqual(1, len(event_notifications))28 def _create_pool(self, creator_account: bytes, description: str, options: List[str]):29 self.engine.add_signer_account(creator_account)30 return self.engine.run(self.nef_path, 'create_pool', creator_account, description, options)31 def test_create_pool_success(self):32 self.engine.reset_engine()33 creator_account = bytes(20)34 description = 'Bet for testing'35 options = ['choice1', 'choice2', 'choice3']36 result = self._create_pool(creator_account, description, options)37 self.assertEqual(VMState.HALT, self.engine.vm_state)38 self.assertIsInstance(result, bytes)39 self.assertEqual(32, len(result))40 def test_create_pool_fail_check_witness(self):41 self.engine.reset_engine()42 creator_account = bytes(20)43 description = 'Bet for testing'44 options = []45 # need signing46 self.engine.run(self.nef_path, 'create_pool', creator_account, description, options)47 self.assertEqual(VMState.FAULT, self.engine.vm_state)48 self.assertTrue(self.engine.error.endswith('No authorization.'))49 def test_create_pool_fail_not_enough_options(self):50 self.engine.reset_engine()51 creator_account = bytes(20)52 description = 'Bet for testing'53 options = []54 self.engine.add_signer_account(creator_account)55 # need at least two different options56 self.engine.run(self.nef_path, 'create_pool', creator_account, description, options)57 self.assertEqual(VMState.FAULT, self.engine.vm_state)58 self.assertTrue(self.engine.error.endswith('Not enough options to create a pool'))59 options = ['choice1', 'choice1'] # need at least two different options60 self.engine.add_signer_account(creator_account)61 self.engine.run(self.nef_path, 'create_pool', creator_account, description, options)62 self.assertEqual(VMState.FAULT, self.engine.vm_state)63 self.assertTrue(self.engine.error.endswith('Not enough options to create a pool'))64 def test_create_pool_fail_empty_option(self):65 self.engine.reset_engine()66 creator_account = bytes(20)67 description = 'Bet for testing'68 options = ['choice1', '', 'choice2']69 self.engine.add_signer_account(creator_account)70 self.engine.run(self.nef_path, 'create_pool', creator_account, description, options)71 self.assertEqual(VMState.FAULT, self.engine.vm_state)72 self.assertTrue(self.engine.error.endswith('Cannot have an empty option'))73 def _bet(self, pool_id: bytes, player: bytes, option: str):74 price_in_gas = 1 * 10 ** 8 # 1 GAS75 self.engine.add_gas(player, price_in_gas)76 self.engine.add_signer_account(player)77 self.engine.run(self.nef_path, 'bet', player, pool_id, option)78 def test_bet_success(self):79 self.engine.reset_engine()80 creator_account = bytes(20)81 description = 'Bet for testing'82 options = ['choice1', 'choice2', 'choice3']83 pool_id = self._create_pool(creator_account, description, options)84 player = bytes(range(20))85 bet_option = 'choice1'86 price_in_gas = 1 * 10 ** 8 # 1 GAS87 self.engine.add_signer_account(player)88 self.engine.add_gas(player, price_in_gas)89 result = self.engine.run(self.nef_path, 'bet', player, pool_id, bet_option)90 self.assertEqual(VMState.HALT, self.engine.vm_state)91 self.assertEqual(VoidType, result)92 def test_bet_fail_doesnt_exist(self):93 self.engine.reset_engine()94 pool_id = bytes(32)95 player = bytes(20)96 bet_option = 'choice1'97 self.engine.run(self.nef_path, 'bet', player, pool_id, bet_option)98 self.assertEqual(VMState.FAULT, self.engine.vm_state)99 self.assertTrue(self.engine.error.endswith("Pool doesn't exist."))100 def test_bet_fail_check_witness(self):101 self.engine.reset_engine()102 creator_account = bytes(20)103 description = 'Bet for testing'104 options = ['choice1', 'choice2', 'choice3']105 pool_id = self._create_pool(creator_account, description, options)106 player = bytes(range(20))107 bet_option = 'choice1'108 self.engine.run(self.nef_path, 'bet', player, pool_id, bet_option)109 self.assertEqual(VMState.FAULT, self.engine.vm_state)110 self.assertTrue(self.engine.error.endswith('No authorization.'))111 def test_bet_fail_finished_already(self):112 self.engine.reset_engine()113 creator_account = bytes(20)114 description = 'Bet for testing'115 options = ['choice1', 'choice2', 'choice3']116 pool_id = self._create_pool(creator_account, description, options)117 bet_option = 'choice1'118 self._finish_pool(creator_account, pool_id, [bet_option])119 player = bytes(range(20))120 self._bet(pool_id, player, bet_option)121 self.assertEqual(VMState.FAULT, self.engine.vm_state)122 self.assertTrue(self.engine.error.endswith('Pool is finished already'))123 def test_bet_fail_voted_already(self):124 self.engine.reset_engine()125 creator_account = bytes(20)126 description = 'Bet for testing'127 options = ['choice1', 'choice2', 'choice3']128 pool_id = self._create_pool(creator_account, description, options)129 player = bytes(range(20))130 bet_option = 'choice1'131 self._bet(pool_id, player, bet_option)132 self.assertEqual(VMState.HALT, self.engine.vm_state)133 self._bet(pool_id, player, 'choice2')134 self.assertEqual(VMState.FAULT, self.engine.vm_state)135 self.assertTrue(self.engine.error.endswith('Only one bet is allowed per account'))136 def test_bet_fail_invalid_option(self):137 self.engine.reset_engine()138 creator_account = bytes(20)139 description = 'Bet for testing'140 options = ['choice1', 'choice2', 'choice3']141 pool_id = self._create_pool(creator_account, description, options)142 player = bytes(range(20))143 bet_option = 'choice4'144 self._bet(pool_id, player, bet_option)145 self.assertEqual(VMState.FAULT, self.engine.vm_state)146 self.assertTrue(self.engine.error.endswith('Invalid option for this pool'))147 def test_bet_fail_not_enough_gas(self):148 self.engine.reset_engine()149 creator_account = bytes(20)150 description = 'Bet for testing'151 options = ['choice1', 'choice2', 'choice3']152 pool_id = self._create_pool(creator_account, description, options)153 player = bytes(range(20))154 bet_option = 'choice1'155 self.engine.add_signer_account(player)156 self.engine.run(self.nef_path, 'bet', player, pool_id, bet_option)157 self.assertEqual(VMState.FAULT, self.engine.vm_state)158 self.assertTrue(self.engine.error.endswith('GAS transfer was not successful'))159 def _finish_pool(self, creator_account: bytes, pool_id: bytes, winners: List[str]):160 self.engine.add_signer_account(creator_account)161 self.engine.run(self.nef_path, 'finish_pool', pool_id, winners)162 def test_finish_pool_success_one_winner(self):163 self.engine.reset_engine()164 creator_account = bytes(20)165 description = 'Bet for testing'166 options = ['choice1', 'choice2', 'choice3']167 pool_id = self._create_pool(creator_account, description, options)168 player = bytes(range(20))169 bet_option = 'choice2'170 self._bet(pool_id, player, bet_option)171 winner_option = [bet_option]172 self.engine.add_signer_account(creator_account)173 self.engine.run(self.nef_path, 'finish_pool', pool_id, winner_option)174 self.assertEqual(VMState.HALT, self.engine.vm_state)175 def test_finish_pool_success_two_winners(self):176 self.engine.reset_engine()177 creator_account = bytes(20)178 description = 'Bet for testing'179 options = ['choice1', 'choice2', 'choice3']180 pool_id = self._create_pool(creator_account, description, options)181 player = bytes(range(20))182 bet_option = 'choice2'183 self._bet(pool_id, player, bet_option)184 winner_option = ['choice1', bet_option]185 self.engine.add_signer_account(creator_account)186 self.engine.run(self.nef_path, 'finish_pool', pool_id, winner_option)187 self.assertEqual(VMState.HALT, self.engine.vm_state)188 def test_finish_pool_fail_doesnt_exist(self):189 self.engine.reset_engine()190 pool_id = bytes(32)191 winner_option = []192 self.engine.run(self.nef_path, 'finish_pool', pool_id, winner_option)193 self.assertEqual(VMState.FAULT, self.engine.vm_state)194 self.assertTrue(self.engine.error.endswith("Pool doesn't exist."))195 def test_finish_pool_fail_check_witness(self):196 self.engine.reset_engine()197 creator_account = bytes(20)198 description = 'Bet for testing'199 options = ['choice1', 'choice2', 'choice3']200 pool_id = self._create_pool(creator_account, description, options)201 winner_option = ['choice1']202 self.engine.run(self.nef_path, 'finish_pool', pool_id, winner_option)203 self.assertEqual(VMState.FAULT, self.engine.vm_state)204 self.assertTrue(self.engine.error.endswith('No authorization.'))205 def test_finish_pool_fail_finished_already(self):206 self.engine.reset_engine()207 creator_account = bytes(20)208 description = 'Bet for testing'209 options = ['choice1', 'choice2', 'choice3']210 pool_id = self._create_pool(creator_account, description, options)211 winner_option = ['choice1']212 self._finish_pool(creator_account, pool_id, winner_option)213 self.assertEqual(VMState.HALT, self.engine.vm_state)214 self._finish_pool(creator_account, pool_id, winner_option)215 self.assertEqual(VMState.FAULT, self.engine.vm_state)216 self.assertTrue(self.engine.error.endswith('Pool is finished already'))217 def test_finish_pool_fail_not_enough_winner_options(self):218 self.engine.reset_engine()219 creator_account = bytes(20)220 description = 'Bet for testing'221 options = ['choice1', 'choice2', 'choice3']222 pool_id = self._create_pool(creator_account, description, options)223 winner_option = []224 self.engine.add_signer_account(creator_account)225 self.engine.run(self.nef_path, 'finish_pool', pool_id, winner_option)226 self.assertEqual(VMState.FAULT, self.engine.vm_state)227 self.assertTrue(self.engine.error.endswith('At least one winner is required'))228 def test_finish_pool_fail_invalid_winner_option(self):229 self.engine.reset_engine()230 creator_account = bytes(20)231 description = 'Bet for testing'232 options = ['choice1', 'choice2', 'choice3']233 pool_id = self._create_pool(creator_account, description, options)234 winner_option = ['choice4']235 self.engine.add_signer_account(creator_account)236 self.engine.run(self.nef_path, 'finish_pool', pool_id, winner_option)237 self.assertEqual(VMState.FAULT, self.engine.vm_state)238 self.assertTrue(self.engine.error.endswith('Invalid option for this pool'))239 def _cancel_pool(self, creator_account: bytes, pool_id: bytes):240 self.engine.add_signer_account(creator_account)241 self.engine.run(self.nef_path, 'cancel_pool', pool_id)242 def test_cancel_pool_success(self):243 self.engine.reset_engine()244 creator_account = bytes(20)245 description = 'Bet for testing'246 options = ['choice1', 'choice2', 'choice3']247 pool_id = self._create_pool(creator_account, description, options)248 self.engine.add_signer_account(creator_account)249 self.engine.run(self.nef_path, 'cancel_pool', pool_id)250 self.assertEqual(VMState.HALT, self.engine.vm_state)251 def test_cancel_pool_fail_doesnt_exist(self):252 self.engine.reset_engine()253 pool_id = bytes(32)254 self.engine.run(self.nef_path, 'cancel_pool', pool_id)255 self.assertEqual(VMState.FAULT, self.engine.vm_state)256 self.assertTrue(self.engine.error.endswith("Pool doesn't exist."))257 def test_cancel_pool_fail_check_witness(self):258 self.engine.reset_engine()259 creator_account = bytes(20)260 description = 'Bet for testing'261 options = ['choice1', 'choice2', 'choice3']262 pool_id = self._create_pool(creator_account, description, options)263 self.engine.run(self.nef_path, 'cancel_pool', pool_id)264 self.assertEqual(VMState.FAULT, self.engine.vm_state)265 self.assertTrue(self.engine.error.endswith('No authorization.'))266 def test_cancel_pool_fail_finished_already(self):267 self.engine.reset_engine()268 creator_account = bytes(20)269 description = 'Bet for testing'270 options = ['choice1', 'choice2', 'choice3']271 pool_id = self._create_pool(creator_account, description, options)272 winner_option = ['choice1']273 self._finish_pool(creator_account, pool_id, winner_option)274 self.assertEqual(VMState.HALT, self.engine.vm_state)275 self.engine.add_signer_account(creator_account)276 self.engine.run(self.nef_path, 'cancel_pool', pool_id)277 self.assertEqual(VMState.FAULT, self.engine.vm_state)278 self.assertTrue(self.engine.error.endswith('Pool is finished already'))279 def test_cancel_player_bet_success(self):280 self.engine.reset_engine()281 creator_account = bytes(20)282 description = 'Bet for testing'283 options = ['choice1', 'choice2', 'choice3']284 pool_id = self._create_pool(creator_account, description, options)285 player = bytes(range(20))286 bet_option = 'choice1'287 self._bet(pool_id, player, bet_option)288 self.engine.add_signer_account(player)289 self.engine.run(self.nef_path, 'cancel_player_bet', player, pool_id)290 self.assertEqual(VMState.HALT, self.engine.vm_state)291 def test_cancel_player_bet_fail_doesnt_exist(self):292 self.engine.reset_engine()293 pool_id = bytes(32)294 player = bytes(20)295 self.engine.run(self.nef_path, 'cancel_player_bet', player, pool_id)296 self.assertEqual(VMState.FAULT, self.engine.vm_state)297 self.assertTrue(self.engine.error.endswith("Pool doesn't exist."))298 def test_cancel_player_bet_fail_check_witness(self):299 self.engine.reset_engine()300 creator_account = bytes(20)301 description = 'Bet for testing'302 options = ['choice1', 'choice2', 'choice3']303 pool_id = self._create_pool(creator_account, description, options)304 player = bytes(range(20))305 self.engine.run(self.nef_path, 'cancel_player_bet', player, pool_id)306 self.assertEqual(VMState.FAULT, self.engine.vm_state)307 self.assertTrue(self.engine.error.endswith('No authorization.'))308 def test_cancel_player_bet_fail_player_didnt_bet(self):309 self.engine.reset_engine()310 creator_account = bytes(20)311 description = 'Bet for testing'312 options = ['choice1', 'choice2', 'choice3']313 pool_id = self._create_pool(creator_account, description, options)314 player = bytes(range(20))315 self.engine.run(self.nef_path, 'cancel_player_bet', player, pool_id)316 self.assertEqual(VMState.FAULT, self.engine.vm_state)317 self.assertTrue(self.engine.error.endswith('No authorization.'))318 def test_cancel_player_bet_fail_finished_already(self):319 self.engine.reset_engine()320 creator_account = bytes(20)321 description = 'Bet for testing'322 options = ['choice1', 'choice2', 'choice3']323 pool_id = self._create_pool(creator_account, description, options)324 player = bytes(range(20))325 self.engine.add_signer_account(player)326 self.engine.run(self.nef_path, 'cancel_player_bet', player, pool_id)327 self.assertEqual(VMState.FAULT, self.engine.vm_state)328 self.assertTrue(self.engine.error.endswith("Player didn't bet on this pool"))329 def test_get_pool_success_on_going(self):330 self.engine.reset_engine()331 creator_account = bytes(20)332 description = 'Bet for testing'333 options = ['choice1', 'choice2', 'choice3']334 pool_id = self._create_pool(creator_account, description, options)335 self.assertEqual(VMState.HALT, self.engine.vm_state)336 result = self.engine.run(self.nef_path, 'get_pool', pool_id)337 self.assertEqual(VMState.HALT, self.engine.vm_state)338 self.assertIsInstance(result, list)339 self.assertEqual(6, len(result))340 if isinstance(result[1], str): # test engine converts to string whenever is possible341 result[1] = result[1].encode('utf-8')342 self.assertEqual(pool_id, result[0]) # pool_id343 self.assertEqual(creator_account, result[1]) # creator344 self.assertEqual(description, result[2]) # description345 self.assertEqual(options, result[3]) # options346 self.assertIsNone(result[4]) # result - is None because it's on going347 self.assertEqual({}, result[5]) # bets on this pool348 def test_get_pool_success_has_bets(self):349 self.engine.reset_engine()350 creator_account = bytes(20)351 description = 'Bet for testing'352 options = ['choice1', 'choice2', 'choice3']353 pool_id = self._create_pool(creator_account, description, options)354 self.assertEqual(VMState.HALT, self.engine.vm_state)355 player = bytes(range(20))356 bet_option = 'choice1'357 self._bet(pool_id, player, bet_option)358 self.assertEqual(VMState.HALT, self.engine.vm_state)359 result = self.engine.run(self.nef_path, 'get_pool', pool_id)360 self.assertEqual(VMState.HALT, self.engine.vm_state)361 self.assertIsInstance(result, list)362 self.assertEqual(6, len(result))363 if isinstance(result[1], str): # test engine converts to string whenever is possible364 result[1] = result[1].encode('utf-8')365 self.assertEqual(pool_id, result[0]) # pool_id366 self.assertEqual(creator_account, result[1]) # creator367 self.assertEqual(description, result[2]) # description368 self.assertEqual(options, result[3]) # options369 self.assertIsNone(result[4]) # result - is None because it's on going370 self.assertIsInstance(result[5], dict) # bets on this pool371 self.assertEqual(1, len(result[5]))372 bet1_player, bet1_choice = list(result[5].items())[0]373 if isinstance(bet1_player, str): # test engine converts to string whenever is possible374 bet1_player = bet1_player.encode('utf-8')375 self.assertEqual(player, bet1_player)376 self.assertEqual(bet_option, bet1_choice)377 def test_get_pool_success_finished(self):378 self.engine.reset_engine()379 creator_account = bytes(20)380 description = 'Bet for testing'381 options = ['choice1', 'choice2', 'choice3']382 pool_id = self._create_pool(creator_account, description, options)383 self.assertEqual(VMState.HALT, self.engine.vm_state)384 winners = ['choice1']385 self._finish_pool(creator_account, pool_id, winners)386 self.assertEqual(VMState.HALT, self.engine.vm_state)387 result = self.engine.run(self.nef_path, 'get_pool', pool_id)388 self.assertEqual(VMState.HALT, self.engine.vm_state)389 self.assertIsInstance(result, list)390 self.assertEqual(6, len(result))391 if isinstance(result[1], str): # test engine converts to string whenever is possible392 result[1] = result[1].encode('utf-8')393 self.assertEqual(pool_id, result[0]) # pool_id394 self.assertEqual(creator_account, result[1]) # creator395 self.assertEqual(description, result[2]) # description396 self.assertEqual(options, result[3]) # options397 self.assertEqual(winners, result[4]) # result398 self.assertEqual({}, result[5]) # bets on this pool399 def test_get_pool_success_cancelled(self):400 self.engine.reset_engine()401 creator_account = bytes(20)402 description = 'Bet for testing'403 options = ['choice1', 'choice2', 'choice3']404 pool_id = self._create_pool(creator_account, description, options)405 self.assertEqual(VMState.HALT, self.engine.vm_state)406 self._cancel_pool(creator_account, pool_id)407 self.assertEqual(VMState.HALT, self.engine.vm_state)408 result = self.engine.run(self.nef_path, 'get_pool', pool_id)409 self.assertEqual(VMState.HALT, self.engine.vm_state)410 self.assertIsInstance(result, list)411 self.assertEqual(6, len(result))412 if isinstance(result[1], str): # test engine converts to string whenever is possible413 result[1] = result[1].encode('utf-8')414 self.assertEqual(pool_id, result[0]) # pool_id415 self.assertEqual(creator_account, result[1]) # creator416 self.assertEqual(description, result[2]) # description417 self.assertEqual(options, result[3]) # options418 self.assertEqual('Cancelled by owner', result[4]) # result419 self.assertEqual({}, result[5]) # bets on this pool420 def test_get_pool_fail_doesnt_exist(self):421 self.engine.reset_engine()422 pool_id = bytes(32)423 self.engine.run(self.nef_path, 'get_pool', pool_id)424 self.assertEqual(VMState.FAULT, self.engine.vm_state)425 self.assertTrue(self.engine.error.endswith("Pool doesn't exist."))426 def test_list_open_pools_success_no_pools_created(self):427 self.engine.reset_engine()428 result = self.engine.run(self.nef_path, 'list_on_going_pools')429 self.assertEqual(VMState.HALT, self.engine.vm_state)430 self.assertIsInstance(result, list)431 self.assertEqual(0, len(result))432 def test_list_open_pools_success_found_pool(self):433 self.engine.reset_engine()434 creator_account = bytes(20)435 description = 'Bet for testing'436 options = ['choice1', 'choice2', 'choice3']437 pool_id = self._create_pool(creator_account, description, options)438 self.assertEqual(VMState.HALT, self.engine.vm_state)439 result = self.engine.run(self.nef_path, 'list_on_going_pools')440 self.assertEqual(VMState.HALT, self.engine.vm_state)441 self.assertIsInstance(result, list)442 self.assertEqual(1, len(result))443 pool = result[0]444 self.assertIsInstance(pool, list)445 self.assertEqual(6, len(pool))446 if isinstance(pool[1], str): # test engine converts to string whenever is possible447 pool[1] = pool[1].encode('utf-8')448 self.assertEqual(pool_id, pool[0]) # pool_id449 self.assertEqual(creator_account, pool[1]) # creator450 self.assertEqual(description, pool[2]) # description451 self.assertEqual(options, pool[3]) # options452 self.assertIsNone(pool[4]) # result - is None because it's on going453 def test_list_open_pools_success_finished_pool(self):454 self.engine.reset_engine()455 creator_account = bytes(20)456 description = 'Bet for testing'457 options = ['choice1', 'choice2', 'choice3']458 pool_id = self._create_pool(creator_account, description, options)459 self.assertEqual(VMState.HALT, self.engine.vm_state)460 self._finish_pool(creator_account, pool_id, ['choice1'])461 self.assertEqual(VMState.HALT, self.engine.vm_state)462 result = self.engine.run(self.nef_path, 'list_on_going_pools')463 self.assertEqual(VMState.HALT, self.engine.vm_state)464 self.assertIsInstance(result, list)465 self.assertEqual(0, len(result))466 def test_list_open_pools_success_canceled_pool(self):467 self.engine.reset_engine()468 creator_account = bytes(20)469 description = 'Bet for testing'470 options = ['choice1', 'choice2', 'choice3']471 pool_id = self._create_pool(creator_account, description, options)472 self.assertEqual(VMState.HALT, self.engine.vm_state)473 self._cancel_pool(creator_account, pool_id)474 self.assertEqual(VMState.HALT, self.engine.vm_state)475 result = self.engine.run(self.nef_path, 'list_on_going_pools')476 self.assertEqual(VMState.HALT, self.engine.vm_state)477 self.assertIsInstance(result, list)...
boosting_ranker.py
Source:boosting_ranker.py
...15class BoostingRanker(CandidatesRanker):16 def __init__(self, features: List[Feature]):17 super(BoostingRanker, self).__init__(features)18 self._ranker = CatBoostRanker(verbose=False, loss_function="YetiRankPairwise")19 def _create_pool(self, queries: List[str], docs: List[List[str]], labels: Optional[List[str]] = None) -> Pool:20 X, queries_pool, labels_pool = [], [], []21 for query_id in range(len(queries)):22 for doc_id in range(len(docs[query_id])):23 queries_pool.append(query_id + 1)24 features = [feature(queries[query_id], docs[query_id][doc_id]) for feature in self._features]25 X.append(features)26 if labels is not None:27 labels_pool.append(0)28 if labels is not None:29 queries_pool.append(query_id + 1)30 features = [feature(queries[query_id], labels[query_id]) for feature in self._features]31 X.append(features)32 if labels is not None:33 labels_pool.append(1)34 if labels is None:35 return Pool(data=X, group_id=queries_pool)36 return Pool(data=X, group_id=queries_pool, label=labels_pool)37 def fit(self, train_dataset: List[DatasetEntry],38 val_dataset: Optional[List[DatasetEntry]] = None) -> "CandidatesRanker":39 train_pool = self._create_pool(*convert(train_dataset))40 if val_dataset is not None:41 self._ranker.fit(train_pool, eval_set=self._create_pool(*convert(val_dataset)))42 else:43 self._ranker.fit(train_pool)44 return self45 def rank(self, misspelled: str, candidates: List[str]) -> List[str]:46 pool = self._create_pool([misspelled], [candidates])47 scores = self._ranker.predict(pool)...
redis_pool.py
Source:redis_pool.py
2from app.core import config3class RedisPool:4 redis_pool_dict = {}5 def __await__(self):6 self._create_pool()7 return self._create_pool().__await__()8 async def _create_pool(self):9 for i in config.REDIS_DB_LIST:10 pool = aioredis.ConnectionPool.from_url(config.REDIS_URI + f"/{i}", max_connections=10)11 redis = aioredis.Redis(connection_pool=pool)12 self.redis_pool_dict.update({i: redis})13 return self14 def select_db(self, db=52):15 c = self.redis_pool_dict[db]16 if not c:17 raise ValueError('è°ç¨çRedisæ°æ®åºæªå建è¿æ¥æ± ')18 return c19 def close_pool(self):20 for i in self.redis_pool_dict.values():...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run lisa automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
