KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use put_user_policy method in localstack
Best Python code snippet using localstack_python
test_iam.py
Source:test_iam.py 
...18@attr(operation='Verify Put User Policy')19@attr(assertion='succeeds')20@attr('user-policy')21@attr('test_of_iam')22def test_put_user_policy():23 client = get_iam_client()24 policy_document = json.dumps(25 {"Version": "2012-10-17",26 "Statement": {27 "Effect": "Allow",28 "Action": "*",29 "Resource": "*"}}30 )31 response = client.put_user_policy(PolicyDocument=policy_document, PolicyName='AllAccessPolicy',32 UserName=get_alt_user_id())33 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)34 response = client.delete_user_policy(PolicyName='AllAccessPolicy',35 UserName=get_alt_user_id())36 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)37@attr(resource='user-policy')38@attr(method='put')39@attr(operation='Verify Put User Policy with invalid user')40@attr(assertion='succeeds')41@attr('user-policy')42@attr('test_of_iam')43def test_put_user_policy_invalid_user():44 client = get_iam_client()45 policy_document = json.dumps(46 {"Version": "2012-10-17",47 "Statement": {48 "Effect": "Allow",49 "Action": "*",50 "Resource": "*"}}51 )52 e = assert_raises(ClientError, client.put_user_policy, PolicyDocument=policy_document,53 PolicyName='AllAccessPolicy', UserName="some-non-existing-user-id")54 status = _get_status(e.response)55 eq(status, 404)56@attr(resource='user-policy')57@attr(method='put')58@attr(operation='Verify Put User Policy using parameter value outside limit')59@attr(assertion='succeeds')60@attr('user-policy')61@attr('test_of_iam')62def test_put_user_policy_parameter_limit():63 client = get_iam_client()64 policy_document = json.dumps(65 {"Version": "2012-10-17",66 "Statement": [{67 "Effect": "Allow",68 "Action": "*",69 "Resource": "*"}] * 100070 }71 )72 e = assert_raises(ClientError, client.put_user_policy, PolicyDocument=policy_document,73 PolicyName='AllAccessPolicy' * 10, UserName=get_alt_user_id())74 status = _get_status(e.response)75 eq(status, 400)76@attr(resource='user-policy')77@attr(method='put')78@attr(operation='Verify Put User Policy using invalid policy document elements')79@attr(assertion='succeeds')80@attr('user-policy')81@attr('test_of_iam')82@attr('fails_on_rgw')83def test_put_user_policy_invalid_element():84 client = get_iam_client()85 # With Version other than 2012-10-1786 policy_document = json.dumps(87 {"Version": "2010-10-17",88 "Statement": [{89 "Effect": "Allow",90 "Action": "*",91 "Resource": "*"}]92 }93 )94 e = assert_raises(ClientError, client.put_user_policy, PolicyDocument=policy_document,95 PolicyName='AllAccessPolicy', UserName=get_alt_user_id())96 status = _get_status(e.response)97 eq(status, 400)98 # With no Statement99 policy_document = json.dumps(100 {101 "Version": "2012-10-17",102 }103 )104 e = assert_raises(ClientError, client.put_user_policy, PolicyDocument=policy_document,105 PolicyName='AllAccessPolicy', UserName=get_alt_user_id())106 status = _get_status(e.response)107 eq(status, 400)108 # with same Sid for 2 statements109 policy_document = json.dumps(110 {"Version": "2012-10-17",111 "Statement": [112 {"Sid": "98AB54CF",113 "Effect": "Allow",114 "Action": "*",115 "Resource": "*"},116 {"Sid": "98AB54CF",117 "Effect": "Allow",118 "Action": "*",119 "Resource": "*"}]120 }121 )122 e = assert_raises(ClientError, client.put_user_policy, PolicyDocument=policy_document,123 PolicyName='AllAccessPolicy', UserName=get_alt_user_id())124 status = _get_status(e.response)125 eq(status, 400)126 # with Principal127 policy_document = json.dumps(128 {"Version": "2012-10-17",129 "Statement": [{130 "Effect": "Allow",131 "Action": "*",132 "Resource": "*",133 "Principal": "arn:aws:iam:::username"}]134 }135 )136 e = assert_raises(ClientError, client.put_user_policy, PolicyDocument=policy_document,137 PolicyName='AllAccessPolicy', UserName=get_alt_user_id())138 status = _get_status(e.response)139 eq(status, 400)140@attr(resource='user-policy')141@attr(method='put')142@attr(operation='Verify Put a policy that already exists')143@attr(assertion='succeeds')144@attr('user-policy')145@attr('test_of_iam')146def test_put_existing_user_policy():147 client = get_iam_client()148 policy_document = json.dumps(149 {"Version": "2012-10-17",150 "Statement": {151 "Effect": "Allow",152 "Action": "*",153 "Resource": "*"}154 }155 )156 response = client.put_user_policy(PolicyDocument=policy_document, PolicyName='AllAccessPolicy',157 UserName=get_alt_user_id())158 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)159 client.put_user_policy(PolicyDocument=policy_document, PolicyName='AllAccessPolicy',160 UserName=get_alt_user_id())161 client.delete_user_policy(PolicyName='AllAccessPolicy', UserName=get_alt_user_id())162@attr(resource='user-policy')163@attr(method='put')164@attr(operation='Verify List User policies')165@attr(assertion='succeeds')166@attr('user-policy')167@attr('test_of_iam')168def test_list_user_policy():169 client = get_iam_client()170 policy_document = json.dumps(171 {"Version": "2012-10-17",172 "Statement": {173 "Effect": "Allow",174 "Action": "*",175 "Resource": "*"}176 }177 )178 response = client.put_user_policy(PolicyDocument=policy_document, PolicyName='AllAccessPolicy',179 UserName=get_alt_user_id())180 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)181 response = client.list_user_policies(UserName=get_alt_user_id())182 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)183 client.delete_user_policy(PolicyName='AllAccessPolicy', UserName=get_alt_user_id())184@attr(resource='user-policy')185@attr(method='put')186@attr(operation='Verify List User policies with invalid user')187@attr(assertion='succeeds')188@attr('user-policy')189@attr('test_of_iam')190def test_list_user_policy_invalid_user():191 client = get_iam_client()192 e = assert_raises(ClientError, client.list_user_policies, UserName="some-non-existing-user-id")193 status = _get_status(e.response)194 eq(status, 404)195@attr(resource='user-policy')196@attr(method='get')197@attr(operation='Verify Get User policy')198@attr(assertion='succeeds')199@attr('user-policy')200@attr('test_of_iam')201def test_get_user_policy():202 client = get_iam_client()203 policy_document = json.dumps(204 {"Version": "2012-10-17",205 "Statement": {206 "Effect": "Allow",207 "Action": "*",208 "Resource": "*"}}209 )210 response = client.put_user_policy(PolicyDocument=policy_document, PolicyName='AllAccessPolicy',211 UserName=get_alt_user_id())212 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)213 response = client.get_user_policy(PolicyName='AllAccessPolicy', UserName=get_alt_user_id())214 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)215 response = client.delete_user_policy(PolicyName='AllAccessPolicy',216 UserName=get_alt_user_id())217 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)218@attr(resource='user-policy')219@attr(method='get')220@attr(operation='Verify Get User Policy with invalid user')221@attr(assertion='succeeds')222@attr('user-policy')223@attr('test_of_iam')224def test_get_user_policy_invalid_user():225 client = get_iam_client()226 policy_document = json.dumps(227 {"Version": "2012-10-17",228 "Statement": {229 "Effect": "Allow",230 "Action": "*",231 "Resource": "*"}}232 )233 response = client.put_user_policy(PolicyDocument=policy_document, PolicyName='AllAccessPolicy',234 UserName=get_alt_user_id())235 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)236 e = assert_raises(ClientError, client.get_user_policy, PolicyName='AllAccessPolicy',237 UserName="some-non-existing-user-id")238 status = _get_status(e.response)239 eq(status, 404)240 client.delete_user_policy(PolicyName='AllAccessPolicy', UserName=get_alt_user_id())241@attr(resource='user-policy')242@attr(method='get')243@attr(operation='Verify Get User Policy with invalid policy name')244@attr(assertion='succeeds')245@attr('user-policy')246@attr('test_of_iam')247@attr('fails_on_rgw')248def test_get_user_policy_invalid_policy_name():249 client = get_iam_client()250 policy_document = json.dumps(251 {"Version": "2012-10-17",252 "Statement": {253 "Effect": "Allow",254 "Action": "*",255 "Resource": "*"}}256 )257 client.put_user_policy(PolicyDocument=policy_document, PolicyName='AllAccessPolicy',258 UserName=get_alt_user_id())259 e = assert_raises(ClientError, client.get_user_policy, PolicyName='non-existing-policy-name',260 UserName=get_alt_user_id())261 status = _get_status(e.response)262 eq(status, 404)263 client.delete_user_policy(PolicyName='AllAccessPolicy', UserName=get_alt_user_id())264@attr(resource='user-policy')265@attr(method='get')266@attr(operation='Verify Get Deleted User Policy')267@attr(assertion='succeeds')268@attr('user-policy')269@attr('test_of_iam')270@attr('fails_on_rgw')271def test_get_deleted_user_policy():272 client = get_iam_client()273 policy_document = json.dumps(274 {"Version": "2012-10-17",275 "Statement": {276 "Effect": "Allow",277 "Action": "*",278 "Resource": "*"}}279 )280 client.put_user_policy(PolicyDocument=policy_document, PolicyName='AllAccessPolicy',281 UserName=get_alt_user_id())282 client.delete_user_policy(PolicyName='AllAccessPolicy', UserName=get_alt_user_id())283 e = assert_raises(ClientError, client.get_user_policy, PolicyName='AllAccessPolicy',284 UserName=get_alt_user_id())285 status = _get_status(e.response)286 eq(status, 404)287@attr(resource='user-policy')288@attr(method='get')289@attr(operation='Verify Get a policy from multiple policies for a user')290@attr(assertion='succeeds')291@attr('user-policy')292@attr('test_of_iam')293def test_get_user_policy_from_multiple_policies():294 client = get_iam_client()295 policy_document_allow = json.dumps(296 {"Version": "2012-10-17",297 "Statement": {298 "Effect": "Allow",299 "Action": "*",300 "Resource": "*"}}301 )302 response = client.put_user_policy(PolicyDocument=policy_document_allow,303 PolicyName='AllowAccessPolicy1',304 UserName=get_alt_user_id())305 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)306 response = client.put_user_policy(PolicyDocument=policy_document_allow,307 PolicyName='AllowAccessPolicy2',308 UserName=get_alt_user_id())309 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)310 response = client.get_user_policy(PolicyName='AllowAccessPolicy2',311 UserName=get_alt_user_id())312 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)313 response = client.delete_user_policy(PolicyName='AllowAccessPolicy1',314 UserName=get_alt_user_id())315 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)316 response = client.delete_user_policy(PolicyName='AllowAccessPolicy2',317 UserName=get_alt_user_id())318 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)319@attr(resource='user-policy')320@attr(method='delete')321@attr(operation='Verify Delete User Policy')322@attr(assertion='succeeds')323@attr('user-policy')324@attr('test_of_iam')325def test_delete_user_policy():326 client = get_iam_client()327 policy_document_allow = json.dumps(328 {"Version": "2012-10-17",329 "Statement": {330 "Effect": "Allow",331 "Action": "*",332 "Resource": "*"}}333 )334 response = client.put_user_policy(PolicyDocument=policy_document_allow,335 PolicyName='AllowAccessPolicy',336 UserName=get_alt_user_id())337 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)338 response = client.delete_user_policy(PolicyName='AllowAccessPolicy',339 UserName=get_alt_user_id())340 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)341@attr(resource='user-policy')342@attr(method='delete')343@attr(operation='Verify Delete User Policy with invalid user')344@attr(assertion='succeeds')345@attr('user-policy')346@attr('test_of_iam')347def test_delete_user_policy_invalid_user():348 client = get_iam_client()349 policy_document_allow = json.dumps(350 {"Version": "2012-10-17",351 "Statement": {352 "Effect": "Allow",353 "Action": "*",354 "Resource": "*"}}355 )356 response = client.put_user_policy(PolicyDocument=policy_document_allow,357 PolicyName='AllowAccessPolicy',358 UserName=get_alt_user_id())359 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)360 e = assert_raises(ClientError, client.delete_user_policy, PolicyName='AllAccessPolicy',361 UserName="some-non-existing-user-id")362 status = _get_status(e.response)363 eq(status, 404)364 response = client.delete_user_policy(PolicyName='AllowAccessPolicy',365 UserName=get_alt_user_id())366 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)367@attr(resource='user-policy')368@attr(method='delete')369@attr(operation='Verify Delete User Policy with invalid policy name')370@attr(assertion='succeeds')371@attr('user-policy')372@attr('test_of_iam')373def test_delete_user_policy_invalid_policy_name():374 client = get_iam_client()375 policy_document_allow = json.dumps(376 {"Version": "2012-10-17",377 "Statement": {378 "Effect": "Allow",379 "Action": "*",380 "Resource": "*"}}381 )382 response = client.put_user_policy(PolicyDocument=policy_document_allow,383 PolicyName='AllowAccessPolicy',384 UserName=get_alt_user_id())385 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)386 e = assert_raises(ClientError, client.delete_user_policy, PolicyName='non-existing-policy-name',387 UserName=get_alt_user_id())388 status = _get_status(e.response)389 eq(status, 404)390 response = client.delete_user_policy(PolicyName='AllowAccessPolicy',391 UserName=get_alt_user_id())392 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)393@attr(resource='user-policy')394@attr(method='delete')395@attr(operation='Verify Delete multiple User policies for a user')396@attr(assertion='succeeds')397@attr('user-policy')398@attr('test_of_iam')399def test_delete_user_policy_from_multiple_policies():400 client = get_iam_client()401 policy_document_allow = json.dumps(402 {"Version": "2012-10-17",403 "Statement": {404 "Effect": "Allow",405 "Action": "*",406 "Resource": "*"}}407 )408 response = client.put_user_policy(PolicyDocument=policy_document_allow,409 PolicyName='AllowAccessPolicy1',410 UserName=get_alt_user_id())411 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)412 response = client.put_user_policy(PolicyDocument=policy_document_allow,413 PolicyName='AllowAccessPolicy2',414 UserName=get_alt_user_id())415 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)416 response = client.put_user_policy(PolicyDocument=policy_document_allow,417 PolicyName='AllowAccessPolicy3',418 UserName=get_alt_user_id())419 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)420 response = client.delete_user_policy(PolicyName='AllowAccessPolicy1',421 UserName=get_alt_user_id())422 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)423 response = client.delete_user_policy(PolicyName='AllowAccessPolicy2',424 UserName=get_alt_user_id())425 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)426 response = client.get_user_policy(PolicyName='AllowAccessPolicy3',427 UserName=get_alt_user_id())428 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)429 response = client.delete_user_policy(PolicyName='AllowAccessPolicy3',430 UserName=get_alt_user_id())431 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)432@attr(resource='user-policy')433@attr(method='s3 Actions')434@attr(operation='Verify Allow Bucket Actions in user Policy')435@attr(assertion='succeeds')436@attr('user-policy')437@attr('test_of_iam')438def test_allow_bucket_actions_in_user_policy():439 client = get_iam_client()440 s3_client_alt = get_alt_client()441 s3_client_iam = get_iam_s3client()442 bucket = get_new_bucket(client=s3_client_iam)443 s3_client_iam.put_object(Bucket=bucket, Key='foo', Body='bar')444 policy_document_allow = json.dumps(445 {"Version": "2012-10-17",446 "Statement": {447 "Effect": "Allow",448 "Action": ["s3:ListBucket", "s3:DeleteBucket"],449 "Resource": f"arn:aws:s3:::{bucket}"}}450 )451 response = client.put_user_policy(PolicyDocument=policy_document_allow,452 PolicyName='AllowAccessPolicy', UserName=get_alt_user_id())453 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)454 response = s3_client_alt.list_objects(Bucket=bucket)455 object_found = False456 for object_received in response['Contents']:457 if "foo" == object_received['Key']:458 object_found = True459 break460 if not object_found:461 raise AssertionError("Object is not listed")462 response = s3_client_iam.delete_object(Bucket=bucket, Key='foo')463 eq(response['ResponseMetadata']['HTTPStatusCode'], 204)464 response = s3_client_alt.delete_bucket(Bucket=bucket)465 eq(response['ResponseMetadata']['HTTPStatusCode'], 204)466 response = s3_client_iam.list_buckets()467 for bucket in response['Buckets']:468 if bucket == bucket['Name']:469 raise AssertionError("deleted bucket is getting listed")470 response = client.delete_user_policy(PolicyName='AllowAccessPolicy',471 UserName=get_alt_user_id())472 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)473@attr(resource='user-policy')474@attr(method='s3 Actions')475@attr(operation='Verify Deny Bucket Actions in user Policy')476@attr(assertion='succeeds')477@attr('user-policy')478@attr('test_of_iam')479@attr('fails_on_dbstore')480def test_deny_bucket_actions_in_user_policy():481 client = get_iam_client()482 s3_client = get_alt_client()483 bucket = get_new_bucket(client=s3_client)484 policy_document_deny = json.dumps(485 {"Version": "2012-10-17",486 "Statement": {487 "Effect": "Deny",488 "Action": ["s3:ListAllMyBuckets", "s3:DeleteBucket"],489 "Resource": "arn:aws:s3:::*"}}490 )491 response = client.put_user_policy(PolicyDocument=policy_document_deny,492 PolicyName='DenyAccessPolicy',493 UserName=get_alt_user_id())494 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)495 e = assert_raises(ClientError, s3_client.list_buckets, Bucket=bucket)496 status, error_code = _get_status_and_error_code(e.response)497 eq(status, 403)498 eq(error_code, 'AccessDenied')499 e = assert_raises(ClientError, s3_client.delete_bucket, Bucket=bucket)500 status, error_code = _get_status_and_error_code(e.response)501 eq(status, 403)502 eq(error_code, 'AccessDenied')503 response = client.delete_user_policy(PolicyName='DenyAccessPolicy',504 UserName=get_alt_user_id())505 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)506 response = s3_client.delete_bucket(Bucket=bucket)507 eq(response['ResponseMetadata']['HTTPStatusCode'], 204)508@attr(resource='user-policy')509@attr(method='s3 Actions')510@attr(operation='Verify Allow Object Actions in user Policy')511@attr(assertion='succeeds')512@attr('user-policy')513@attr('test_of_iam')514def test_allow_object_actions_in_user_policy():515 client = get_iam_client()516 s3_client_alt = get_alt_client()517 s3_client_iam = get_iam_s3client()518 bucket = get_new_bucket(client=s3_client_iam)519 policy_document_allow = json.dumps(520 {"Version": "2012-10-17",521 "Statement": {522 "Effect": "Allow",523 "Action": ["s3:PutObject", "s3:GetObject", "s3:DeleteObject"],524 "Resource": f"arn:aws:s3:::{bucket}/*"}}525 )526 response = client.put_user_policy(PolicyDocument=policy_document_allow,527 PolicyName='AllowAccessPolicy', UserName=get_alt_user_id())528 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)529 s3_client_alt.put_object(Bucket=bucket, Key='foo', Body='bar')530 response = s3_client_alt.get_object(Bucket=bucket, Key='foo')531 body = response['Body'].read()532 if type(body) is bytes:533 body = body.decode()534 eq(body, "bar")535 response = s3_client_alt.delete_object(Bucket=bucket, Key='foo')536 eq(response['ResponseMetadata']['HTTPStatusCode'], 204)537 e = assert_raises(ClientError, s3_client_iam.get_object, Bucket=bucket, Key='foo')538 status, error_code = _get_status_and_error_code(e.response)539 eq(status, 404)540 eq(error_code, 'NoSuchKey')541 response = s3_client_iam.delete_bucket(Bucket=bucket)542 eq(response['ResponseMetadata']['HTTPStatusCode'], 204)543 response = client.delete_user_policy(PolicyName='AllowAccessPolicy',544 UserName=get_alt_user_id())545 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)546@attr(resource='user-policy')547@attr(method='s3 Actions')548@attr(operation='Verify Deny Object Actions in user Policy')549@attr(assertion='succeeds')550@attr('user-policy')551@attr('test_of_iam')552@attr('fails_on_dbstore')553def test_deny_object_actions_in_user_policy():554 client = get_iam_client()555 s3_client_alt = get_alt_client()556 bucket = get_new_bucket(client=s3_client_alt)557 s3_client_alt.put_object(Bucket=bucket, Key='foo', Body='bar')558 policy_document_deny = json.dumps(559 {"Version": "2012-10-17",560 "Statement": [{561 "Effect": "Deny",562 "Action": ["s3:PutObject", "s3:GetObject", "s3:DeleteObject"],563 "Resource": f"arn:aws:s3:::{bucket}/*"}, {564 "Effect": "Allow",565 "Action": ["s3:DeleteBucket"],566 "Resource": f"arn:aws:s3:::{bucket}"}]}567 )568 client.put_user_policy(PolicyDocument=policy_document_deny, PolicyName='DenyAccessPolicy',569 UserName=get_alt_user_id())570 e = assert_raises(ClientError, s3_client_alt.put_object, Bucket=bucket, Key='foo')571 status, error_code = _get_status_and_error_code(e.response)572 eq(status, 403)573 eq(error_code, 'AccessDenied')574 e = assert_raises(ClientError, s3_client_alt.get_object, Bucket=bucket, Key='foo')575 status, error_code = _get_status_and_error_code(e.response)576 eq(status, 403)577 eq(error_code, 'AccessDenied')578 e = assert_raises(ClientError, s3_client_alt.delete_object, Bucket=bucket, Key='foo')579 status, error_code = _get_status_and_error_code(e.response)580 eq(status, 403)581 eq(error_code, 'AccessDenied')582 response = client.delete_user_policy(PolicyName='DenyAccessPolicy',583 UserName=get_alt_user_id())584 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)585@attr(resource='user-policy')586@attr(method='s3 Actions')587@attr(operation='Verify Allow Multipart Actions in user Policy')588@attr(assertion='succeeds')589@attr('user-policy')590@attr('test_of_iam')591def test_allow_multipart_actions_in_user_policy():592 client = get_iam_client()593 s3_client_alt = get_alt_client()594 s3_client_iam = get_iam_s3client()595 bucket = get_new_bucket(client=s3_client_iam)596 policy_document_allow = json.dumps(597 {"Version": "2012-10-17",598 "Statement": {599 "Effect": "Allow",600 "Action": ["s3:ListBucketMultipartUploads", "s3:AbortMultipartUpload"],601 "Resource": "arn:aws:s3:::*"}}602 )603 response = client.put_user_policy(PolicyDocument=policy_document_allow,604 PolicyName='AllowAccessPolicy', UserName=get_alt_user_id())605 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)606 key = "mymultipart"607 mb = 1024 * 1024608 (upload_id, _, _) = _multipart_upload(client=s3_client_iam, bucket_name=bucket, key=key,609 size=5 * mb)610 response = s3_client_alt.list_multipart_uploads(Bucket=bucket)611 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)612 response = s3_client_alt.abort_multipart_upload(Bucket=bucket, Key=key, UploadId=upload_id)613 eq(response['ResponseMetadata']['HTTPStatusCode'], 204)614 response = s3_client_iam.delete_bucket(Bucket=bucket)615 eq(response['ResponseMetadata']['HTTPStatusCode'], 204)616 response = client.delete_user_policy(PolicyName='AllowAccessPolicy',617 UserName=get_alt_user_id())618 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)619@attr(resource='user-policy')620@attr(method='s3 Actions')621@attr(operation='Verify Deny Multipart Actions in user Policy')622@attr(assertion='succeeds')623@attr('user-policy')624@attr('test_of_iam')625@attr('fails_on_dbstore')626def test_deny_multipart_actions_in_user_policy():627 client = get_iam_client()628 s3_client = get_alt_client()629 bucket = get_new_bucket(client=s3_client)630 policy_document_deny = json.dumps(631 {"Version": "2012-10-17",632 "Statement": {633 "Effect": "Deny",634 "Action": ["s3:ListBucketMultipartUploads", "s3:AbortMultipartUpload"],635 "Resource": "arn:aws:s3:::*"}}636 )637 response = client.put_user_policy(PolicyDocument=policy_document_deny,638 PolicyName='DenyAccessPolicy',639 UserName=get_alt_user_id())640 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)641 key = "mymultipart"642 mb = 1024 * 1024643 (upload_id, _, _) = _multipart_upload(client=s3_client, bucket_name=bucket, key=key,644 size=5 * mb)645 e = assert_raises(ClientError, s3_client.list_multipart_uploads, Bucket=bucket)646 status, error_code = _get_status_and_error_code(e.response)647 eq(status, 403)648 eq(error_code, 'AccessDenied')649 e = assert_raises(ClientError, s3_client.abort_multipart_upload, Bucket=bucket,650 Key=key, UploadId=upload_id)651 status, error_code = _get_status_and_error_code(e.response)652 eq(status, 403)653 eq(error_code, 'AccessDenied')654 response = s3_client.delete_bucket(Bucket=bucket)655 eq(response['ResponseMetadata']['HTTPStatusCode'], 204)656 response = client.delete_user_policy(PolicyName='DenyAccessPolicy',657 UserName=get_alt_user_id())658 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)659@attr(resource='user-policy')660@attr(method='s3 Actions')661@attr(operation='Verify Allow Tagging Actions in user Policy')662@attr(assertion='succeeds')663@attr('user-policy')664@attr('test_of_iam')665@attr('fails_on_dbstore')666def test_allow_tagging_actions_in_user_policy():667 client = get_iam_client()668 s3_client_alt = get_alt_client()669 s3_client_iam = get_iam_s3client()670 bucket = get_new_bucket(client=s3_client_iam)671 policy_document_allow = json.dumps(672 {"Version": "2012-10-17",673 "Statement": {674 "Effect": "Allow",675 "Action": ["s3:PutBucketTagging", "s3:GetBucketTagging",676 "s3:PutObjectTagging", "s3:GetObjectTagging"],677 "Resource": f"arn:aws:s3:::*"}}678 )679 client.put_user_policy(PolicyDocument=policy_document_allow, PolicyName='AllowAccessPolicy',680 UserName=get_alt_user_id())681 tags = {'TagSet': [{'Key': 'Hello', 'Value': 'World'}, ]}682 response = s3_client_alt.put_bucket_tagging(Bucket=bucket, Tagging=tags)683 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)684 response = s3_client_alt.get_bucket_tagging(Bucket=bucket)685 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)686 eq(response['TagSet'][0]['Key'], 'Hello')687 eq(response['TagSet'][0]['Value'], 'World')688 obj_key = 'obj'689 response = s3_client_iam.put_object(Bucket=bucket, Key=obj_key, Body='obj_body')690 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)691 response = s3_client_alt.put_object_tagging(Bucket=bucket, Key=obj_key, Tagging=tags)692 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)693 response = s3_client_alt.get_object_tagging(Bucket=bucket, Key=obj_key)694 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)695 eq(response['TagSet'], tags['TagSet'])696 response = s3_client_iam.delete_object(Bucket=bucket, Key=obj_key)697 eq(response['ResponseMetadata']['HTTPStatusCode'], 204)698 response = s3_client_iam.delete_bucket(Bucket=bucket)699 eq(response['ResponseMetadata']['HTTPStatusCode'], 204)700 response = client.delete_user_policy(PolicyName='AllowAccessPolicy',701 UserName=get_alt_user_id())702 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)703@attr(resource='user-policy')704@attr(method='s3 Actions')705@attr(operation='Verify Deny Tagging Actions in user Policy')706@attr(assertion='succeeds')707@attr('user-policy')708@attr('test_of_iam')709@attr('fails_on_dbstore')710def test_deny_tagging_actions_in_user_policy():711 client = get_iam_client()712 s3_client = get_alt_client()713 bucket = get_new_bucket(client=s3_client)714 policy_document_deny = json.dumps(715 {"Version": "2012-10-17",716 "Statement": {717 "Effect": "Deny",718 "Action": ["s3:PutBucketTagging", "s3:GetBucketTagging",719 "s3:PutObjectTagging", "s3:DeleteObjectTagging"],720 "Resource": "arn:aws:s3:::*"}}721 )722 client.put_user_policy(PolicyDocument=policy_document_deny, PolicyName='DenyAccessPolicy',723 UserName=get_alt_user_id())724 tags = {'TagSet': [{'Key': 'Hello', 'Value': 'World'}, ]}725 e = assert_raises(ClientError, s3_client.put_bucket_tagging, Bucket=bucket, Tagging=tags)726 status, error_code = _get_status_and_error_code(e.response)727 eq(status, 403)728 eq(error_code, 'AccessDenied')729 e = assert_raises(ClientError, s3_client.get_bucket_tagging, Bucket=bucket)730 status, error_code = _get_status_and_error_code(e.response)731 eq(status, 403)732 eq(error_code, 'AccessDenied')733 obj_key = 'obj'734 response = s3_client.put_object(Bucket=bucket, Key=obj_key, Body='obj_body')735 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)736 e = assert_raises(ClientError, s3_client.put_object_tagging, Bucket=bucket, Key=obj_key,737 Tagging=tags)738 status, error_code = _get_status_and_error_code(e.response)739 eq(status, 403)740 eq(error_code, 'AccessDenied')741 e = assert_raises(ClientError, s3_client.delete_object_tagging, Bucket=bucket, Key=obj_key)742 status, error_code = _get_status_and_error_code(e.response)743 eq(status, 403)744 eq(error_code, 'AccessDenied')745 response = s3_client.delete_object(Bucket=bucket, Key=obj_key)746 eq(response['ResponseMetadata']['HTTPStatusCode'], 204)747 response = s3_client.delete_bucket(Bucket=bucket)748 eq(response['ResponseMetadata']['HTTPStatusCode'], 204)749 response = client.delete_user_policy(PolicyName='DenyAccessPolicy',750 UserName=get_alt_user_id())751 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)752@attr(resource='user-policy')753@attr(method='put')754@attr(operation='Verify conflicting user policy statements')755@attr(assertion='succeeds')756@attr('user-policy')757@attr('test_of_iam')758@attr('fails_on_dbstore')759def test_verify_conflicting_user_policy_statements():760 s3client = get_alt_client()761 bucket = get_new_bucket(client=s3client)762 policy_document = json.dumps(763 {"Version": "2012-10-17",764 "Statement": [765 {"Sid": "98AB54CG",766 "Effect": "Allow",767 "Action": "s3:ListBucket",768 "Resource": f"arn:aws:s3:::{bucket}"},769 {"Sid": "98AB54CA",770 "Effect": "Deny",771 "Action": "s3:ListBucket",772 "Resource": f"arn:aws:s3:::{bucket}"}773 ]}774 )775 client = get_iam_client()776 response = client.put_user_policy(PolicyDocument=policy_document, PolicyName='DenyAccessPolicy',777 UserName=get_alt_user_id())778 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)779 e = assert_raises(ClientError, s3client.list_objects, Bucket=bucket)780 status, error_code = _get_status_and_error_code(e.response)781 eq(status, 403)782 eq(error_code, 'AccessDenied')783 response = client.delete_user_policy(PolicyName='DenyAccessPolicy',784 UserName=get_alt_user_id())785 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)786@attr(resource='user-policy')787@attr(method='put')788@attr(operation='Verify conflicting user policies')789@attr(assertion='succeeds')790@attr('user-policy')791@attr('test_of_iam')792@attr('fails_on_dbstore')793def test_verify_conflicting_user_policies():794 s3client = get_alt_client()795 bucket = get_new_bucket(client=s3client)796 policy_allow = json.dumps(797 {"Version": "2012-10-17",798 "Statement": {"Sid": "98AB54CG",799 "Effect": "Allow",800 "Action": "s3:ListBucket",801 "Resource": f"arn:aws:s3:::{bucket}"}}802 )803 policy_deny = json.dumps(804 {"Version": "2012-10-17",805 "Statement": {"Sid": "98AB54CGZ",806 "Effect": "Deny",807 "Action": "s3:ListBucket",808 "Resource": f"arn:aws:s3:::{bucket}"}}809 )810 client = get_iam_client()811 response = client.put_user_policy(PolicyDocument=policy_allow, PolicyName='AllowAccessPolicy',812 UserName=get_alt_user_id())813 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)814 response = client.put_user_policy(PolicyDocument=policy_deny, PolicyName='DenyAccessPolicy',815 UserName=get_alt_user_id())816 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)817 e = assert_raises(ClientError, s3client.list_objects, Bucket=bucket)818 status, error_code = _get_status_and_error_code(e.response)819 eq(status, 403)820 eq(error_code, 'AccessDenied')821 response = client.delete_user_policy(PolicyName='AllowAccessPolicy',822 UserName=get_alt_user_id())823 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)824 response = client.delete_user_policy(PolicyName='DenyAccessPolicy',825 UserName=get_alt_user_id())826 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)827@attr(resource='user-policy')828@attr(operation='Verify Allow Actions for IAM user policies')829@attr(assertion='succeeds')830@attr('user-policy')831@attr('test_of_iam')832def test_verify_allow_iam_actions():833 policy1 = json.dumps(834 {"Version": "2012-10-17",835 "Statement": {"Sid": "98AB54CGA",836 "Effect": "Allow",837 "Action": ["iam:PutUserPolicy", "iam:GetUserPolicy",838 "iam:ListUserPolicies", "iam:DeleteUserPolicy"],839 "Resource": f"arn:aws:iam:::user/{get_alt_user_id()}"}}840 )841 client1 = get_iam_client()842 iam_client_alt = get_alt_iam_client()843 response = client1.put_user_policy(PolicyDocument=policy1, PolicyName='AllowAccessPolicy',844 UserName=get_alt_user_id())845 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)846 response = iam_client_alt.get_user_policy(PolicyName='AllowAccessPolicy',847 UserName=get_alt_user_id())848 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)849 response = iam_client_alt.list_user_policies(UserName=get_alt_user_id())850 eq(response['ResponseMetadata']['HTTPStatusCode'], 200)851 response = iam_client_alt.delete_user_policy(PolicyName='AllowAccessPolicy',852 UserName=get_alt_user_id())...
awsmanager.py
Source:awsmanager.py
...18 def delete_user(self, username=None):19 return self.aws_manager.IAM.delete_user(username=username)20 def get_user(self, username=None):21 return self.aws_manager.IAM.get_user(username=username)22 def put_user_policy(self, username=None, policyname=None, policydocument=None):23 return self.aws_manager.IAM.put_user_policy(username=username, policyname=policyname,24 policydocument=policydocument)25 def get_usernames(self):26 return self.aws_manager.IAM.get_usernames()27 def create_user_key_pair(self, username=None):28 return self.aws_manager.IAM.create_user_key_pair(username=username)29 def get_s3_policy(self, bucket=None, folder=None):30 return self.aws_manager.S3.get_s3_policy(bucket=bucket, folder=folder)31 def get_sqs_policy(self, queue_name=None, full_access=None):32 return self.aws_manager.SQS.get_sqs_policy(queue_name=queue_name, full_access=full_access)33 def allow_sqs_queue_for_user(self, username=None, queue_name=None, full_access=None):34 policy = self.get_sqs_policy(queue_name=queue_name, full_access=full_access)35 self.put_user_policy(username=username, policyname='sqs_policy_%s_%s'%(username, queue_name),36 policydocument=policy)37 def create_s3_for_user(self, username=None, bucket=None, region=None):38 return self.aws_manager.S3.create_s3_for_user(username=username,39 bucket=bucket, region=region)40 def allow_s3_folder_for_user(self, username=None, bucket=None, folder=None):41 return self.aws_manager.S3.allow_s3_folder_for_user(username=username,42 bucket=bucket, folder=folder)43 def get_hosted_zones(self):44 return self.aws_manager.Route53.get_hosted_zones()45class ServiceParameters(BaseComponent):46 def service_parameters(self,pane,datapath=None,**kwargs):47 fb = pane.formbuilder(datapath=datapath)48 fb.textbox(value='^.aws_access_key_id',lbl='Aws Access Key Id')49 fb.textbox(value='^.aws_secret_access_key',lbl='Aws Secret Access Key')...
create_config_aws_event_relay_user.py
Source:create_config_aws_event_relay_user.py
...22 UserName=username23 )24except IAM.exceptions.EntityAlreadyExistsException:25 pass26IAM.put_user_policy(27 UserName=username,28 PolicyName='sns_publisher',29 PolicyDocument=json.dumps({30 'Version': '2012-10-17',31 'Statement': [32 {33 'Action': [34 'sns:Publish'35 ],36 'Effect': 'Allow',37 'Resource': sns_arn38 }39 ]40 })41)42IAM.put_user_policy(43 UserName=username,44 PolicyName='sqs_sender',45 PolicyDocument=json.dumps({46 'Version': '2012-10-17',47 'Statement': [48 {49 'Action': [50 'sqs:SendMessage'51 ],52 'Effect': 'Allow',53 'Resource': sqs_arn54 }55 ]56 })...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
