This commit is contained in:
Stephan 2019-04-29 13:26:33 +02:00
commit 498419462d
5 changed files with 231 additions and 122 deletions

View file

@ -4,8 +4,9 @@ import json
import sure #noqa
import boto3
from botocore.exceptions import ClientError
from moto import mock_iot
from botocore.exceptions import ClientError
from nose.tools import assert_raises
@mock_iot
def test_attach_policy():
@ -384,6 +385,96 @@ def test_certs():
res.should.have.key('certificates').which.should.have.length_of(0)
@mock_iot
def test_delete_policy_validation():
doc = """{
"Version": "2012-10-17",
"Statement":[
{
"Effect":"Allow",
"Action":[
"iot: *"
],
"Resource":"*"
}
]
}
"""
client = boto3.client('iot', region_name='ap-northeast-1')
cert = client.create_keys_and_certificate(setAsActive=True)
cert_arn = cert['certificateArn']
policy_name = 'my-policy'
client.create_policy(policyName=policy_name, policyDocument=doc)
client.attach_principal_policy(policyName=policy_name, principal=cert_arn)
with assert_raises(ClientError) as e:
client.delete_policy(policyName=policy_name)
e.exception.response['Error']['Message'].should.contain(
'The policy cannot be deleted as the policy is attached to one or more principals (name=%s)' % policy_name)
res = client.list_policies()
res.should.have.key('policies').which.should.have.length_of(1)
client.detach_principal_policy(policyName=policy_name, principal=cert_arn)
client.delete_policy(policyName=policy_name)
res = client.list_policies()
res.should.have.key('policies').which.should.have.length_of(0)
@mock_iot
def test_delete_certificate_validation():
doc = """{
"Version": "2012-10-17",
"Statement":[
{
"Effect":"Allow",
"Action":[
"iot: *"
],
"Resource":"*"
}
]
}
"""
client = boto3.client('iot', region_name='ap-northeast-1')
cert = client.create_keys_and_certificate(setAsActive=True)
cert_id = cert['certificateId']
cert_arn = cert['certificateArn']
policy_name = 'my-policy'
thing_name = 'thing-1'
client.create_policy(policyName=policy_name, policyDocument=doc)
client.attach_principal_policy(policyName=policy_name, principal=cert_arn)
client.create_thing(thingName=thing_name)
client.attach_thing_principal(thingName=thing_name, principal=cert_arn)
with assert_raises(ClientError) as e:
client.delete_certificate(certificateId=cert_id)
e.exception.response['Error']['Message'].should.contain(
'Certificate must be deactivated (not ACTIVE) before deletion.')
res = client.list_certificates()
res.should.have.key('certificates').which.should.have.length_of(1)
client.update_certificate(certificateId=cert_id, newStatus='REVOKED')
with assert_raises(ClientError) as e:
client.delete_certificate(certificateId=cert_id)
e.exception.response['Error']['Message'].should.contain(
'Things must be detached before deletion (arn: %s)' % cert_arn)
res = client.list_certificates()
res.should.have.key('certificates').which.should.have.length_of(1)
client.detach_thing_principal(thingName=thing_name, principal=cert_arn)
with assert_raises(ClientError) as e:
client.delete_certificate(certificateId=cert_id)
e.exception.response['Error']['Message'].should.contain(
'Certificate policies must be detached before deletion (arn: %s)' % cert_arn)
res = client.list_certificates()
res.should.have.key('certificates').which.should.have.length_of(1)
client.detach_principal_policy(policyName=policy_name, principal=cert_arn)
client.delete_certificate(certificateId=cert_id)
res = client.list_certificates()
res.should.have.key('certificates').which.should.have.length_of(0)
@mock_iot
def test_certs_create_inactive():
client = boto3.client('iot', region_name='ap-northeast-1')
@ -432,6 +523,47 @@ def test_policy():
@mock_iot
def test_principal_policy():
client = boto3.client('iot', region_name='ap-northeast-1')
policy_name = 'my-policy'
doc = '{}'
client.create_policy(policyName=policy_name, policyDocument=doc)
cert = client.create_keys_and_certificate(setAsActive=True)
cert_arn = cert['certificateArn']
client.attach_policy(policyName=policy_name, target=cert_arn)
res = client.list_principal_policies(principal=cert_arn)
res.should.have.key('policies').which.should.have.length_of(1)
for policy in res['policies']:
policy.should.have.key('policyName').which.should_not.be.none
policy.should.have.key('policyArn').which.should_not.be.none
# do nothing if policy have already attached to certificate
client.attach_policy(policyName=policy_name, target=cert_arn)
res = client.list_principal_policies(principal=cert_arn)
res.should.have.key('policies').which.should.have.length_of(1)
for policy in res['policies']:
policy.should.have.key('policyName').which.should_not.be.none
policy.should.have.key('policyArn').which.should_not.be.none
res = client.list_policy_principals(policyName=policy_name)
res.should.have.key('principals').which.should.have.length_of(1)
for principal in res['principals']:
principal.should_not.be.none
client.detach_policy(policyName=policy_name, target=cert_arn)
res = client.list_principal_policies(principal=cert_arn)
res.should.have.key('policies').which.should.have.length_of(0)
res = client.list_policy_principals(policyName=policy_name)
res.should.have.key('principals').which.should.have.length_of(0)
with assert_raises(ClientError) as e:
client.detach_policy(policyName=policy_name, target=cert_arn)
e.exception.response['Error']['Code'].should.equal('ResourceNotFoundException')
@mock_iot
def test_principal_policy_deprecated():
client = boto3.client('iot', region_name='ap-northeast-1')
policy_name = 'my-policy'
doc = '{}'