Best Python code snippet using playwright-python
test_targetstate.py
Source:test_targetstate.py
...42 our_address=our_address,43 partner_address=UNIT_TRANSFER_SENDER,44 partner_balance=amount,45 )46 from_route = factories.route_from_channel(from_channel)47 if expiration is None:48 expiration = from_channel.reveal_timeout + block_number + 149 from_transfer = factories.make_signed_transfer_for(50 from_channel,51 amount,52 initiator,53 our_address,54 expiration,55 UNIT_SECRET,56 )57 state_change = ActionInitTarget(from_route, from_transfer)58 iteration = target.handle_inittarget(59 state_change,60 from_channel,61 pseudo_random_generator,62 block_number,63 )64 return from_channel, iteration.new_state65def test_events_for_onchain_secretreveal():66 """ Secret must be registered on-chain when the unsafe region is reached and67 the secret is known.68 """69 amount = 370 block_number = 1071 expiration = block_number + 3072 initiator = HOP173 target_address = UNIT_TRANSFER_TARGET74 from_channel = factories.make_channel(75 our_address=target_address,76 partner_address=UNIT_TRANSFER_SENDER,77 partner_balance=amount,78 )79 from_route = factories.route_from_channel(from_channel)80 from_transfer = factories.make_signed_transfer_for(81 from_channel,82 amount,83 initiator,84 target_address,85 expiration,86 UNIT_SECRET,87 )88 channel.handle_receive_lockedtransfer(89 from_channel,90 from_transfer,91 )92 channel.register_offchain_secret(from_channel, UNIT_SECRET, UNIT_SECRETHASH)93 safe_to_wait = expiration - from_channel.reveal_timeout - 194 unsafe_to_wait = expiration - from_channel.reveal_timeout95 state = TargetTransferState(from_route, from_transfer)96 events = target.events_for_onchain_secretreveal(state, from_channel, safe_to_wait)97 assert not events98 events = target.events_for_onchain_secretreveal(state, from_channel, unsafe_to_wait)99 assert events100 assert isinstance(events[0], ContractSendSecretReveal)101 assert events[0].secret == UNIT_SECRET102def test_handle_inittarget():103 """ Init transfer must send a secret request if the expiration is valid. """104 amount = 3105 block_number = 1106 initiator = factories.HOP1107 target_address = UNIT_TRANSFER_TARGET108 pseudo_random_generator = random.Random()109 from_channel = factories.make_channel(110 our_address=target_address,111 partner_address=UNIT_TRANSFER_SENDER,112 partner_balance=amount,113 )114 from_route = factories.route_from_channel(from_channel)115 expiration = from_channel.reveal_timeout + block_number + 1116 from_transfer = factories.make_signed_transfer(117 amount,118 initiator,119 target_address,120 expiration,121 UNIT_SECRET,122 channel_identifier=from_channel.identifier,123 token_network_address=from_channel.token_network_identifier,124 )125 state_change = ActionInitTarget(126 from_route,127 from_transfer,128 )129 iteration = target.handle_inittarget(130 state_change,131 from_channel,132 pseudo_random_generator,133 block_number,134 )135 assert must_contain_entry(iteration.events, SendSecretRequest, {136 'payment_identifier': from_transfer.payment_identifier,137 'amount': from_transfer.lock.amount,138 'secrethash': from_transfer.lock.secrethash,139 'recipient': initiator,140 })141 assert must_contain_entry(iteration.events, SendProcessed, {})142def test_handle_inittarget_bad_expiration():143 """ Init transfer must do nothing if the expiration is bad. """144 block_number = 1145 amount = 3146 initiator = factories.HOP1147 target_address = UNIT_TRANSFER_TARGET148 pseudo_random_generator = random.Random()149 from_channel = factories.make_channel(150 our_address=target_address,151 partner_address=UNIT_TRANSFER_SENDER,152 partner_balance=amount,153 )154 from_route = factories.route_from_channel(from_channel)155 expiration = from_channel.reveal_timeout + block_number + 1156 from_transfer = factories.make_signed_transfer_for(157 from_channel,158 amount,159 initiator,160 target_address,161 expiration,162 UNIT_SECRET,163 )164 channel.handle_receive_lockedtransfer(165 from_channel,166 from_transfer,167 )168 state_change = ActionInitTarget(from_route, from_transfer)169 iteration = target.handle_inittarget(170 state_change,171 from_channel,172 pseudo_random_generator,173 block_number,174 )175 assert must_contain_entry(iteration.events, EventUnlockClaimFailed, {})176def test_handle_offchain_secretreveal():177 """ The target node needs to inform the secret to the previous node to178 receive an updated balance proof.179 """180 amount = 3181 block_number = 1182 expiration = block_number + factories.UNIT_REVEAL_TIMEOUT183 initiator = factories.HOP1184 our_address = factories.ADDR185 secret = factories.UNIT_SECRET186 pseudo_random_generator = random.Random()187 channel_state, state = make_target_state(188 our_address,189 amount,190 block_number,191 initiator,192 expiration,193 )194 state_change = ReceiveSecretReveal(secret, initiator)195 iteration = target.handle_offchain_secretreveal(196 target_state=state,197 state_change=state_change,198 channel_state=channel_state,199 pseudo_random_generator=pseudo_random_generator,200 block_number=block_number,201 )202 assert len(iteration.events) == 1203 reveal = iteration.events[0]204 assert isinstance(reveal, SendSecretReveal)205 assert iteration.new_state.state == 'reveal_secret'206 assert reveal.secret == secret207 assert reveal.recipient == state.route.node_address208 # if we get an empty hash secret make sure it's rejected209 secret = EMPTY_HASH210 state_change = ReceiveSecretReveal(secret, initiator)211 iteration = target.handle_offchain_secretreveal(212 target_state=state,213 state_change=state_change,214 channel_state=channel_state,215 pseudo_random_generator=pseudo_random_generator,216 block_number=block_number,217 )218 assert len(iteration.events) == 0219def test_handle_offchain_secretreveal_after_lock_expired():220 """Test that getting the secret revealed after lock expiration for the221 target does not end up continuously emitting EventUnlockClaimFailed222 Target part for https://github.com/raiden-network/raiden/issues/3086223 """224 amount = 3225 block_number = 1226 expiration = block_number + factories.UNIT_REVEAL_TIMEOUT227 initiator = factories.HOP1228 our_address = factories.ADDR229 secret = factories.UNIT_SECRET230 pseudo_random_generator = random.Random()231 channel_state, state = make_target_state(232 our_address,233 amount,234 block_number,235 initiator,236 expiration,237 )238 lock_expiration = state.transfer.lock.expiration239 lock_expiration_block_number = lock_expiration + DEFAULT_NUMBER_OF_BLOCK_CONFIRMATIONS * 2240 lock_expiration_block = Block(241 block_number=lock_expiration_block_number,242 gas_limit=1,243 block_hash=factories.make_transaction_hash(),244 )245 iteration = target.state_transition(246 target_state=state,247 state_change=lock_expiration_block,248 channel_state=channel_state,249 pseudo_random_generator=pseudo_random_generator,250 block_number=lock_expiration_block_number,251 )252 state = iteration.new_state253 msg = 'At the expiration block we should get an EventUnlockClaimFailed'254 assert must_contain_entry(iteration.events, EventUnlockClaimFailed, {}), msg255 iteration = target.state_transition(256 target_state=state,257 state_change=ReceiveSecretReveal(secret, initiator),258 channel_state=channel_state,259 pseudo_random_generator=pseudo_random_generator,260 block_number=lock_expiration_block_number + 1,261 )262 state = iteration.new_state263 next_block = Block(264 block_number=lock_expiration_block_number + 1,265 gas_limit=1,266 block_hash=factories.make_transaction_hash(),267 )268 iteration = target.state_transition(269 target_state=state,270 state_change=next_block,271 channel_state=channel_state,272 pseudo_random_generator=pseudo_random_generator,273 block_number=lock_expiration_block_number + 1,274 )275 msg = 'At the next block we should not get the same event'276 assert not must_contain_entry(iteration.events, EventUnlockClaimFailed, {}), msg277def test_handle_onchain_secretreveal():278 """ The target node must update the lock state when the secret is279 registered in the blockchain.280 """281 amount = 3282 block_number = 1283 expiration = block_number + factories.UNIT_REVEAL_TIMEOUT284 initiator = factories.HOP1285 our_address = factories.ADDR286 secret = factories.UNIT_SECRET287 pseudo_random_generator = random.Random()288 channel_state, state = make_target_state(289 our_address,290 amount,291 block_number,292 initiator,293 expiration,294 )295 assert factories.UNIT_SECRETHASH in channel_state.partner_state.secrethashes_to_lockedlocks296 offchain_secret_reveal_iteration = target.state_transition(297 state,298 ReceiveSecretReveal(secret, initiator),299 channel_state,300 pseudo_random_generator,301 block_number,302 )303 assert factories.UNIT_SECRETHASH in channel_state.partner_state.secrethashes_to_unlockedlocks304 assert factories.UNIT_SECRETHASH not in channel_state.partner_state.secrethashes_to_lockedlocks305 # Make sure that an emptyhash on chain reveal is rejected.306 block_number_prior_the_expiration = expiration - 2307 onchain_reveal = ContractReceiveSecretReveal(308 transaction_hash=factories.make_address(),309 secret_registry_address=factories.make_address(),310 secrethash=EMPTY_HASH_KECCAK,311 secret=EMPTY_HASH,312 block_number=block_number_prior_the_expiration,313 )314 onchain_secret_reveal_iteration = target.state_transition(315 offchain_secret_reveal_iteration.new_state,316 onchain_reveal,317 channel_state,318 pseudo_random_generator,319 block_number_prior_the_expiration,320 )321 unlocked_onchain = channel_state.partner_state.secrethashes_to_onchain_unlockedlocks322 assert EMPTY_HASH_KECCAK not in unlocked_onchain323 # now let's go for the actual secret324 onchain_reveal.secret = secret325 onchain_reveal.secrethash = factories.UNIT_SECRETHASH326 onchain_secret_reveal_iteration = target.state_transition(327 offchain_secret_reveal_iteration.new_state,328 onchain_reveal,329 channel_state,330 pseudo_random_generator,331 block_number_prior_the_expiration,332 )333 unlocked_onchain = channel_state.partner_state.secrethashes_to_onchain_unlockedlocks334 assert factories.UNIT_SECRETHASH in unlocked_onchain335 # Check that after we register a lock on-chain handling the block again will336 # not cause us to attempt an onchain re-register337 extra_block_handle_transition = target.handle_block(338 onchain_secret_reveal_iteration.new_state,339 channel_state,340 block_number_prior_the_expiration + 1,341 )342 assert len(extra_block_handle_transition.events) == 0343def test_handle_block():344 """ Increase the block number. """345 initiator = factories.HOP6346 our_address = factories.ADDR347 amount = 3348 block_number = 1349 pseudo_random_generator = random.Random()350 from_channel, state = make_target_state(351 our_address,352 amount,353 block_number,354 initiator,355 )356 new_block = Block(357 block_number=block_number + 1,358 gas_limit=1,359 block_hash=factories.make_transaction_hash(),360 )361 iteration = target.state_transition(362 state,363 new_block,364 from_channel,365 pseudo_random_generator,366 new_block.block_number,367 )368 assert iteration.new_state369 assert not iteration.events370def test_handle_block_equal_block_number():371 """ Nothing changes. """372 initiator = factories.HOP6373 our_address = factories.ADDR374 amount = 3375 block_number = 1376 pseudo_random_generator = random.Random()377 from_channel, state = make_target_state(378 our_address,379 amount,380 block_number,381 initiator,382 )383 new_block = Block(384 block_number=block_number,385 gas_limit=1,386 block_hash=factories.make_transaction_hash(),387 )388 iteration = target.state_transition(389 state,390 new_block,391 from_channel,392 pseudo_random_generator,393 new_block.block_number,394 )395 assert iteration.new_state396 assert not iteration.events397def test_handle_block_lower_block_number():398 """ Nothing changes. """399 initiator = factories.HOP6400 our_address = factories.ADDR401 amount = 3402 block_number = 10403 pseudo_random_generator = random.Random()404 from_channel, state = make_target_state(405 our_address,406 amount,407 block_number,408 initiator,409 )410 new_block = Block(411 block_number=block_number - 1,412 gas_limit=1,413 block_hash=factories.make_transaction_hash(),414 )415 iteration = target.state_transition(416 state,417 new_block,418 from_channel,419 pseudo_random_generator,420 new_block.block_number,421 )422 assert iteration.new_state423 assert not iteration.events424def test_state_transition():425 """ Happy case testing. """426 lock_amount = 7427 block_number = 1428 initiator = factories.HOP6429 pseudo_random_generator = random.Random()430 our_balance = 100431 our_address = factories.make_address()432 partner_balance = 130433 from_channel = factories.make_channel(434 our_address=our_address,435 our_balance=our_balance,436 partner_address=UNIT_TRANSFER_SENDER,437 partner_balance=partner_balance,438 )439 from_route = factories.route_from_channel(from_channel)440 expiration = block_number + from_channel.settle_timeout - from_channel.reveal_timeout441 from_transfer = factories.make_signed_transfer_for(442 from_channel,443 lock_amount,444 initiator,445 our_address,446 expiration,447 UNIT_SECRET,448 )449 init = ActionInitTarget(450 from_route,451 from_transfer,452 )453 init_transition = target.state_transition(454 None,455 init,456 from_channel,457 pseudo_random_generator,458 block_number,459 )460 assert init_transition.new_state is not None461 assert init_transition.new_state.route == from_route462 assert init_transition.new_state.transfer == from_transfer463 first_new_block = Block(464 block_number=block_number + 1,465 gas_limit=1,466 block_hash=factories.make_transaction_hash(),467 )468 first_block_iteration = target.state_transition(469 init_transition.new_state,470 first_new_block,471 from_channel,472 pseudo_random_generator,473 first_new_block.block_number,474 )475 secret_reveal = ReceiveSecretReveal(factories.UNIT_SECRET, initiator)476 reveal_iteration = target.state_transition(477 first_block_iteration.new_state,478 secret_reveal,479 from_channel,480 pseudo_random_generator,481 first_new_block.block_number,482 )483 assert reveal_iteration.events484 second_new_block = Block(485 block_number=block_number + 2,486 gas_limit=1,487 block_hash=factories.make_transaction_hash(),488 )489 iteration = target.state_transition(490 init_transition.new_state,491 second_new_block,492 from_channel,493 pseudo_random_generator,494 second_new_block.block_number,495 )496 assert not iteration.events497 nonce = from_transfer.balance_proof.nonce + 1498 transferred_amount = lock_amount499 locksroot = EMPTY_MERKLE_ROOT500 invalid_message_hash = b'\x00' * 32501 locked_amount = 0502 balance_proof = factories.make_signed_balance_proof(503 nonce,504 transferred_amount,505 locked_amount,506 from_channel.token_network_identifier,507 from_route.channel_identifier,508 locksroot,509 invalid_message_hash,510 UNIT_TRANSFER_PKEY,511 UNIT_TRANSFER_SENDER,512 )513 balance_proof_state_change = ReceiveUnlock(514 message_identifier=random.randint(0, UINT64_MAX),515 secret=UNIT_SECRET,516 balance_proof=balance_proof,517 )518 proof_iteration = target.state_transition(519 init_transition.new_state,520 balance_proof_state_change,521 from_channel,522 pseudo_random_generator,523 block_number + 2,524 )525 assert proof_iteration.new_state is None526def test_target_reject_keccak_empty_hash():527 lock_amount = 7528 block_number = 1529 initiator = factories.HOP6530 pseudo_random_generator = random.Random()531 our_balance = 100532 our_address = factories.make_address()533 partner_balance = 130534 from_channel = factories.make_channel(535 our_address=our_address,536 our_balance=our_balance,537 partner_address=UNIT_TRANSFER_SENDER,538 partner_balance=partner_balance,539 )540 from_route = factories.route_from_channel(from_channel)541 expiration = block_number + from_channel.settle_timeout - from_channel.reveal_timeout542 from_transfer = factories.make_signed_transfer_for(543 channel_state=from_channel,544 amount=lock_amount,545 initiator=initiator,546 target=our_address,547 expiration=expiration,548 secret=EMPTY_HASH,549 allow_invalid=True,550 )551 init = ActionInitTarget(552 route=from_route,553 transfer=from_transfer,554 )555 init_transition = target.state_transition(556 None,557 init,558 from_channel,559 pseudo_random_generator,560 block_number,561 )562 assert init_transition.new_state is None563def test_target_receive_lock_expired():564 lock_amount = 7565 block_number = 1566 initiator = factories.HOP6567 pseudo_random_generator = random.Random()568 our_balance = 100569 our_address = factories.make_address()570 partner_balance = 130571 from_channel = factories.make_channel(572 our_address=our_address,573 our_balance=our_balance,574 partner_address=UNIT_TRANSFER_SENDER,575 partner_balance=partner_balance,576 )577 from_route = factories.route_from_channel(from_channel)578 expiration = block_number + from_channel.settle_timeout - from_channel.reveal_timeout579 from_transfer = factories.make_signed_transfer_for(580 channel_state=from_channel,581 amount=lock_amount,582 initiator=initiator,583 target=our_address,584 expiration=expiration,585 secret=UNIT_SECRET,586 )587 init = ActionInitTarget(588 from_route,589 from_transfer,590 )591 init_transition = target.state_transition(592 None,593 init,594 from_channel,595 pseudo_random_generator,596 block_number,597 )598 assert init_transition.new_state is not None599 assert init_transition.new_state.route == from_route600 assert init_transition.new_state.transfer == from_transfer601 balance_proof = factories.make_signed_balance_proof(602 2,603 from_transfer.balance_proof.transferred_amount,604 0,605 from_transfer.balance_proof.token_network_identifier,606 from_channel.identifier,607 EMPTY_MERKLE_ROOT,608 from_transfer.lock.secrethash,609 sender_address=UNIT_TRANSFER_SENDER,610 private_key=UNIT_TRANSFER_PKEY,611 )612 lock_expired_state_change = ReceiveLockExpired(613 balance_proof=balance_proof,614 secrethash=from_transfer.lock.secrethash,615 message_identifier=1,616 )617 block_before_confirmed_expiration = expiration + DEFAULT_NUMBER_OF_BLOCK_CONFIRMATIONS - 1618 iteration = target.state_transition(619 init_transition.new_state,620 lock_expired_state_change,621 from_channel,622 pseudo_random_generator,623 block_before_confirmed_expiration,624 )625 assert not must_contain_entry(iteration.events, SendProcessed, {})626 block_lock_expired = block_before_confirmed_expiration + 1627 iteration = target.state_transition(628 init_transition.new_state,629 lock_expired_state_change,630 from_channel,631 pseudo_random_generator,632 block_lock_expired,633 )634 assert must_contain_entry(iteration.events, SendProcessed, {})635def test_target_lock_is_expired_if_secret_is_not_registered_onchain():636 lock_amount = 7637 block_number = 1638 initiator = factories.HOP6639 pseudo_random_generator = random.Random()640 our_balance = 100641 our_address = factories.make_address()642 partner_balance = 130643 from_channel = factories.make_channel(644 our_address=our_address,645 our_balance=our_balance,646 partner_address=UNIT_TRANSFER_SENDER,647 partner_balance=partner_balance,648 )649 from_route = factories.route_from_channel(from_channel)650 expiration = block_number + from_channel.settle_timeout - from_channel.reveal_timeout651 from_transfer = factories.make_signed_transfer_for(652 from_channel,653 lock_amount,654 initiator,655 our_address,656 expiration,657 UNIT_SECRET,658 )659 init = ActionInitTarget(660 from_route,661 from_transfer,662 )663 init_transition = target.state_transition(...
transfer.py
Source:transfer.py
...235 # skipping the netting channel register_transfer because the message is not236 # signed237 from_channel.our_state.register_direct_transfer(direct_transfer_message)238 to_channel.partner_state.register_direct_transfer(direct_transfer_message)239def make_direct_transfer_from_channel(block_number, from_channel, partner_channel, amount, pkey):240 """ Helper to create and register a direct transfer from `from_channel` to241 `partner_channel`.242 """243 identifier = from_channel.get_next_nonce()244 direct_transfer_msg = from_channel.create_directtransfer(245 amount,246 identifier=identifier,247 )248 address = privatekey_to_address(pkey)249 sign_key = PrivateKey(pkey)250 direct_transfer_msg.sign(sign_key, address)251 # if this fails it's not the right key for the current `from_channel`252 assert direct_transfer_msg.sender == from_channel.our_state.address253 from_channel.register_transfer(...
core.py
Source:core.py
...137 with self.channels_lock:138 listeners = self.channels.get(channel, set())139 listeners.add(listener)140 self.channels[channel] = listeners141 def unsubscribe_from_channel(self, channel, listener):142 logger.debug("Unsubscribing from {}: {}".format(channel, listener))143 with self.channels_lock:144 # TODO: And if some key error occurs?145 listeners = self.channels[channel]146 listeners.remove(listener)147 def create_message(self, from_channel, to_channel, message_type, *args):148 # TODO: This might also write into a queue, which a thread, representing149 # the MD and/or SS, reads from, isolating the Agent thread to otherwise150 # keep in just the agent module.151 logger.debug("Creating message: {} -> {}: {}".format(152 from_channel, to_channel, message_type,153 ))154 with self.channels_lock:155 listeners = self.channels[to_channel].copy()...
lambda_function.py
Source:lambda_function.py
1#2# Main Lambda Function for the Security Minion3#4import os5import time6import re7import json8import urllib9import datetime10import boto311import base6412import logFunctions13import secretsManagement14import settings15import slackFunctions16verificationToken = os.environ.get('VerificationToken')17isBase64Encoded = False18def main(event, context):19 return_headers = {}20 output = ''21 slack_event = {"type": ""}22 try: # load the request context23 secretsManagement.requestContext = event.get('requestContext', {})24 except:25 secretsManagement.requestContext = {}26 try: # load the GET parameters27 get_params = event.get('queryStringParameters', '{}')28 except:29 get_params = {}30 try: # load JSON post31 jsonbody = json.loads(event.get('body', {}))32 except Exception as e:33 jsonbody = {}34 try: # it's not JSON. lets try to decode it as URL with multiple query string 35 if 'body' in event:36 bodyString = event['body']37 else:38 bodyString = ''39 if ('isBase64Encoded' in event) and event['isBase64Encoded']:40 slack_event = base64.decodestring(bytes(bodyString, 'utf-8')).decode('utf-8')41 slack_event = urllib.parse.parse_qs(slack_event)42 else:43 slack_event = urllib.parse.parse_qs(bodyString)44 except Exception as e:45 logFunctions.log('Invalid data in body '+str(e))46 47 if (type(get_params) is dict) and (get_params.get('presigned', '') != '1'): # the user wants to upload a file48 return_headers = { 'Content-Type': 'text/html' }49 output = secretsManagement.HTML_uploadForm(get_params['presigned'])50 elif ('type' in jsonbody) and (jsonbody['type'] == "url_verification"):51 output = jsonbody["challenge"]52 # log the request53 logFunctions.log('New challenge replied')54 elif not('token' in slack_event) or not(verificationToken in slack_event['token']):55 output = ''56 logFunctions.log('invalid token')57 elif "bot_id" in slack_event:58 logFunctions.log('Ignoring event from bot')59 elif 'X-Slack-Retry-Num' in event['headers']:60 logFunctions.log('Ignoring a duplicated event')61 elif ("command" in slack_event) and ('/'+settings.botName in slack_event["command"]):62 output = ""63 text = ''64 from_user = ''65 from_team = ''66 from_channel = ''67 from_username = ''68 from_domain = ''69 try:70 if len(slack_event["text"]) > 0:71 text = slack_event["text"][0].strip()72 if len(slack_event["user_id"]) > 0:73 from_user = slack_event["user_id"][0]74 if len(slack_event["team_id"]) > 0:75 from_team = slack_event["team_id"][0]76 if len(slack_event["channel_id"]) > 0:77 from_channel = slack_event["channel_id"][0]78 #response_url = slack_event["response_url"][0]79 if len(slack_event["user_name"]) > 0:80 from_username = slack_event["user_name"][0]81 if len(slack_event["team_domain"]) > 0:82 from_domain = slack_event["team_domain"][0]83 from_user = from_username+'@'+from_domain84 # check for commands85 # get help86 if text.lower() == "help":87 output = settings.help_message88 # get the extended secrets list89 elif text == "??":90 output = secretsManagement.get_secret_list_extended(from_team, from_channel)91 # get the secrets list92 elif text == "?":93 output = secretsManagement.get_secret_list(from_team, from_channel)94 elif text.lower() == "info":95 output = 'id: '+ from_user + ' team domain: '+ from_domain+ ' channel: '+from_channel+' ' + slackFunctions.get_channel_info(from_channel, True)96 elif text.lower().startswith("delete "):97 output = secretsManagement.delete_secret(from_user, from_team, from_channel, text[7:].lower().strip())98 elif text.lower() == "delete":99 output = "missing name to delete"100 elif text.lower().startswith("file "):101 output = secretsManagement.set_secret(from_user, from_team, from_channel, text[5:].lower().strip(), '+' )102 elif text.lower() == "file":103 output = "missing file name"104 # create a new secret105 elif '=' in text:106 newSecretName, newSecretValue = '', ''107 values = text.strip().split(sep='=', maxsplit=1)108 newSecretName = values[0]109 if len(values) > 1:110 newSecretValue = values[1].strip()111 if bool(newSecretName): # the name is not empty112 if newSecretName in settings.reservedWords:113 output = "Sorry, '"+newSecretName+"' is a reserverd word."114 else:115 if bool(newSecretValue.strip()): # the secret is not empty116 old_secret_value = secretsManagement.aws_secretsmanager_get_secret( from_team+'.'+from_channel+'.'+newSecretName.strip() )117 if len(old_secret_value) > 0: # if exists, update it118 output = secretsManagement.update_secret(from_user, from_team, from_channel, newSecretName.strip(), newSecretValue, old_secret_value )119 else: # create a new secret120 output = secretsManagement.set_secret(from_user, from_team, from_channel, newSecretName.strip(), newSecretValue )121 else: # if empty, just delete it122 output = secretsManagement.delete_secret(from_user, from_team, from_channel, newSecretName.strip())123 else:124 output = "Your secret needs a name"125 # get a secret value126 else:127 SecretName = text.strip()128 output = secretsManagement.get_secret(from_user, from_team, from_channel, SecretName)129 except Exception as e:130 logFunctions.log("Error or missing paramter from slack. "+str(e))131 logFunctions.log(slack_event)132 return {133 'statusCode': 200,134 'body': output,135 'headers': return_headers,136 'isBase64Encoded': isBase64Encoded137 }138if __name__ == '__main__':...
LambdaTest’s Playwright tutorial will give you a broader idea about the Playwright automation framework, its unique features, and use cases with examples to exceed your understanding of Playwright testing. This tutorial will give A to Z guidance, from installing the Playwright framework to some best practices and advanced concepts.
Get 100 minutes of automation test minutes FREE!!