moto/tests/test_iot/test_iot_ca_certificates.py

import boto3
import pytest

from botocore.exceptions import ClientError
from moto import mock_iot


@mock_iot
def test_register_ca_certificate_simple():
    client = boto3.client("iot", region_name="us-east-1")

    resp = client.register_ca_certificate(
        caCertificate="ca_certificate",
        verificationCertificate="verification_certificate",
    )

    resp.should.have.key("certificateArn")
    resp.should.have.key("certificateId")


@mock_iot
def test_describe_ca_certificate_unknown():
    client = boto3.client("iot", region_name="us-east-2")

    with pytest.raises(ClientError) as exc:
        client.describe_ca_certificate(certificateId="a" * 70)
    err = exc.value.response["Error"]
    err["Code"].should.equal("ResourceNotFoundException")


@mock_iot
def test_describe_ca_certificate_simple():
    client = boto3.client("iot", region_name="us-east-1")

    resp = client.register_ca_certificate(
        caCertificate="ca_certificate",
        verificationCertificate="verification_certificate",
    )

    describe = client.describe_ca_certificate(certificateId=resp["certificateId"])

    describe.should.have.key("certificateDescription")
    description = describe["certificateDescription"]
    description.should.have.key("certificateArn").equals(resp["certificateArn"])
    description.should.have.key("certificateId").equals(resp["certificateId"])
    description.should.have.key("status").equals("INACTIVE")
    description.should.have.key("certificatePem").equals("ca_certificate")


@mock_iot
def test_describe_ca_certificate_advanced():
    client = boto3.client("iot", region_name="us-east-1")

    resp = client.register_ca_certificate(
        caCertificate="ca_certificate",
        verificationCertificate="verification_certificate",
        setAsActive=True,
        registrationConfig={
            "templateBody": "template_b0dy",
            "roleArn": "aws:iot:arn:role/asdfqwerwe",
        },
    )

    describe = client.describe_ca_certificate(certificateId=resp["certificateId"])

    describe.should.have.key("certificateDescription")
    description = describe["certificateDescription"]
    description.should.have.key("certificateArn").equals(resp["certificateArn"])
    description.should.have.key("certificateId").equals(resp["certificateId"])
    description.should.have.key("status").equals("ACTIVE")
    description.should.have.key("certificatePem").equals("ca_certificate")

    describe.should.have.key("registrationConfig")
    config = describe["registrationConfig"]
    config.should.have.key("templateBody").equals("template_b0dy")
    config.should.have.key("roleArn").equals("aws:iot:arn:role/asdfqwerwe")


@mock_iot
def test_list_certificates_by_ca():
    client = boto3.client("iot", region_name="us-east-1")

    # create ca
    ca_cert = client.register_ca_certificate(
        caCertificate="ca_certificate",
        verificationCertificate="verification_certificate",
    )
    ca_cert_id = ca_cert["certificateId"]

    # list certificates should be empty at first
    certs = client.list_certificates_by_ca(caCertificateId=ca_cert_id)
    certs.should.have.key("certificates").equals([])

    # create one certificate
    cert1 = client.register_certificate(
        certificatePem="pem" * 20, caCertificatePem="ca_certificate", setAsActive=False
    )

    # list certificates should return this
    certs = client.list_certificates_by_ca(caCertificateId=ca_cert_id)
    certs.should.have.key("certificates").length_of(1)
    certs["certificates"][0]["certificateId"].should.equal(cert1["certificateId"])

    # create another certificate, without ca
    client.register_certificate(
        certificatePem="pam" * 20,
        caCertificatePem="unknown_ca_certificate",
        setAsActive=False,
    )

    # list certificate should still only return the first certificate
    certs = client.list_certificates_by_ca(caCertificateId=ca_cert_id)
    certs.should.have.key("certificates").length_of(1)


@mock_iot
def test_delete_ca_certificate():
    client = boto3.client("iot", region_name="us-east-1")

    cert_id = client.register_ca_certificate(
        caCertificate="ca_certificate",
        verificationCertificate="verification_certificate",
    )["certificateId"]

    client.delete_ca_certificate(certificateId=cert_id)

    with pytest.raises(ClientError) as exc:
        client.describe_ca_certificate(certificateId=cert_id)
    err = exc.value.response["Error"]
    err["Code"].should.equal("ResourceNotFoundException")


@mock_iot
def test_update_ca_certificate__status():
    client = boto3.client("iot", region_name="us-east-1")
    cert_id = client.register_ca_certificate(
        caCertificate="ca_certificate",
        verificationCertificate="verification_certificate",
        registrationConfig={"templateBody": "tb", "roleArn": "my:old_and_busted:arn"},
    )["certificateId"]

    client.update_ca_certificate(certificateId=cert_id, newStatus="DISABLE")
    cert = client.describe_ca_certificate(certificateId=cert_id)
    cert_desc = cert["certificateDescription"]
    cert_desc.should.have.key("status").which.should.equal("DISABLE")
    cert.should.have.key("registrationConfig").equals(
        {"roleArn": "my:old_and_busted:arn", "templateBody": "tb"}
    )


@mock_iot
def test_update_ca_certificate__config():
    client = boto3.client("iot", region_name="us-east-1")
    cert_id = client.register_ca_certificate(
        caCertificate="ca_certificate",
        verificationCertificate="verification_certificate",
    )["certificateId"]

    client.update_ca_certificate(
        certificateId=cert_id,
        registrationConfig={"templateBody": "tb", "roleArn": "my:new_and_fancy:arn"},
    )
    cert = client.describe_ca_certificate(certificateId=cert_id)
    cert_desc = cert["certificateDescription"]
    cert_desc.should.have.key("status").which.should.equal("INACTIVE")
    cert.should.have.key("registrationConfig").equals(
        {"roleArn": "my:new_and_fancy:arn", "templateBody": "tb"}
    )


@mock_iot
def test_get_registration_code():
    client = boto3.client("iot", region_name="us-west-1")

    resp = client.get_registration_code()
    resp.should.have.key("registrationCode")