From ce4059f6d9902eca6eb0f259118eb5f13706f9da Mon Sep 17 00:00:00 2001 From: Don Kuntz Date: Mon, 22 Jul 2019 21:50:09 -0500 Subject: [PATCH 1/2] Use a consistent owner id between EC2 resources Previously there were a couple models which used different owner ids by default, which could make tests relying on them fail if someone wasn't expecting that. This change ensures a uniform owner id between resources. --- moto/ec2/models.py | 14 ++++++++------ tests/test_ec2/test_elastic_block_store.py | 3 ++- 2 files changed, 10 insertions(+), 7 deletions(-) diff --git a/moto/ec2/models.py b/moto/ec2/models.py index 60a0257b0..41a84ec48 100644 --- a/moto/ec2/models.py +++ b/moto/ec2/models.py @@ -142,6 +142,8 @@ AMIS = json.load( __name__, 'resources/amis.json'), 'r') ) +OWNER_ID = "111122223333" + def utc_date_and_time(): return datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%S.000Z') @@ -1087,7 +1089,7 @@ class TagBackend(object): class Ami(TaggedEC2Resource): def __init__(self, ec2_backend, ami_id, instance=None, source_ami=None, - name=None, description=None, owner_id=111122223333, + name=None, description=None, owner_id=OWNER_ID, public=False, virtualization_type=None, architecture=None, state='available', creation_date=None, platform=None, image_type='machine', image_location=None, hypervisor=None, @@ -1200,7 +1202,7 @@ class AmiBackend(object): ami = Ami(self, ami_id, instance=instance, source_ami=None, name=name, description=description, - owner_id=context.get_current_user() if context else '111122223333') + owner_id=context.get_current_user() if context else OWNER_ID) self.amis[ami_id] = ami return ami @@ -1468,7 +1470,7 @@ class SecurityGroup(TaggedEC2Resource): self.egress_rules = [SecurityRule(-1, None, None, ['0.0.0.0/0'], [])] self.enis = {} self.vpc_id = vpc_id - self.owner_id = "123456789012" + self.owner_id = OWNER_ID @classmethod def create_from_cloudformation_json(cls, resource_name, cloudformation_json, region_name): @@ -1989,7 +1991,7 @@ class Volume(TaggedEC2Resource): class Snapshot(TaggedEC2Resource): - def __init__(self, ec2_backend, snapshot_id, volume, description, encrypted=False, owner_id='123456789012'): + def __init__(self, ec2_backend, snapshot_id, volume, description, encrypted=False, owner_id=OWNER_ID): self.id = snapshot_id self.volume = volume self.description = description @@ -2491,7 +2493,7 @@ class VPCPeeringConnectionBackend(object): class Subnet(TaggedEC2Resource): def __init__(self, ec2_backend, subnet_id, vpc_id, cidr_block, availability_zone, default_for_az, - map_public_ip_on_launch, owner_id=111122223333, assign_ipv6_address_on_creation=False): + map_public_ip_on_launch, owner_id=OWNER_ID, assign_ipv6_address_on_creation=False): self.ec2_backend = ec2_backend self.id = subnet_id self.vpc_id = vpc_id @@ -2657,7 +2659,7 @@ class SubnetBackend(object): raise InvalidAvailabilityZoneError(availability_zone, ", ".join([zone.name for zones in RegionsAndZonesBackend.zones.values() for zone in zones])) subnet = Subnet(self, subnet_id, vpc_id, cidr_block, availability_zone_data, default_for_az, map_public_ip_on_launch, - owner_id=context.get_current_user() if context else '111122223333', assign_ipv6_address_on_creation=False) + owner_id=context.get_current_user() if context else OWNER_ID, assign_ipv6_address_on_creation=False) # AWS associates a new subnet with the default Network ACL self.associate_default_network_acl_with_subnet(subnet_id, vpc_id) diff --git a/tests/test_ec2/test_elastic_block_store.py b/tests/test_ec2/test_elastic_block_store.py index ab5b31ba0..9dbaa5ea6 100644 --- a/tests/test_ec2/test_elastic_block_store.py +++ b/tests/test_ec2/test_elastic_block_store.py @@ -12,6 +12,7 @@ from freezegun import freeze_time import sure # noqa from moto import mock_ec2_deprecated, mock_ec2 +from moto.ec2.models import OWNER_ID @mock_ec2_deprecated @@ -395,7 +396,7 @@ def test_snapshot_filters(): ).should.equal({snapshot3.id}) snapshots_by_owner_id = conn.get_all_snapshots( - filters={'owner-id': '123456789012'}) + filters={'owner-id': OWNER_ID}) set([snap.id for snap in snapshots_by_owner_id] ).should.equal({snapshot1.id, snapshot2.id, snapshot3.id}) From abf3db8d8a4287946a7dc0c5dfaa233139e3e71f Mon Sep 17 00:00:00 2001 From: Don Kuntz Date: Mon, 22 Jul 2019 21:57:15 -0500 Subject: [PATCH 2/2] Add a test to ensure that ec2.copy_image sets the proper owner id This test is useful because before the last commit using copy_image would not set the owner_id to the same one used when calling describe_images. For example, this code conn = boto3.client("ec2") copy_resp = conn.copy_image( SourceImageId="ami-whatever", ... ) describe_resp = conn.describe_images( Owners=["self"] ) Would result in describe_resp being empty, when it should contain the image from the copy_resp before it. By ensuring the owner ids are the same (see ce4059f6) the code example now works as expected. --- tests/test_ec2/test_amis.py | 25 ++++++++++++++++++++++++- 1 file changed, 24 insertions(+), 1 deletion(-) diff --git a/tests/test_ec2/test_amis.py b/tests/test_ec2/test_amis.py index fd7234511..feff4a16c 100644 --- a/tests/test_ec2/test_amis.py +++ b/tests/test_ec2/test_amis.py @@ -10,7 +10,7 @@ from nose.tools import assert_raises import sure # noqa from moto import mock_ec2_deprecated, mock_ec2 -from moto.ec2.models import AMIS +from moto.ec2.models import AMIS, OWNER_ID from tests.helpers import requires_boto_gte @@ -152,6 +152,29 @@ def test_ami_copy(): cm.exception.request_id.should_not.be.none +@mock_ec2 +def test_copy_image_changes_owner_id(): + conn = boto3.client('ec2', region_name='us-east-1') + + # this source AMI ID is from moto/ec2/resources/amis.json + source_ami_id = "ami-03cf127a" + + # confirm the source ami owner id is different from the default owner id. + # if they're ever the same it means this test is invalid. + check_resp = conn.describe_images(ImageIds=[source_ami_id]) + check_resp["Images"][0]["OwnerId"].should_not.equal(OWNER_ID) + + copy_resp = conn.copy_image( + SourceImageId=source_ami_id, + Name="new-image", + Description="a copy of an image", + SourceRegion="us-east-1") + + describe_resp = conn.describe_images(Owners=["self"]) + describe_resp["Images"][0]["OwnerId"].should.equal(OWNER_ID) + describe_resp["Images"][0]["ImageId"].should.equal(copy_resp["ImageId"]) + + @mock_ec2_deprecated def test_ami_tagging(): conn = boto.connect_vpc('the_key', 'the_secret')