How to use expect_request method in Playwright Python

Best Python code snippet using playwright-python

test_integration.py

Source:test_integration.py Github

copy

Full Screen

...11from random import choice, randint12# Approve13def test_create_reward_normal(client, verified_user, program, deposit_wallet, deposit_account, httpserver):14 settings.TREASURY_URL = httpserver.url_for("")[:-1]15 httpserver.expect_request(f"/wallets", query_string=f"traderId={MainTrader.objects.get(user__email=verified_user).trader_id}&currencyCode=DGTX&kind=DEPOSIT").respond_with_json([{"id": 100, "traderId": MainTrader.objects.get(user__email=verified_user).trader_id, "currency": {"id": 1, "currencyPairId": 0, "name": "Digitex", "code": "DGTX", "scale": 4, "sellPrice": "1.00000000000000000000", "buyPrice": "1.00000000000000000000"}, "balance": "10000.00000000000000000000", "blockchainAddress": "0x0A190CE3Ba33e0D2c999f35E58AF1202b064fe2F"}])16 httpserver.expect_request("/wallets/100/transfer/to-main", query_string=f"wallet2={deposit_account.treasury_id}&amount=10").respond_with_json([])17 total_stack_amount_before_stack = deposit_wallet.balance18 total_account_stack_amount_before_stack = deposit_account.balance19 Transaction.objects.create(user=User.objects.get(email=verified_user), currency_id=1, amount=10, wallet=deposit_wallet, account=deposit_account, type=Transaction.Type.STAKE, program=program)20 tasks.calculating_rewards()21 total_stack_amount_after_stack = Wallet.objects.get(program=program, user=deposit_wallet.user, type=Wallet.Type.DEPOSIT).balance22 total_account_stack_amount_after_stack = Account.objects.get(pk=deposit_account.id).balance23 assert total_stack_amount_after_stack == total_stack_amount_before_stack + 10, 'Total user wallet stack amount has been changed'24 assert total_account_stack_amount_after_stack == total_account_stack_amount_before_stack + 10, 'Total program account stack amount has been changed'25# Approve26def test_create_reward_empty_stack_zero(client, verified_user, program, deposit_wallet, deposit_account, httpserver):27 settings.TREASURY_URL = httpserver.url_for("")[:-1]28 httpserver.expect_request(f"/wallets", query_string=f"traderId={MainTrader.objects.get(user__email=verified_user).trader_id}&currencyCode=DGTX&kind=DEPOSIT").respond_with_json([{"id": 100, "traderId": MainTrader.objects.get(user__email=verified_user).trader_id, "currency": {"id": 1, "currencyPairId": 0, "name": "Digitex", "code": "DGTX", "scale": 4, "sellPrice": "1.00000000000000000000", "buyPrice": "1.00000000000000000000"}, "balance": "10000.00000000000000000000", "blockchainAddress": "0x0A190CE3Ba33e0D2c999f35E58AF1202b064fe2F"}])29 httpserver.expect_request("/wallets/100/transfer/to-main", query_string=f"wallet2={deposit_account.treasury_id}&amount=0").respond_with_json([])30 total_stack_amount_before_stack = deposit_wallet.balance31 total_account_stack_amount_before_stack = deposit_account.balance32 tx = Transaction.objects.create(user=User.objects.get(email=verified_user), currency_id=1, amount=0, wallet=deposit_wallet, account=deposit_account, type=Transaction.Type.STAKE, program=program)33 assert tx.status == Transaction.Status.SUCCESS34 tasks.calculating_rewards()35 total_stack_amount_after_stack = Wallet.objects.get(program=program, user=deposit_wallet.user, type=Wallet.Type.DEPOSIT).balance36 total_account_stack_amount_after_stack = Account.objects.get(pk=deposit_account.id).balance37 assert total_stack_amount_before_stack == total_stack_amount_after_stack, 'Total stack amount has been changed'38 assert total_account_stack_amount_after_stack == total_account_stack_amount_before_stack, 'Total program account stack amount has been changed'39# Approve40def test_create_reward_2_iterations(client, verified_user, httpserver):41 program = Program.objects.create(slug="".join(choice(string.ascii_letters + string.digits) for x in range(randint(5, 10))), transaction_currency_id=1, reward_currency_id=2, is_enable=True, is_visible=True, begin_date=timezone.now()-timedelta(days=5), emit_duration=timedelta(hours=1), iteration=timedelta(minutes=30))42 deposit_account = Account.objects.get(type=Account.Type.DEPOSIT, program=program)43 deposit_account.balance = 10044 deposit_account.treasury_id = 145 deposit_account.save()46 reward_account = Account.objects.get(type=Account.Type.REWARD, program=program)47 reward_account.balance = 148 reward_account.treasury_id = 249 reward_account.save()50 deposit_wallet = Wallet.objects.create(type=Wallet.Type.DEPOSIT, balance=100, program=program, user=User.objects.get(email=verified_user))51 reward_wallet = Wallet.objects.create(type=Wallet.Type.REWARD, balance=100, program=program, user=User.objects.get(email=verified_user))52 program.reward_currency.custodian_scale = 853 program.reward_currency.save()54 settings.TREASURY_URL = httpserver.url_for("")[:-1]55 httpserver.expect_request(f"/wallets", query_string=f"traderId={MainTrader.objects.get(user__email=verified_user).trader_id}&currencyCode=DGTX&kind=DEPOSIT").respond_with_json([{"id": 100, "traderId": MainTrader.objects.get(user__email=verified_user).trader_id, "currency": {"id": 1, "currencyPairId": 0, "name": "Digitex", "code": "DGTX", "scale": 4, "sellPrice": "1.00000000000000000000", "buyPrice": "1.00000000000000000000"}, "balance": "10000.00000000000000000000", "blockchainAddress": "0x0A190CE3Ba33e0D2c999f35E58AF1202b064fe2F"}])56 httpserver.expect_request("/wallets/100/transfer/to-main", query_string=f"wallet2={deposit_account.treasury_id}&amount=10000").respond_with_json([])57 httpserver.expect_request("/wallets/100/transfer/to-main", query_string=f"wallet2={deposit_account.treasury_id}&amount=15").respond_with_json([])58 tx_1 = Transaction.objects.create(user=User.objects.get(email=verified_user), currency_id=1, amount=10000, wallet=deposit_wallet, account=deposit_account, type=Transaction.Type.STAKE, program=program)59 assert tx_1.status == Transaction.Status.SUCCESS60 Transaction.objects.filter(id=tx_1.id).update(created_at=program.begin_date - timedelta(minutes=30))61 tasks.calculating_rewards()62 real_reward = Reward.objects.get(program=program, user__email=verified_user)63 real_reward_amount = real_reward.amount64 to_1_reward = 0.565 assert real_reward_amount == to_1_reward, 'Reward account balance is not equal to reward computing amount'66 assert Wallet.objects.get(user__email=verified_user, type=Wallet.Type.REWARD, program=program).balance == 100.567 tx_2 = Transaction.objects.create(user=User.objects.get(email=verified_user), currency_id=1, amount=15, wallet=deposit_wallet, account=deposit_account, type=Transaction.Type.STAKE, program=program)68 assert tx_2.status == Transaction.Status.SUCCESS69 Transaction.objects.filter(id=tx_2.id).update(created_at=program.begin_date + timedelta(minutes=30))70 tasks.calculating_rewards()71 real_reward = Reward.objects.filter(program=program, user__email=verified_user)[0]72 real_reward_amount = real_reward.amount73 to_2_reward = 0.574 assert real_reward_amount == to_2_reward, 'Reward account balance is not equal to reward computing amount'75 assert Wallet.objects.get(user__email=verified_user, type=Wallet.Type.REWARD).balance == 10176# Approve77def test_create_reward_transaction_at_the_same_time_with_iteration(client, verified_user, httpserver):78 program = Program.objects.create(slug="".join(choice(string.ascii_letters + string.digits) for x in range(randint(5, 10))), transaction_currency_id=1, reward_currency_id=2, is_enable=True, is_visible=True, begin_date=timezone.now()-timedelta(days=5), emit_duration=timedelta(hours=1), iteration=timedelta(minutes=30))79 deposit_account = Account.objects.get(type=Account.Type.DEPOSIT, program=program)80 deposit_account.balance = 10081 deposit_account.treasury_id = 182 deposit_account.save()83 reward_account = Account.objects.get(type=Account.Type.REWARD, program=program)84 reward_account.balance = 185 reward_account.treasury_id = 286 reward_account.save()87 deposit_wallet = Wallet.objects.create(type=Wallet.Type.DEPOSIT, balance=100, program=program, user=User.objects.get(email=verified_user))88 reward_wallet = Wallet.objects.create(type=Wallet.Type.REWARD, balance=100, program=program, user=User.objects.get(email=verified_user))89 program.reward_currency.custodian_scale = 890 program.reward_currency.save()91 program = Program.objects.create(slug="".join(choice(string.ascii_letters + string.digits) for x in range(randint(5, 10))), transaction_currency_id=1, reward_currency_id=2, is_enable=True, is_visible=True, begin_date=timezone.now()-timedelta(days=5), emit_duration=timedelta(hours=1), iteration=timedelta(minutes=30))92 deposit_account = Account.objects.get(type=Account.Type.DEPOSIT, program=program)93 deposit_account.balance = 10094 deposit_account.treasury_id = 195 deposit_account.save()96 reward_account = Account.objects.get(type=Account.Type.REWARD, program=program)97 reward_account.balance = 198 reward_account.treasury_id = 299 reward_account.save()100 deposit_wallet = Wallet.objects.create(type=Wallet.Type.DEPOSIT, balance=100, program=program, user=User.objects.get(email=verified_user))101 reward_wallet = Wallet.objects.create(type=Wallet.Type.REWARD, balance=100, program=program, user=User.objects.get(email=verified_user))102 program.reward_currency.custodian_scale = 8103 program.reward_currency.save()104 settings.TREASURY_URL = httpserver.url_for("")[:-1]105 httpserver.expect_request(f"/wallets", query_string=f"traderId={MainTrader.objects.get(user__email=verified_user).trader_id}&currencyCode=DGTX&kind=DEPOSIT").respond_with_json([{"id": 100, "traderId": MainTrader.objects.get(user__email=verified_user).trader_id, "currency": {"id": 1, "currencyPairId": 0, "name": "Digitex", "code": "DGTX", "scale": 4, "sellPrice": "1.00000000000000000000", "buyPrice": "1.00000000000000000000"}, "balance": "10000.00000000000000000000", "blockchainAddress": "0x0A190CE3Ba33e0D2c999f35E58AF1202b064fe2F"}])106 httpserver.expect_request("/wallets/100/transfer/to-main", query_string=f"wallet2={deposit_account.treasury_id}&amount=10").respond_with_json([])107 total_stack_amount_before_stack = deposit_wallet.balance108 tx = Transaction.objects.create(user=User.objects.get(email=verified_user), currency_id=1, amount=10, wallet=deposit_wallet, account=deposit_account, type=Transaction.Type.STAKE, program=program)109 Transaction.objects.filter(id=tx.id).update(created_at=program.begin_date)110 assert tx.status == Transaction.Status.SUCCESS111 total_stack_amount_after_stack = Wallet.objects.get(program=program, user__email=verified_user, type=Wallet.Type.DEPOSIT).balance112 assert total_stack_amount_before_stack + 10 == total_stack_amount_after_stack, 'Total stack amount has been changed'113 tasks.calculating_rewards()114 assert Reward.objects.get(program=program, user__email=verified_user).amount == 0.5115 assert Wallet.objects.get(user__email=verified_user, program=program, type=Wallet.Type.REWARD).balance == 100.5116# Approve117def test_create_reward(client, verified_user, httpserver):118 program = Program.objects.create(slug="".join(choice(string.ascii_letters + string.digits) for x in range(randint(5, 10))), transaction_currency_id=1, reward_currency_id=2, is_enable=True, is_visible=True, begin_date=timezone.now()-timedelta(days=5), emit_duration=timedelta(hours=1), iteration=timedelta(minutes=30))119 deposit_account = Account.objects.get(type=Account.Type.DEPOSIT, program=program)120 deposit_account.balance = 100121 deposit_account.treasury_id = 1122 deposit_account.save()123 reward_account = Account.objects.get(type=Account.Type.REWARD, program=program)124 reward_account.balance = 1125 reward_account.treasury_id = 2126 reward_account.save()127 deposit_wallet = Wallet.objects.create(type=Wallet.Type.DEPOSIT, balance=100, program=program, user=User.objects.get(email=verified_user))128 reward_wallet = Wallet.objects.create(type=Wallet.Type.REWARD, balance=100, program=program, user=User.objects.get(email=verified_user))129 program.reward_currency.custodian_scale = 8130 program.reward_currency.save()131 settings.TREASURY_URL = httpserver.url_for("")[:-1]132 httpserver.expect_request(f"/wallets", query_string=f"traderId={MainTrader.objects.get(user__email=verified_user).trader_id}&currencyCode=DGTX&kind=DEPOSIT").respond_with_json([{"id": 100, "traderId": MainTrader.objects.get(user__email=verified_user).trader_id, "currency": {"id": 1, "currencyPairId": 0, "name": "Digitex", "code": "DGTX", "scale": 4, "sellPrice": "1.00000000000000000000", "buyPrice": "1.00000000000000000000"}, "balance": "10000.00000000000000000000", "blockchainAddress": "0x0A190CE3Ba33e0D2c999f35E58AF1202b064fe2F"}])133 httpserver.expect_request("/wallets/100/transfer/to-main", query_string=f"wallet2={deposit_account.treasury_id}&amount=10").respond_with_json([])134 tx = Transaction.objects.create(user=User.objects.get(email=verified_user), currency_id=1, amount=10, wallet=deposit_wallet, account=deposit_account, type=Transaction.Type.STAKE, program=program)135 Transaction.objects.filter(id=tx.id).update(created_at=program.begin_date - timedelta(minutes=30))136 assert tx.status == Transaction.Status.SUCCESS137 tasks.calculating_rewards()138 real_reward = Reward.objects.get(program=program, user__email=verified_user)139 real_reward_amount = real_reward.amount140 # Staked half of time141 # (10 / 10) * (1800 / 3600) * 100142 to_reward = 0.5143 assert real_reward_amount == to_reward, 'Reward account balance is not equal to reward computing amount'144 assert Wallet.objects.get(user__email=verified_user, type=Wallet.Type.REWARD).balance == 100.5145# Approve146def test_create_claim_reward(client, verified_user, httpserver):147 program = Program.objects.create(slug="".join(choice(string.ascii_letters + string.digits) for x in range(randint(5, 10))), transaction_currency_id=1, reward_currency_id=2, is_enable=True, is_visible=True, begin_date=timezone.now()-timedelta(days=5), emit_duration=timedelta(hours=1), iteration=timedelta(minutes=30))148 deposit_account = Account.objects.get(type=Account.Type.DEPOSIT, program=program)149 deposit_account.balance = 100150 deposit_account.treasury_id = 1151 deposit_account.save()152 reward_account = Account.objects.get(type=Account.Type.REWARD, program=program)153 reward_account.balance = 1154 reward_account.treasury_id = 2155 reward_account.save()156 deposit_wallet = Wallet.objects.create(type=Wallet.Type.DEPOSIT, balance=100, program=program, user=User.objects.get(email=verified_user))157 reward_wallet = Wallet.objects.create(type=Wallet.Type.REWARD, balance=100, program=program, user=User.objects.get(email=verified_user))158 program.reward_currency.custodian_scale = 8159 program.reward_currency.save()160 settings.TREASURY_URL = httpserver.url_for("")[:-1]161 httpserver.expect_request(f"/wallets", query_string=f"traderId={MainTrader.objects.get(user__email=verified_user).trader_id}&currencyCode=DGTX&kind=DEPOSIT").respond_with_json([{"id": 100, "traderId": MainTrader.objects.get(user__email=verified_user).trader_id, "currency": {"id": 1, "currencyPairId": 0, "name": "Digitex", "code": "DGTX", "scale": 4, "sellPrice": "1.00000000000000000000", "buyPrice": "1.00000000000000000000"}, "balance": "10000.00000000000000000000", "blockchainAddress": "0x0A190CE3Ba33e0D2c999f35E58AF1202b064fe2F"}])162 httpserver.expect_request("/wallets/100/transfer/to-main", query_string=f"wallet2=1&amount=10").respond_with_json([])163 httpserver.expect_request("/wallets/2/transfer/to-main", query_string=f"wallet2=100&amount=0.5").respond_with_json([])164 tx = Transaction.objects.create(user=User.objects.get(email=verified_user), currency_id=1, amount=10, wallet=deposit_wallet, account=deposit_account, type=Transaction.Type.STAKE, program=program)165 Transaction.objects.filter(id=tx.id).update(created_at=program.begin_date - timedelta(minutes=30))166 assert tx.status == Transaction.Status.SUCCESS167 tasks.calculating_rewards()168 real_reward = Reward.objects.get(program=program, user__email=verified_user)169 real_reward_amount = real_reward.amount170 assert real_reward_amount == 0.5171 account_balance_before_claim_reward = reward_account.balance172 claim_reward = ClaimReward.objects.create(user=User.objects.get(email=verified_user), program=program, currency_id=1, amount=0.5, wallet=reward_wallet, account=reward_account)173 assert claim_reward.status == ClaimReward.Status.SUCCESS174# Approve175def test_create_stack(client, verified_user, program, deposit_wallet, deposit_account, httpserver):176 settings.TREASURY_URL = httpserver.url_for("")[:-1]177 httpserver.expect_request(f"/wallets", query_string=f"traderId={MainTrader.objects.get(user__email=verified_user).trader_id}&currencyCode=DGTX&kind=DEPOSIT").respond_with_json([{"id": 100, "traderId": MainTrader.objects.get(user__email=verified_user).trader_id, "currency": {"id": 1, "currencyPairId": 0, "name": "Digitex", "code": "DGTX", "scale": 4, "sellPrice": "1.00000000000000000000", "buyPrice": "1.00000000000000000000"}, "balance": "10000.00000000000000000000", "blockchainAddress": "0x0A190CE3Ba33e0D2c999f35E58AF1202b064fe2F"}])178 httpserver.expect_request("/wallets/100/transfer/to-main", query_string=f"wallet2={deposit_account.treasury_id}&amount=10").respond_with_json([])179 account_balance_before_stack = deposit_account.balance180 total_stack_balance_before_stack = deposit_wallet.balance181 transaction = Transaction.objects.create(user=User.objects.get(email=verified_user), currency_id=1, amount=10, wallet=deposit_wallet, account=deposit_account, type=Transaction.Type.STAKE, program=program)182 assert transaction.status == Transaction.Status.SUCCESS183 account = Account.objects.get(program=program, type=Account.Type.DEPOSIT)184 total_stack = Wallet.objects.get(program=program, user__email=verified_user, type=Wallet.Type.DEPOSIT).balance185 assert account.balance == account_balance_before_stack + 10, 'Account balance before stack is not biggest than account balance after stack on amount sum.'186 assert total_stack == total_stack_balance_before_stack + 10, 'Total stack balance before stack is not biggest than total stack balance after stack on amount sum.'187# Approve188def test_create_unstack(client, verified_user, program, deposit_wallet, deposit_account, httpserver):189 settings.TREASURY_URL = httpserver.url_for("")[:-1]190 httpserver.expect_request(f"/wallets", query_string=f"traderId={MainTrader.objects.get(user__email=verified_user).trader_id}&currencyCode=DGTX&kind=DEPOSIT").respond_with_json([{"id": 100, "traderId": MainTrader.objects.get(user__email=verified_user).trader_id, "currency": {"id": 1, "currencyPairId": 0, "name": "Digitex", "code": "DGTX", "scale": 4, "sellPrice": "1.00000000000000000000", "buyPrice": "1.00000000000000000000"}, "balance": "10000.00000000000000000000", "blockchainAddress": "0x0A190CE3Ba33e0D2c999f35E58AF1202b064fe2F"}])191 httpserver.expect_request(f"/wallets/{deposit_account.treasury_id}/transfer/to-main", query_string=f"wallet2=100&amount=10").respond_with_json([])192 account_balance_before_unstack = deposit_account.balance193 total_stack_balance_before_unstack = deposit_wallet.balance194 transaction = Transaction.objects.create(user=User.objects.get(email=verified_user), currency_id=1, amount=10, wallet=deposit_wallet, account=deposit_account, type=Transaction.Type.UNSTAKE, program=program)195 assert transaction.status == Transaction.Status.SUCCESS196 account = Account.objects.get(program=program, type=Account.Type.DEPOSIT)197 total_stack = Wallet.objects.get(program=program, user__email=verified_user, type=Wallet.Type.DEPOSIT).balance198 assert account.balance + 10 == account_balance_before_unstack, 'Account balance before unstack is not smaller than account balance after unstack on amount sum.'199 assert total_stack + 10 == total_stack_balance_before_unstack, 'Total stack balance before unstack is not smaller than total stack balance after unstack on amount sum.'200# Approve201def test_create_stack_from_2_users(client, program, deposit_account, httpserver, django_user_model):202 user_1 = django_user_model.objects.create(username="user_1", password="12345")203 deposit_wallet, created = Wallet.objects.get_or_create(user=user_1, type=Wallet.Type.DEPOSIT, program=program)204 settings.TREASURY_URL = httpserver.url_for("")[:-1]205 httpserver.expect_request(f"/wallets", query_string=f"traderId={MainTrader.objects.get(user=user_1).trader_id}&currencyCode=DGTX&kind=DEPOSIT").respond_with_json([{"id": 100, "traderId": MainTrader.objects.get(user=user_1).trader_id, "currency": {"id": 1, "currencyPairId": 0, "name": "Digitex", "code": "DGTX", "scale": 4, "sellPrice": "1.00000000000000000000", "buyPrice": "1.00000000000000000000"}, "balance": "10000.00000000000000000000", "blockchainAddress": "0x0A190CE3Ba33e0D2c999f35E58AF1202b064fe2F"}])206 httpserver.expect_request(f"/wallets/100/transfer/to-main", query_string=f"wallet2=1&amount=10").respond_with_json([])207 account_balance_before_unstack = Account.objects.get(program=program, type=Account.Type.DEPOSIT).balance208 total_stack_balance_before_unstack = Wallet.objects.get(program=program, user=user_1, type=Wallet.Type.DEPOSIT).balance209 transaction = Transaction.objects.create(user=user_1, currency_id=1, amount=10, wallet=deposit_wallet, account=deposit_account, type=Transaction.Type.STAKE, program=program)210 assert transaction.status == Transaction.Status.SUCCESS211 account = Account.objects.get(program=program, type=Account.Type.DEPOSIT)212 total_stack = Wallet.objects.get(program=program, user=user_1, type=Wallet.Type.DEPOSIT).balance213 assert account.balance == account_balance_before_unstack + 10, 'Account balance before unstack is not smaller than account balance after unstack on amount sum.'214 assert total_stack == total_stack_balance_before_unstack + 10, 'Total stack balance before unstack is not smaller than total stack balance after unstack on amount sum.'215 httpserver.expect_request(f"/wallets", query_string=f"traderId={MainTrader.objects.get(user=user_1).trader_id}&currencyCode=DGTX&kind=DEPOSIT").respond_with_json([{"id": 100, "traderId": MainTrader.objects.get(user=user_1).trader_id, "currency": {"id": 1, "currencyPairId": 0, "name": "Digitex", "code": "DGTX", "scale": 4, "sellPrice": "1.00000000000000000000", "buyPrice": "1.00000000000000000000"}, "balance": "10000.00000000000000000000", "blockchainAddress": "0x0A190CE3Ba33e0D2c999f35E58AF1202b064fe2F"}])216 httpserver.expect_request(f"/wallets/100/transfer/to-main", query_string=f"wallet2=1&amount=15").respond_with_json([])217 account_balance_before_unstack = Account.objects.get(program=program, type=Account.Type.DEPOSIT).balance218 total_stack_balance_before_unstack = Wallet.objects.get(program=program, user=user_1, type=Wallet.Type.DEPOSIT).balance219 transaction = Transaction.objects.create(user=user_1, currency_id=1, amount=15, wallet=deposit_wallet, account=deposit_account, type=Transaction.Type.STAKE, program=program)220 assert transaction.status == Transaction.Status.SUCCESS221 account = Account.objects.get(program=program, type=Account.Type.DEPOSIT)222 total_stack = Wallet.objects.get(program=program, user=user_1, type=Wallet.Type.DEPOSIT).balance223 assert account.balance == account_balance_before_unstack + 15, 'Account balance before unstack is not smaller than account balance after unstack on amount sum.'224 assert total_stack == total_stack_balance_before_unstack + 15, 'Total stack balance before unstack is not smaller than total stack balance after unstack on amount sum.'225 user_2 = django_user_model.objects.create(username="user_2", password="12345", email='q@q.q')226 deposit_wallet, created = Wallet.objects.get_or_create(user=user_2, type=Wallet.Type.DEPOSIT, program=program)227 httpserver.expect_request(f"/wallets", query_string=f"traderId={MainTrader.objects.get(user=user_2).trader_id}&currencyCode=DGTX&kind=DEPOSIT").respond_with_json([{"id": 101, "traderId": MainTrader.objects.get(user=user_2).trader_id, "currency": {"id": 1, "currencyPairId": 0, "name": "Digitex", "code": "DGTX", "scale": 4, "sellPrice": "1.00000000000000000000", "buyPrice": "1.00000000000000000000"}, "balance": "10000.00000000000000000000", "blockchainAddress": "0x0A190CE3Ba33e0D2c999f35E58AF1202b064fe2F"}])228 httpserver.expect_request(f"/wallets/101/transfer/to-main", query_string=f"wallet2=1&amount=10").respond_with_json([])229 account_balance_before_unstack = Account.objects.get(program=program, type=Account.Type.DEPOSIT).balance230 total_stack_balance_before_unstack = Wallet.objects.get(program=program, user=user_2, type=Wallet.Type.DEPOSIT).balance231 transaction = Transaction.objects.create(user=user_2, currency_id=1, amount=10, wallet=deposit_wallet, account=deposit_account, type=Transaction.Type.STAKE, program=program)232 assert transaction.status == Transaction.Status.SUCCESS233 account = Account.objects.get(program=program, type=Account.Type.DEPOSIT)234 total_stack = Wallet.objects.get(program=program, user=user_2, type=Wallet.Type.DEPOSIT).balance235 assert account.balance == account_balance_before_unstack + 10, 'Account balance before unstack is not smaller than account balance after unstack on amount sum.'236 assert total_stack == total_stack_balance_before_unstack + 10, 'Total stack balance before unstack is not smaller than total stack balance after unstack on amount sum.'237 httpserver.expect_request(f"/wallets", query_string=f"traderId={MainTrader.objects.get(user=user_2).trader_id}&currencyCode=DGTX&kind=DEPOSIT").respond_with_json([{"id": 101, "traderId": MainTrader.objects.get(user=user_2).trader_id, "currency": {"id": 1, "currencyPairId": 0, "name": "Digitex", "code": "DGTX", "scale": 4, "sellPrice": "1.00000000000000000000", "buyPrice": "1.00000000000000000000"}, "balance": "10000.00000000000000000000", "blockchainAddress": "0x0A190CE3Ba33e0D2c999f35E58AF1202b064fe2F"}])238 httpserver.expect_request(f"/wallets/101/transfer/to-main", query_string=f"wallet2=1&amount=15").respond_with_json([])239 account_balance_before_unstack = Account.objects.get(program=program, type=Account.Type.DEPOSIT).balance240 total_stack_balance_before_unstack = Wallet.objects.get(program=program, user=user_2, type=Wallet.Type.DEPOSIT).balance241 transaction = Transaction.objects.create(user=user_2, currency_id=1, amount=15, wallet=deposit_wallet, account=deposit_account, type=Transaction.Type.STAKE, program=program)242 assert transaction.status == Transaction.Status.SUCCESS243 account = Account.objects.get(program=program, type=Account.Type.DEPOSIT)244 total_stack = Wallet.objects.get(program=program, user=user_2, type=Wallet.Type.DEPOSIT).balance245 assert account.balance == account_balance_before_unstack + 15, 'Account balance before unstack is not smaller than account balance after unstack on amount sum.'246 assert total_stack == total_stack_balance_before_unstack + 15, 'Total stack balance before unstack is not smaller than total stack balance after unstack on amount sum.'247# Approve248def test_create_reward_all_time(client, verified_user, short_program_with_reward, httpserver):249 deposit_wallet, _ = Wallet.objects.get_or_create(user=User.objects.get(email=verified_user), type=Wallet.Type.DEPOSIT, program=short_program_with_reward)250 settings.TREASURY_URL = httpserver.url_for("")[:-1]251 httpserver.expect_request(f"/wallets", query_string=f"traderId={MainTrader.objects.get(user__email=verified_user).trader_id}&currencyCode=DGTX&kind=DEPOSIT").respond_with_json([{"id": 100, "traderId": MainTrader.objects.get(user__email=verified_user).trader_id, "currency": {"id": 1, "currencyPairId": 0, "name": "Digitex", "code": "DGTX", "scale": 4, "sellPrice": "1.00000000000000000000", "buyPrice": "1.00000000000000000000"}, "balance": "10000.00000000000000000000", "blockchainAddress": "0x0A190CE3Ba33e0D2c999f35E58AF1202b064fe2F"}])252 httpserver.expect_request("/wallets/100/transfer/to-main", query_string=f"wallet2={short_program_with_reward.deposit_account.treasury_id}&amount=10").respond_with_json([])253 tx = Transaction.objects.create(user=User.objects.get(email=verified_user), currency_id=1, amount=10, wallet=deposit_wallet, account=short_program_with_reward.deposit_account, type=Transaction.Type.STAKE, program=short_program_with_reward)254 Transaction.objects.filter(id=tx.id).update(created_at=short_program_with_reward.begin_date - timedelta(minutes=30))255 assert tx.status == Transaction.Status.SUCCESS256 tasks.calculating_rewards()257 real_reward = Reward.objects.get(program=short_program_with_reward, user__email=verified_user)258 real_reward_amount = real_reward.amount259 # Staked half of time260 # (10 / 10) * (1800 / 3600) * 100261 to_reward = 720262 assert real_reward_amount == to_reward, 'Reward account balance is not equal to reward computing amount'263 assert Wallet.objects.get(user__email=verified_user, type=Wallet.Type.REWARD).balance == 720264# Approve265def test_create_reward_half_time(client, verified_user, short_program_with_reward, httpserver):266 deposit_wallet, _ = Wallet.objects.get_or_create(user=User.objects.get(email=verified_user),267 type=Wallet.Type.DEPOSIT, program=short_program_with_reward)268 settings.TREASURY_URL = httpserver.url_for("")[:-1]269 httpserver.expect_request(f"/wallets",270 query_string=f"traderId={MainTrader.objects.get(user__email=verified_user).trader_id}&currencyCode=DGTX&kind=DEPOSIT").respond_with_json(271 [{"id": 100, "traderId": MainTrader.objects.get(user__email=verified_user).trader_id,272 "currency": {"id": 1, "currencyPairId": 0, "name": "Digitex", "code": "DGTX", "scale": 4,273 "sellPrice": "1.00000000000000000000", "buyPrice": "1.00000000000000000000"},274 "balance": "10000.00000000000000000000", "blockchainAddress": "0x0A190CE3Ba33e0D2c999f35E58AF1202b064fe2F"}])275 httpserver.expect_request("/wallets/100/transfer/to-main",276 query_string=f"wallet2={short_program_with_reward.deposit_account.treasury_id}&amount=10").respond_with_json(277 [])278 tx = Transaction.objects.create(user=User.objects.get(email=verified_user), currency_id=1, amount=10,279 wallet=deposit_wallet, account=short_program_with_reward.deposit_account,280 type=Transaction.Type.STAKE, program=short_program_with_reward)281 Transaction.objects.filter(id=tx.id).update(created_at=short_program_with_reward.begin_date + timedelta(minutes=30))282 assert tx.status == Transaction.Status.SUCCESS283 tasks.calculating_rewards()284 real_reward = Reward.objects.get(program=short_program_with_reward, user__email=verified_user)285 real_reward_amount = real_reward.amount286 # Staked half of time287 # (10 / 10) * (1800 / 3600) * 100288 to_reward = 360289 assert real_reward_amount == to_reward, 'Reward account balance is not equal to reward computing amount'290 assert Wallet.objects.get(user__email=verified_user, type=Wallet.Type.REWARD).balance == to_reward291# Approve292def test_create_reward_with_three_tx(client, verified_user, short_program_with_reward, httpserver):293 settings.TREASURY_URL = httpserver.url_for("")[:-1]294 deposit_wallet, _ = Wallet.objects.get_or_create(user=User.objects.get(email=verified_user),295 type=Wallet.Type.DEPOSIT, program=short_program_with_reward)296 httpserver.expect_request(f"/wallets",297 query_string=f"traderId={MainTrader.objects.get(user__email=verified_user).trader_id}&currencyCode=DGTX&kind=DEPOSIT").respond_with_json(298 [{"id": 100, "traderId": MainTrader.objects.get(user__email=verified_user).trader_id,299 "currency": {"id": 1, "currencyPairId": 0, "name": "Digitex", "code": "DGTX", "scale": 4,300 "sellPrice": "1.00000000000000000000", "buyPrice": "1.00000000000000000000"},301 "balance": "10000.00000000000000000000", "blockchainAddress": "0x0A190CE3Ba33e0D2c999f35E58AF1202b064fe2F"}])302 httpserver.expect_request("/wallets/100/transfer/to-main",303 query_string=f"wallet2={short_program_with_reward.deposit_account.treasury_id}&amount=10").respond_with_json(304 [])305 tx = Transaction.objects.create(user=User.objects.get(email=verified_user), currency_id=1, amount=10,306 wallet=deposit_wallet, account=short_program_with_reward.deposit_account,307 type=Transaction.Type.STAKE, program=short_program_with_reward)308 Transaction.objects.filter(id=tx.id).update(created_at=short_program_with_reward.begin_date - timedelta(minutes=30))309 assert tx.status == Transaction.Status.SUCCESS310 tx = Transaction.objects.create(user=User.objects.get(email=verified_user), currency_id=1, amount=10,311 wallet=deposit_wallet, account=short_program_with_reward.deposit_account,312 type=Transaction.Type.STAKE, program=short_program_with_reward)313 Transaction.objects.filter(id=tx.id).update(created_at=short_program_with_reward.begin_date + timedelta(minutes=30))314 assert tx.status == Transaction.Status.SUCCESS315 tx = Transaction.objects.create(user=User.objects.get(email=verified_user), currency_id=1, amount=10,316 wallet=deposit_wallet, account=short_program_with_reward.deposit_account,317 type=Transaction.Type.STAKE, program=short_program_with_reward)318 Transaction.objects.filter(id=tx.id).update(created_at=short_program_with_reward.begin_date + timedelta(minutes=45))319 assert tx.status == Transaction.Status.SUCCESS320 tasks.calculating_rewards()321 assert Reward.objects.filter(program=short_program_with_reward, user__email=verified_user).count() == 3322 real_reward_amount = Reward.objects.filter(program=short_program_with_reward, user__email=verified_user).aggregate(amount=Sum('amount'))['amount']323 # Staked half of time324 # (10 / 10) * (1800 / 3600) * 100325 to_reward = 720326 assert real_reward_amount == to_reward, 'Reward account balance is not equal to reward computing amount'327 assert Wallet.objects.get(user__email=verified_user, type=Wallet.Type.REWARD).balance == to_reward328# Approve329def test_create_reward_with_three_tx_2(client, verified_user, short_program_with_reward_30min, httpserver):330 settings.TREASURY_URL = httpserver.url_for("")[:-1]331 deposit_wallet, _ = Wallet.objects.get_or_create(user=User.objects.get(email=verified_user),332 type=Wallet.Type.DEPOSIT, program=short_program_with_reward_30min)333 httpserver.expect_request(f"/wallets",334 query_string=f"traderId={MainTrader.objects.get(user__email=verified_user).trader_id}&currencyCode=DGTX&kind=DEPOSIT").respond_with_json(335 [{"id": 100, "traderId": MainTrader.objects.get(user__email=verified_user).trader_id,336 "currency": {"id": 1, "currencyPairId": 0, "name": "Digitex", "code": "DGTX", "scale": 4,337 "sellPrice": "1.00000000000000000000", "buyPrice": "1.00000000000000000000"},338 "balance": "10000.00000000000000000000", "blockchainAddress": "0x0A190CE3Ba33e0D2c999f35E58AF1202b064fe2F"}])339 httpserver.expect_request(f"/wallets/100/transfer/to-main",340 query_string=f"wallet2={short_program_with_reward_30min.deposit_account.treasury_id}&amount=10").respond_with_json(341 [])342 httpserver.expect_request(f"/wallets/{short_program_with_reward_30min.reward_account.treasury_id}/transfer/to-main", query_string=f"wallet2=100&amount=360").respond_with_json(343 [])344 tx = Transaction.objects.create(user=User.objects.get(email=verified_user), currency_id=1, amount=10,345 wallet=deposit_wallet, account=short_program_with_reward_30min.deposit_account,346 type=Transaction.Type.STAKE, program=short_program_with_reward_30min)347 Transaction.objects.filter(id=tx.id).update(created_at=short_program_with_reward_30min.begin_date - timedelta(minutes=30))348 assert tx.status == Transaction.Status.SUCCESS349 tx = Transaction.objects.create(user=User.objects.get(email=verified_user), currency_id=1, amount=10,350 wallet=deposit_wallet, account=short_program_with_reward_30min.deposit_account,351 type=Transaction.Type.STAKE, program=short_program_with_reward_30min)352 Transaction.objects.filter(id=tx.id).update(created_at=short_program_with_reward_30min.begin_date + timedelta(minutes=30))353 assert tx.status == Transaction.Status.SUCCESS354 tx = Transaction.objects.create(user=User.objects.get(email=verified_user), currency_id=1, amount=10,355 wallet=deposit_wallet, account=short_program_with_reward_30min.deposit_account,356 type=Transaction.Type.STAKE, program=short_program_with_reward_30min)357 Transaction.objects.filter(id=tx.id).update(created_at=short_program_with_reward_30min.begin_date + timedelta(minutes=45))358 assert tx.status == Transaction.Status.SUCCESS359 tasks.calculating_rewards()360 tasks.calculating_rewards()361 assert Reward.objects.filter(program=short_program_with_reward_30min, user__email=verified_user).count() == 3362 real_reward_amount = Reward.objects.filter(program=short_program_with_reward_30min, user__email=verified_user).aggregate(amount=Sum('amount'))['amount']363 # Staked half of time364 # (10 / 10) * (1800 / 3600) * 100365 to_reward = 720366 assert real_reward_amount == to_reward, 'Reward account balance is not equal to reward computing amount'367 assert Wallet.objects.get(user__email=verified_user, type=Wallet.Type.REWARD).balance == to_reward368# Approve369def test_create_reward_with_three_tx_2_withclaim(client, verified_user, short_program_with_reward_30min, httpserver):370 settings.TREASURY_URL = httpserver.url_for("")[:-1]371 deposit_wallet, _ = Wallet.objects.get_or_create(user=User.objects.get(email=verified_user),372 type=Wallet.Type.DEPOSIT, program=short_program_with_reward_30min)373 httpserver.expect_request(f"/wallets",374 query_string=f"traderId={MainTrader.objects.get(user__email=verified_user).trader_id}&currencyCode=DGTX&kind=DEPOSIT").respond_with_json(375 [{"id": 100, "traderId": MainTrader.objects.get(user__email=verified_user).trader_id,376 "currency": {"id": 1, "currencyPairId": 0, "name": "Digitex", "code": "DGTX", "scale": 4,377 "sellPrice": "1.00000000000000000000", "buyPrice": "1.00000000000000000000"},378 "balance": "10000.00000000000000000000", "blockchainAddress": "0x0A190CE3Ba33e0D2c999f35E58AF1202b064fe2F"}])379 httpserver.expect_request(f"/wallets/100/transfer/to-main",380 query_string=f"wallet2={short_program_with_reward_30min.deposit_account.treasury_id}&amount=10").respond_with_json(381 [])382 httpserver.expect_request(f"/wallets/{short_program_with_reward_30min.reward_account.treasury_id}/transfer/to-main", query_string=f"wallet2=100&amount=360").respond_with_json(383 [])384 tx = Transaction.objects.create(user=User.objects.get(email=verified_user), currency_id=1, amount=10,385 wallet=deposit_wallet, account=short_program_with_reward_30min.deposit_account,386 type=Transaction.Type.STAKE, program=short_program_with_reward_30min)387 Transaction.objects.filter(id=tx.id).update(created_at=short_program_with_reward_30min.begin_date - timedelta(minutes=30))388 assert tx.status == Transaction.Status.SUCCESS389 tx = Transaction.objects.create(user=User.objects.get(email=verified_user), currency_id=1, amount=10,390 wallet=deposit_wallet, account=short_program_with_reward_30min.deposit_account,391 type=Transaction.Type.STAKE, program=short_program_with_reward_30min)392 Transaction.objects.filter(id=tx.id).update(created_at=short_program_with_reward_30min.begin_date + timedelta(minutes=30))393 assert tx.status == Transaction.Status.SUCCESS394 tx = Transaction.objects.create(user=User.objects.get(email=verified_user), currency_id=1, amount=10,395 wallet=deposit_wallet, account=short_program_with_reward_30min.deposit_account,396 type=Transaction.Type.STAKE, program=short_program_with_reward_30min)...

