Add support for IoT detach_policy

This commit is contained in:
Tomoya Iwata 2018-11-06 17:12:13 +09:00
parent 249dd7059e
commit e52158f811
4 changed files with 26 additions and 3 deletions

View File

@ -2376,7 +2376,7 @@
- [ ] unsubscribe_from_event
- [ ] update_assessment_target
## iot - 31% implemented
## iot - 32% implemented
- [ ] accept_certificate_transfer
- [X] add_thing_to_thing_group
- [ ] associate_targets_with_job
@ -2429,7 +2429,7 @@
- [X] describe_thing_group
- [ ] describe_thing_registration_task
- [X] describe_thing_type
- [ ] detach_policy
- [X] detach_policy
- [X] detach_principal_policy
- [X] detach_thing_principal
- [ ] disable_topic_rule

View File

@ -445,6 +445,15 @@ class IoTBackend(BaseBackend):
return
self.principal_policies[k] = (principal, policy)
def detach_policy(self, policy_name, target):
# this may raises ResourceNotFoundException
self._get_principal(target)
self.get_policy(policy_name)
k = (target, policy_name)
if k not in self.principal_policies:
raise ResourceNotFoundException()
del self.principal_policies[k]
def detach_principal_policy(self, policy_name, principal_arn):
# this may raises ResourceNotFoundException
self._get_principal(principal_arn)

View File

@ -242,6 +242,15 @@ class IoTResponse(BaseResponse):
)
return json.dumps(dict())
def detach_policy(self):
policy_name = self._get_param("policyName")
target = self._get_param('target')
self.iot_backend.detach_policy(
policy_name=policy_name,
target=target,
)
return json.dumps(dict())
def detach_principal_policy(self):
policy_name = self._get_param("policyName")
principal = self.headers.get('x-amzn-iot-principal')

View File

@ -5,6 +5,8 @@ import sure # noqa
import boto3
from moto import mock_iot
from botocore.exceptions import ClientError
from nose.tools import assert_raises
@mock_iot
@ -338,11 +340,14 @@ def test_principal_policy():
for principal in res['principals']:
principal.should_not.be.none
client.detach_principal_policy(policyName=policy_name, principal=cert_arn)
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