Filter certs by statuses.
This commit is contained in:
		
							parent
							
								
									94fd5c4128
								
							
						
					
					
						commit
						d669145b71
					
				@ -325,7 +325,7 @@ class AWSCertificateManagerBackend(BaseBackend):
 | 
			
		||||
 | 
			
		||||
        return bundle.arn
 | 
			
		||||
 | 
			
		||||
    def get_certificates_list(self):
 | 
			
		||||
    def get_certificates_list(self, statuses):
 | 
			
		||||
        """
 | 
			
		||||
        Get list of certificates
 | 
			
		||||
 | 
			
		||||
@ -333,7 +333,9 @@ class AWSCertificateManagerBackend(BaseBackend):
 | 
			
		||||
        :rtype: list of CertBundle
 | 
			
		||||
        """
 | 
			
		||||
        for arn in self._certificates.keys():
 | 
			
		||||
            yield self.get_certificate(arn)
 | 
			
		||||
            cert = self.get_certificate(arn)
 | 
			
		||||
            if not statuses or cert.status in statuses:
 | 
			
		||||
                yield cert
 | 
			
		||||
 | 
			
		||||
    def get_certificate(self, arn):
 | 
			
		||||
        if arn not in self._certificates:
 | 
			
		||||
 | 
			
		||||
@ -132,8 +132,8 @@ class AWSCertificateManagerResponse(BaseResponse):
 | 
			
		||||
 | 
			
		||||
    def list_certificates(self):
 | 
			
		||||
        certs = []
 | 
			
		||||
 | 
			
		||||
        for cert_bundle in self.acm_backend.get_certificates_list():
 | 
			
		||||
        statuses = self._get_param('CertificateStatuses')
 | 
			
		||||
        for cert_bundle in self.acm_backend.get_certificates_list(statuses):
 | 
			
		||||
            certs.append({
 | 
			
		||||
                'CertificateArn': cert_bundle.arn,
 | 
			
		||||
                'DomainName': cert_bundle.common_name
 | 
			
		||||
 | 
			
		||||
@ -74,6 +74,31 @@ def test_list_certificates():
 | 
			
		||||
    resp['CertificateSummaryList'][0]['DomainName'].should.equal(SERVER_COMMON_NAME)
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
@mock_acm
 | 
			
		||||
def test_list_certificates_by_status():
 | 
			
		||||
    client = boto3.client('acm', region_name='eu-central-1')
 | 
			
		||||
    issued_arn = _import_cert(client)
 | 
			
		||||
    pending_arn = client.request_certificate(DomainName='google.com')['CertificateArn']
 | 
			
		||||
 | 
			
		||||
    resp = client.list_certificates()
 | 
			
		||||
    len(resp['CertificateSummaryList']).should.equal(2)
 | 
			
		||||
    resp = client.list_certificates(CertificateStatuses=['EXPIRED', 'INACTIVE'])
 | 
			
		||||
    len(resp['CertificateSummaryList']).should.equal(0)
 | 
			
		||||
    resp = client.list_certificates(CertificateStatuses=['PENDING_VALIDATION'])
 | 
			
		||||
    len(resp['CertificateSummaryList']).should.equal(1)
 | 
			
		||||
    resp['CertificateSummaryList'][0]['CertificateArn'].should.equal(pending_arn)
 | 
			
		||||
 | 
			
		||||
    resp = client.list_certificates(CertificateStatuses=['ISSUED'])
 | 
			
		||||
    len(resp['CertificateSummaryList']).should.equal(1)
 | 
			
		||||
    resp['CertificateSummaryList'][0]['CertificateArn'].should.equal(issued_arn)
 | 
			
		||||
    resp = client.list_certificates(CertificateStatuses=['ISSUED', 'PENDING_VALIDATION'])
 | 
			
		||||
    len(resp['CertificateSummaryList']).should.equal(2)
 | 
			
		||||
    arns = {cert['CertificateArn'] for cert in resp['CertificateSummaryList']}
 | 
			
		||||
    arns.should.contain(issued_arn)
 | 
			
		||||
    arns.should.contain(pending_arn)
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
@mock_acm
 | 
			
		||||
def test_get_invalid_certificate():
 | 
			
		||||
    client = boto3.client('acm', region_name='eu-central-1')
 | 
			
		||||
 | 
			
		||||
		Loading…
	
	
			
			x
			
			
		
	
		Reference in New Issue
	
	Block a user