Full Screen

Full Screen

test_duo_openvpn.py

Source:test_duo_openvpn.py Github

copy

Full Screen

...96 return environ97 def compare_params(self, recv_params, sent_params):98 stanzas = sent_params.split('&')99 return len(recv_params.split('&')) == len(stanzas) and all([s in recv_params for s in stanzas])100 def expect_request(self, method, path, params, params_func=None, response=None, raises=None):101 if params_func == None:102 params_func = lambda p: self.compare_params(p, self.EXPECTED_PREAUTH_PARAMS)103 self.expected_calls.request(method, path, mox.Func(params_func), {104 'User-Agent': self.EXPECTED_USER_AGENT,105 'Host': self.HOST,106 'Content-type': 'application/x-www-form-urlencoded',107 'Authorization': mox.Func((lambda s: s.startswith('Basic ') and not s.startswith('Basic b\''))),108 'Date': mox.Func((lambda s: bool(email.utils.parsedate_tz(s))))109 },110 )111 meth = self.expected_calls.getresponse()112 if raises is not None:113 meth.AndRaise(raises)114 else:115 meth.AndReturn(response)116 self.expected_calls.close()117 def expect_preauth(self, result, path=EXPECTED_PREAUTH_PATH, factor='push1'):118 self.expect_request(119 method='POST',120 path=path,121 params=self.EXPECTED_PREAUTH_PARAMS,122 response=MockResponse(123 status=200,124 body=json.dumps({125 'stat': 'OK',126 'response': {127 'result': result,128 'status': 'expected status',129 'factors': {'default': factor},130 },131 }),132 ),133 )134 def expect_auth(self, result, path=EXPECTED_AUTH_PATH):135 self.expect_request(136 method='POST',137 path=path,138 params=self.EXPECTED_AUTH_PARAMS,139 params_func = lambda p: self.compare_params(p, self.EXPECTED_AUTH_PARAMS),140 response=MockResponse(141 status=200,142 body=json.dumps({143 'stat': 'OK',144 'response': {145 'result': result,146 'status': 'expected status',147 },148 }),149 ),150 )151 def test_preauth_allow(self):152 environ = self.normal_environ()153 self.expect_preauth('allow')154 self.assert_auth(155 environ=environ,156 expected_control='1',157 )158 def test_preauth_deny(self):159 environ = self.normal_environ()160 self.expect_preauth('deny')161 self.assert_auth(162 environ=environ,163 expected_control='0',164 )165 def test_preauth_enroll(self):166 environ = self.normal_environ()167 self.expect_preauth('enroll')168 self.assert_auth(169 environ=environ,170 expected_control='0',171 )172 def test_preauth_bogus(self):173 environ = self.normal_environ()174 self.expect_preauth('bogus')175 self.assert_auth(176 environ=environ,177 expected_control='0',178 )179 def test_preauth_missing_result(self):180 environ = self.normal_environ()181 self.expect_request(182 method='POST',183 path=self.EXPECTED_PREAUTH_PATH,184 params=self.EXPECTED_PREAUTH_PARAMS,185 response=MockResponse(186 status=200,187 body=json.dumps({188 'stat': 'OK',189 'response': {190 'status': 'expected status',191 },192 }),193 ),194 )195 self.assert_auth(196 environ=environ,197 expected_control='0',198 )199 def test_preauth_missing_status(self):200 environ = self.normal_environ()201 self.expect_request(202 method='POST',203 path=self.EXPECTED_PREAUTH_PATH,204 params=self.EXPECTED_PREAUTH_PARAMS,205 response=MockResponse(206 status=200,207 body=json.dumps({208 'stat': 'OK',209 'response': {210 'result': 'deny',211 },212 }),213 ),214 )215 self.assert_auth(216 environ=environ,217 expected_control='0',218 )219 def test_preauth_exception(self):220 environ = self.normal_environ()221 self.expect_request(222 method='POST',223 path=self.EXPECTED_PREAUTH_PATH,224 params=self.EXPECTED_PREAUTH_PARAMS,225 raises=Exception('whoops'),226 )227 self.assert_auth(228 environ=environ,229 expected_control='0',230 )231 def test_auth_allow(self):232 environ = self.normal_environ()233 self.expect_preauth('auth')234 self.expect_auth('allow')235 self.assert_auth(236 environ=environ,237 expected_control='1',238 )239 def test_auth_deny(self):240 environ = self.normal_environ()241 self.expect_preauth('auth')242 self.expect_auth('deny')243 self.assert_auth(244 environ=environ,245 expected_control='0',246 )247 def test_auth_bogus(self):248 environ = self.normal_environ()249 self.expect_preauth('auth')250 self.expect_auth('bogus')251 self.assert_auth(252 environ=environ,253 expected_control='0',254 )255 def test_auth_missing_reason(self):256 environ = self.normal_environ()257 self.expect_preauth('auth')258 self.expect_request(259 method='POST',260 path=self.EXPECTED_AUTH_PATH,261 params=self.EXPECTED_AUTH_PARAMS,262 params_func = lambda p: self.compare_params(p, self.EXPECTED_AUTH_PARAMS),263 response=MockResponse(264 status=200,265 body=json.dumps({266 'stat': 'OK',267 'response': {268 'status': 'expected status',269 },270 }),271 ),272 )273 self.assert_auth(274 environ=environ,275 expected_control='0',276 )277 def test_auth_missing_status(self):278 environ = self.normal_environ()279 self.expect_preauth('auth')280 self.expect_request(281 method='POST',282 path=self.EXPECTED_AUTH_PATH,283 params=self.EXPECTED_AUTH_PARAMS,284 params_func = lambda p: self.compare_params(p, self.EXPECTED_AUTH_PARAMS),285 response=MockResponse(286 status=200,287 body=json.dumps({288 'stat': 'OK',289 'response': {290 'result': 'allow',291 },292 }),293 ),294 )295 self.assert_auth(296 environ=environ,297 expected_control='0',298 )299 def test_auth_exception(self):300 environ = self.normal_environ()301 self.expect_preauth('auth')302 self.expect_request(303 method='POST',304 path=self.EXPECTED_AUTH_PATH,305 params=self.EXPECTED_AUTH_PARAMS,306 params_func = lambda p: self.compare_params(p, self.EXPECTED_AUTH_PARAMS),307 raises=Exception('whoops'),308 )309 self.assert_auth(310 environ=environ,311 expected_control='0',312 )313 def test_auth_no_ipaddr(self):314 preauth_noip_params='ipaddr=0.0.0.0' \315 '&user=expected+username'316 environ = self.normal_environ()317 environ.pop('ipaddr')318 self.expect_request(319 method='POST',320 path=self.EXPECTED_PREAUTH_PATH,321 params=preauth_noip_params,322 params_func = lambda p: self.compare_params(p, preauth_noip_params),323 response=MockResponse(324 status=200,325 body=json.dumps({326 'stat': 'OK',327 'response': {328 'result': 'auth',329 'status': 'expected status',330 'factors': {'default': 'push1'},331 },332 }),333 ),334 )335 auth_noip_params='auto=expected+passcode' \336 '&ipaddr=0.0.0.0' \337 '&user=expected+username' \338 '&factor=auto'339 self.expect_request(340 method='POST',341 path=self.EXPECTED_AUTH_PATH,342 params=auth_noip_params,343 params_func = lambda p: self.compare_params(p, auth_noip_params),344 response=MockResponse(345 status=200,346 body=json.dumps({347 'stat': 'OK',348 'response': {349 'result': 'allow',350 'status': 'expected status',351 },352 }),353 ),...

Full Screen

Full Screen

test_clients.py

Source:test_clients.py Github

copy

Full Screen

...26 with mock.patch(27 "mergify_engine.config.GITHUB_API_URL",28 httpserver.url_for("/")[:-1],29 ):30 httpserver.expect_request("/users/owner/installation").respond_with_json(31 {32 "id": 12345,33 "target_type": "User",34 "permissions": {35 "checks": "write",36 "contents": "write",37 "pull_requests": "write",38 },39 "account": {"login": "testing", "id": 12345},40 }41 )42 httpserver.expect_request(43 "/app/installations/12345/access_tokens"44 ).respond_with_json(45 {"token": "<app_token>", "expires_at": "2100-12-31T23:59:59Z"}46 )47 httpserver.expect_request(48 "/", headers={"Authorization": "token <installation-token>"}49 ).respond_with_json({"work": True}, status=200)50 with github.GithubInstallationClient(github.get_auth("owner")) as client:51 ret = client.get(httpserver.url_for("/"))52 assert ret.json()["work"]53 assert len(httpserver.log) == 354 httpserver.check_assertions()55@mock.patch.object(github.CachedToken, "STORAGE", {})56def test_client_user_token(httpserver):57 with mock.patch(58 "mergify_engine.config.GITHUB_API_URL",59 httpserver.url_for("/")[:-1],60 ):61 httpserver.expect_request("/users/owner").respond_with_json(62 {63 "login": "testing",64 "id": 12345,65 }66 )67 httpserver.expect_request(68 "/", headers={"Authorization": "token <user-token>"}69 ).respond_with_json({"work": True}, status=200)70 with github.GithubInstallationClient(github.get_auth("owner")) as client:71 ret = client.get(httpserver.url_for("/"), oauth_token="<user-token>")72 assert ret.json()["work"]73 assert len(httpserver.log) == 274@mock.patch.object(github.CachedToken, "STORAGE", {})75def test_client_401_raise_ratelimit(httpserver):76 owner = "owner"77 repo = "repo"78 httpserver.expect_request("/users/owner/installation").respond_with_json(79 {80 "id": 12345,81 "target_type": "User",82 "permissions": {83 "checks": "write",84 "contents": "write",85 "pull_requests": "write",86 },87 "account": {"login": "testing", "id": 12345},88 }89 )90 httpserver.expect_request(91 "/app/installations/12345/access_tokens"92 ).respond_with_json({"token": "<token>", "expires_at": "2100-12-31T23:59:59Z"})93 httpserver.expect_oneshot_request("/rate_limit").respond_with_json(94 {"resources": {"core": {"remaining": 5000, "reset": 1234567890}}}95 )96 httpserver.expect_oneshot_request("/repos/owner/repo/pull/1").respond_with_json(97 {"message": "quota !"}, status=40398 )99 httpserver.expect_oneshot_request("/rate_limit").respond_with_json(100 {"resources": {"core": {"remaining": 0, "reset": 1234567890}}}101 )102 with mock.patch(103 "mergify_engine.config.GITHUB_API_URL",104 httpserver.url_for("/")[:-1],105 ):106 client = github.get_client(owner)107 with pytest.raises(exceptions.RateLimited):108 client.item(f"/repos/{owner}/{repo}/pull/1")109 httpserver.check_assertions()110def test_client_HTTP_400(httpserver):111 httpserver.expect_oneshot_request("/").respond_with_json(112 {"message": "This is an 4XX error"}, status=400113 )114 with http.Client() as client:115 with pytest.raises(http.HTTPClientSideError) as exc_info:116 client.get(httpserver.url_for("/"))117 assert exc_info.value.message == "This is an 4XX error"118 assert exc_info.value.status_code == 400119 assert exc_info.value.response.status_code == 400120 assert str(exc_info.value.request.url) == httpserver.url_for("/")121 httpserver.check_assertions()122def test_client_HTTP_500(httpserver):123 httpserver.expect_request("/").respond_with_data("This is an 5XX error", status=500)124 with http.Client() as client:125 with pytest.raises(http.HTTPServerSideError) as exc_info:126 client.get(httpserver.url_for("/"))127 # 5 retries128 assert len(httpserver.log) == 5129 assert exc_info.value.message == "This is an 5XX error"130 assert exc_info.value.status_code == 500131 assert exc_info.value.response.status_code == 500132 assert str(exc_info.value.request.url) == httpserver.url_for("/")133 httpserver.check_assertions()134def test_client_temporary_HTTP_500(httpserver):135 httpserver.expect_oneshot_request("/").respond_with_data(136 "This is an 5XX error", status=500137 )138 httpserver.expect_oneshot_request("/").respond_with_data(139 "This is an 5XX error", status=500140 )141 httpserver.expect_oneshot_request("/").respond_with_data(142 "This is an 5XX error", status=500143 )144 httpserver.expect_request("/").respond_with_data("It works now !", status=200)145 with http.Client() as client:146 client.get(httpserver.url_for("/"))147 # 4 retries148 assert len(httpserver.log) == 4149 httpserver.check_assertions()150def test_client_connection_error():151 with http.Client() as client:152 with pytest.raises(http.RequestError):153 client.get("http://localhost:12345")154def _do_test_client_retry_429(httpserver, retry_after, expected_seconds):155 records = []156 def record_date(_):157 records.append(datetime.datetime.utcnow())158 return Response("It works now !", 200)159 httpserver.expect_oneshot_request("/").respond_with_data(160 "This is an 429 error", status=429, headers={"Retry-After": retry_after}161 )162 httpserver.expect_request("/").respond_with_handler(record_date)163 with http.Client() as client:164 now = datetime.datetime.utcnow()165 client.get(httpserver.url_for("/"))166 assert len(httpserver.log) == 2167 elapsed_seconds = (records[0] - now).total_seconds()168 assert expected_seconds - 1 < elapsed_seconds <= expected_seconds + 1169 httpserver.check_assertions()170def test_client_retry_429_retry_after_as_seconds(httpserver):171 _do_test_client_retry_429(httpserver, 3, 3)172def test_client_retry_429_retry_after_as_absolute_date(httpserver):173 retry_after = http_date(datetime.datetime.utcnow() + datetime.timedelta(seconds=3))174 _do_test_client_retry_429(httpserver, retry_after, 3)175@mock.patch.object(github.CachedToken, "STORAGE", {})176def test_client_access_token_HTTP_500(httpserver):177 httpserver.expect_request("/users/owner/installation").respond_with_json(178 {179 "id": 12345,180 "target_type": "User",181 "permissions": {182 "checks": "write",183 "contents": "write",184 "pull_requests": "write",185 },186 "account": {"login": "testing", "id": 12345},187 }188 )189 httpserver.expect_request(190 "/app/installations/12345/access_tokens"191 ).respond_with_data("This is an 5XX error", status=500)192 with mock.patch(193 "mergify_engine.config.GITHUB_API_URL",194 httpserver.url_for("/")[:-1],195 ):196 with github.GithubInstallationClient(github.get_auth("owner")) as client:197 with pytest.raises(http.HTTPServerSideError) as exc_info:198 client.get(httpserver.url_for("/"))199 # installation request + 5 retries200 assert len(httpserver.log) == 6201 assert exc_info.value.message == "This is an 5XX error"202 assert exc_info.value.status_code == 500203 assert exc_info.value.response.status_code == 500204 assert str(exc_info.value.request.url) == httpserver.url_for(205 "/app/installations/12345/access_tokens"206 )207 httpserver.check_assertions()208@mock.patch.object(github.CachedToken, "STORAGE", {})209def test_client_installation_HTTP_500(httpserver):210 httpserver.expect_request("/users/owner/installation").respond_with_data(211 "This is an 5XX error", status=500212 )213 with mock.patch(214 "mergify_engine.config.GITHUB_API_URL",215 httpserver.url_for("/")[:-1],216 ):217 with github.GithubInstallationClient(github.get_auth("owner")) as client:218 with pytest.raises(http.HTTPServerSideError) as exc_info:219 client.get(httpserver.url_for("/"))220 # 5 retries221 assert len(httpserver.log) == 5222 assert exc_info.value.message == "This is an 5XX error"223 assert exc_info.value.status_code == 500224 assert exc_info.value.response.status_code == 500225 assert str(exc_info.value.request.url) == httpserver.url_for(226 "/users/owner/installation"227 )228 httpserver.check_assertions()229@mock.patch.object(github.CachedToken, "STORAGE", {})230def test_client_installation_HTTP_404(httpserver):231 httpserver.expect_request("/users/owner/installation").respond_with_json(232 {"message": "Repository not found"}, status=404233 )234 with mock.patch(235 "mergify_engine.config.GITHUB_API_URL",236 httpserver.url_for("/")[:-1],237 ):238 with github.GithubInstallationClient(github.get_auth("owner")) as client:239 with pytest.raises(exceptions.MergifyNotInstalled):240 client.get(httpserver.url_for("/"))241 assert len(httpserver.log) == 1242 httpserver.check_assertions()243@mock.patch.object(github.CachedToken, "STORAGE", {})244def test_client_installation_HTTP_301(httpserver):245 httpserver.expect_request("/users/owner/installation").respond_with_data(246 status=301,247 headers={"Location": httpserver.url_for("/repositories/12345/installation")},248 )249 httpserver.expect_request("/repositories/12345/installation").respond_with_json(250 {"message": "Repository not found"}, status=404251 )252 with mock.patch(253 "mergify_engine.config.GITHUB_API_URL",254 httpserver.url_for("/")[:-1],255 ):256 with github.GithubInstallationClient(github.get_auth("owner")) as client:257 with pytest.raises(exceptions.MergifyNotInstalled):258 client.get(httpserver.url_for("/"))259 assert len(httpserver.log) == 2...

Full Screen

Full Screen

test_pci_request.py

Source:test_pci_request.py Github

copy

Full Screen

1# vim: tabstop=4 shiftwidth=4 softtabstop=42# Copyright 2013 Intel Corporation3# All Rights Reserved.4#5# Licensed under the Apache License, Version 2.0 (the "License"); you may6# not use this file except in compliance with the License. You may obtain7# a copy of the License at8#9# http://www.apache.org/licenses/LICENSE-2.010#11# Unless required by applicable law or agreed to in writing, software12# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT13# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the14# License for the specific language governing permissions and limitations15# under the License.16# @author: Yongli He, Intel Corporation.17"""Tests for PCI request."""18from nova import exception19from nova.openstack.common import jsonutils20from nova.pci import pci_request as pci_request21from nova import test22_fake_alias1 = """{23 "name": "QuicAssist",24 "capability_type": "pci",25 "product_id": "4443",26 "vendor_id": "8086",27 "device_type": "ACCEL"28 }"""29_fake_alias11 = """{30 "name": "QuicAssist",31 "capability_type": "pci",32 "product_id": "4444",33 "vendor_id": "8086",34 "device_type": "ACCEL"35 }"""36_fake_alias2 = """{37 "name": "xxx",38 "capability_type": "pci",39 "product_id": "1111",40 "vendor_id": "1111",41 "device_type": "N"42 }"""43_fake_alias3 = """{44 "name": "IntelNIC",45 "capability_type": "pci",46 "product_id": "1111",47 "vendor_id": "8086",48 "device_type": "NIC"49 }"""50class AliasTestCase(test.NoDBTestCase):51 def setUp(self):52 super(AliasTestCase, self).setUp()53 def test_good_alias(self):54 self.flags(pci_alias=[_fake_alias1])55 als = pci_request._get_alias_from_config()56 self.assertEqual(type(als['QuicAssist']), list)57 expect_dict = {58 "capability_type": "pci",59 "product_id": "4443",60 "vendor_id": "8086",61 "device_type": "ACCEL"62 }63 self.assertEqual(expect_dict, als['QuicAssist'][0])64 def test_multispec_alias(self):65 self.flags(pci_alias=[_fake_alias1, _fake_alias11])66 als = pci_request._get_alias_from_config()67 self.assertEqual(type(als['QuicAssist']), list)68 expect_dict1 = {69 "capability_type": "pci",70 "product_id": "4443",71 "vendor_id": "8086",72 "device_type": "ACCEL"73 }74 expect_dict2 = {75 "capability_type": "pci",76 "product_id": "4444",77 "vendor_id": "8086",78 "device_type": "ACCEL"79 }80 self.assertEqual(expect_dict1, als['QuicAssist'][0])81 self.assertEqual(expect_dict2, als['QuicAssist'][1])82 def test_wrong_type_aliase(self):83 self.flags(pci_alias=[_fake_alias2])84 self.assertRaises(exception.PciInvalidAlias,85 pci_request._get_alias_from_config)86 def test_wrong_product_id_aliase(self):87 self.flags(pci_alias=[88 """{89 "name": "xxx",90 "capability_type": "pci",91 "product_id": "g111",92 "vendor_id": "1111",93 "device_type": "NIC"94 }"""])95 self.assertRaises(exception.PciInvalidAlias,96 pci_request._get_alias_from_config)97 def test_wrong_vendor_id_aliase(self):98 self.flags(pci_alias=[99 """{100 "name": "xxx",101 "capability_type": "pci",102 "product_id": "1111",103 "vendor_id": "0xg111",104 "device_type": "NIC"105 }"""])106 self.assertRaises(exception.PciInvalidAlias,107 pci_request._get_alias_from_config)108 def test_wrong_cap_type_aliase(self):109 self.flags(pci_alias=[110 """{111 "name": "xxx",112 "capability_type": "usb",113 "product_id": "1111",114 "vendor_id": "8086",115 "device_type": "NIC"116 }"""])117 self.assertRaises(exception.PciInvalidAlias,118 pci_request._get_alias_from_config)119 def test_dup_aliase(self):120 self.flags(pci_alias=[121 """{122 "name": "xxx",123 "capability_type": "pci",124 "product_id": "1111",125 "vendor_id": "8086",126 "device_type": "NIC"127 }""",128 """{129 "name": "xxx",130 "capability_type": "pci",131 "product_id": "1111",132 "vendor_id": "8086",133 "device_type": "ACCEL"134 }"""])135 self.assertRaises(136 exception.PciInvalidAlias,137 pci_request._get_alias_from_config)138 def test_aliase_2_request(self):139 self.flags(pci_alias=[_fake_alias1, _fake_alias3])140 expect_request = [141 {'count': 3,142 'spec': [{'vendor_id': '8086', 'product_id': '4443',143 'device_type': 'ACCEL',144 'capability_type': 'pci'}],145 'alias_name': 'QuicAssist'},146 {'count': 1,147 'spec': [{'vendor_id': '8086', 'product_id': '1111',148 'device_type': "NIC",149 'capability_type': 'pci'}],150 'alias_name': 'IntelNIC'}, ]151 requests = pci_request._translate_alias_to_requests(152 "QuicAssist : 3, IntelNIC: 1")153 self.assertEqual(set([p['count'] for p in requests]), set([1, 3]))154 exp_real = zip(expect_request, requests)155 for exp, real in exp_real:156 self.assertEqual(real, exp)157 def test_aliase_2_request_invalid(self):158 self.flags(pci_alias=[_fake_alias1, _fake_alias3])159 self.assertRaises(exception.PciRequestAliasNotDefined,160 pci_request._translate_alias_to_requests,161 "QuicAssistX : 3")162 def test_get_pci_requests_from_flavor(self):163 self.flags(pci_alias=[_fake_alias1, _fake_alias3])164 expect_request = [165 {'count': 3,166 'spec': [{'vendor_id': '8086', 'product_id': '4443',167 'device_type': "ACCEL",168 'capability_type': 'pci'}],169 'alias_name': 'QuicAssist'},170 {'count': 1,171 'spec': [{'vendor_id': '8086', 'product_id': '1111',172 'device_type': "NIC",173 'capability_type': 'pci'}],174 'alias_name': 'IntelNIC'}, ]175 flavor = {'extra_specs': {"pci_passthrough:alias":176 "QuicAssist:3, IntelNIC: 1"}}177 requests = pci_request.get_pci_requests_from_flavor(flavor)178 self.assertEqual(set([p['count'] for p in requests]), set([1, 3]))179 exp_real = zip(expect_request, requests)180 for exp, real in exp_real:181 self.assertEqual(real, exp)182 def test_get_pci_requests_from_flavor_no_extra_spec(self):183 self.flags(pci_alias=[_fake_alias1, _fake_alias3])184 flavor = {}185 requests = pci_request.get_pci_requests_from_flavor(flavor)186 self.assertEqual([], requests)187 def test_get_instance_pci_requests_no_meta(self):188 self.flags(pci_alias=[_fake_alias1, _fake_alias3])189 instance = {}190 requests = pci_request.get_instance_pci_requests(instance)191 self.assertEqual([], requests)192 def test_get_instance_pci_requests_no_request(self):193 self.flags(pci_alias=[_fake_alias1, _fake_alias3])194 instance = {'system_metadata': {'a': 'b'}}195 requests = pci_request.get_instance_pci_requests(instance)196 self.assertEqual([], requests)197 def test_get_instance_pci_requests(self):198 self.flags(pci_alias=[_fake_alias1, _fake_alias3])199 expect_request = [{200 'count': 3,201 'spec': [{'vendor_id': '8086', 'product_id': '4443',202 'device_type': "ACCEL",203 'capability_type': 'pci'}],204 'alias_name': 'QuicAssist'}]205 instance = {"system_metadata": {"pci_requests":206 jsonutils.dumps(expect_request)}}207 requests = pci_request.get_instance_pci_requests(instance)208 exp_real = zip(expect_request, requests)209 for exp, real in exp_real:210 self.assertEqual(real, exp)211 def test_get_instance_pci_requests_prefix(self):212 self.flags(pci_alias=[_fake_alias1, _fake_alias3])213 expect_request = [{214 'count': 3,215 'spec': [{'vendor_id': '8086', 'product_id': '4443',216 'device_type': "ACCEL",217 'capability_type': 'pci'}],218 'alias_name': 'QuicAssist'}]219 instance = {"system_metadata": {"new_pci_requests":220 jsonutils.dumps(expect_request)}}221 requests = pci_request.get_instance_pci_requests(instance, 'new_')222 exp_real = zip(expect_request, requests)223 for exp, real in exp_real:224 self.assertEqual(real, exp)225 def test_save_flavor_pci_info(self):226 self.flags(pci_alias=[_fake_alias1, _fake_alias3])227 expect_request = [228 {'count': 3,229 'spec': [{'vendor_id': '8086', 'product_id': '4443',230 'device_type': "ACCEL",231 'capability_type': 'pci'}],232 'alias_name': 'QuicAssist'},233 {'count': 1,234 'spec': [{'vendor_id': '8086', 'product_id': '1111',235 'device_type': "NIC",236 'capability_type': 'pci'}],237 'alias_name': 'IntelNIC'}, ]238 flavor = {'extra_specs': {"pci_passthrough:alias":239 "QuicAssist:3, IntelNIC: 1"}}240 meta = {}241 pci_request.save_flavor_pci_info(meta, flavor)242 real = jsonutils.loads(meta['pci_requests'])243 exp_real = zip(expect_request, real)244 for exp, real in exp_real:245 self.assertEqual(real, exp)246 meta = {}247 pci_request.save_flavor_pci_info(meta, flavor, "old_")248 real = jsonutils.loads(meta['old_pci_requests'])249 exp_real = zip(expect_request, real)250 for exp, real in exp_real:251 self.assertEqual(real, exp)252 def test_delete_flavor_pci_info(self):253 meta = {"pci_requests": "fake", "old_pci_requests": "fake"}254 pci_request.delete_flavor_pci_info(meta, '')255 self.assertNotIn('pci_requests', meta)256 pci_request.delete_flavor_pci_info(meta, 'old_')...

Full Screen

Full Screen

Playwright tutorial

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.

Chapters:

  1. What is Playwright : Playwright is comparatively new but has gained good popularity. Get to know some history of the Playwright with some interesting facts connected with it.
  2. How To Install Playwright : Learn in detail about what basic configuration and dependencies are required for installing Playwright and run a test. Get a step-by-step direction for installing the Playwright automation framework.
  3. Playwright Futuristic Features: Launched in 2020, Playwright gained huge popularity quickly because of some obliging features such as Playwright Test Generator and Inspector, Playwright Reporter, Playwright auto-waiting mechanism and etc. Read up on those features to master Playwright testing.
  4. What is Component Testing: Component testing in Playwright is a unique feature that allows a tester to test a single component of a web application without integrating them with other elements. Learn how to perform Component testing on the Playwright automation framework.
  5. Inputs And Buttons In Playwright: Every website has Input boxes and buttons; learn about testing inputs and buttons with different scenarios and examples.
  6. Functions and Selectors in Playwright: Learn how to launch the Chromium browser with Playwright. Also, gain a better understanding of some important functions like “BrowserContext,” which allows you to run multiple browser sessions, and “newPage” which interacts with a page.
  7. Handling Alerts and Dropdowns in Playwright : Playwright interact with different types of alerts and pop-ups, such as simple, confirmation, and prompt, and different types of dropdowns, such as single selector and multi-selector get your hands-on with handling alerts and dropdown in Playright testing.
  8. Playwright vs Puppeteer: Get to know about the difference between two testing frameworks and how they are different than one another, which browsers they support, and what features they provide.
  9. Run Playwright Tests on LambdaTest: Playwright testing with LambdaTest leverages test performance to the utmost. You can run multiple Playwright tests in Parallel with the LammbdaTest test cloud. Get a step-by-step guide to run your Playwright test on the LambdaTest platform.
  10. Playwright Python Tutorial: Playwright automation framework support all major languages such as Python, JavaScript, TypeScript, .NET and etc. However, there are various advantages to Python end-to-end testing with Playwright because of its versatile utility. Get the hang of Playwright python testing with this chapter.
  11. Playwright End To End Testing Tutorial: Get your hands on with Playwright end-to-end testing and learn to use some exciting features such as TraceViewer, Debugging, Networking, Component testing, Visual testing, and many more.
  12. Playwright Video Tutorial: Watch the video tutorials on Playwright testing from experts and get a consecutive in-depth explanation of Playwright automation testing.

Run Playwright Python automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

... View sample repo... View DocumentationSign up Free
›
_

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

...
Sign up with GoogleSign up with Email
Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful