Merge pull request #424 from ZuluPro/keypairs

Implemented import key pair

moto/ec2/models.py

@@ -710,6 +710,13 @@ class KeyPairBackend(object):
 
         return results
 
+    def import_key_pair(self, key_name, public_key_material):
+        if key_name in self.keypairs:
+            raise InvalidKeyPairDuplicateError(key_name)
+        self.keypairs[key_name] = keypair = random_key_pair()
+        keypair['name'] = key_name
+        return keypair
+
 
 class TagBackend(object):
 

moto/ec2/responses/key_pairs.py

@@ -28,7 +28,11 @@ class KeyPairs(BaseResponse):
         return template.render(keypairs=keypairs)
 
     def import_key_pair(self):
-        raise NotImplementedError('KeyPairs.import_key_pair is not yet implemented')
+        name = self.querystring.get('KeyName')[0]
+        material = self.querystring.get('PublicKeyMaterial')[0]
+        keypair = self.ec2_backend.import_key_pair(name, material)
+        template = self.response_template(IMPORT_KEYPAIR_RESPONSE)
+        return template.render(**keypair)
 
 
 DESCRIBE_KEY_PAIRS_RESPONSE = """<DescribeKeyPairsResponse xmlns="http://ec2.amazonaws.com/doc/2013-10-15/">
@@ -58,3 +62,10 @@ DELETE_KEY_PAIR_RESPONSE = """<DeleteKeyPairResponse xmlns="http://ec2.amazonaws
   <requestId>59dbff89-35bd-4eac-99ed-be587EXAMPLE</requestId>
   <return>{{ success }}</return>
 </DeleteKeyPairResponse>"""
+
+IMPORT_KEYPAIR_RESPONSE = """<?xml version="1.0" encoding="UTF-8"?>
+  <ImportKeyPairResponse xmlns="http://ec2.amazonaws.com/doc/2014-06-15/">
+    <requestId>471f9fdd-8fe2-4a84-86b0-bd3d3e350979</requestId>
+    <keyName>{{ name }}</keyName>
+    <keyFingerprint>{{ fingerprint }}</keyFingerprint>
+  </ImportKeyPairResponse>"""

tests/test_ec2/test_key_pairs.py

@@ -85,3 +85,27 @@ def test_key_pairs_delete_exist():
     r = conn.delete_key_pair('foo')
     r.should.be.ok
     assert len(conn.get_all_key_pairs()) == 0
+
+
+@mock_ec2
+def test_key_pairs_import():
+    conn = boto.connect_ec2('the_key', 'the_secret')
+    kp = conn.import_key_pair('foo', b'content')
+    assert kp.name == 'foo'
+    kps = conn.get_all_key_pairs()
+    assert len(kps) == 1
+    assert kps[0].name == 'foo'
+
+
+@mock_ec2
+def test_key_pairs_import_exist():
+    conn = boto.connect_ec2('the_key', 'the_secret')
+    kp = conn.import_key_pair('foo', b'content')
+    assert kp.name == 'foo'
+    assert len(conn.get_all_key_pairs()) == 1
+
+    with assert_raises(EC2ResponseError) as cm:
+        conn.create_key_pair('foo')
+    cm.exception.code.should.equal('InvalidKeyPair.Duplicate')
+    cm.exception.status.should.equal(400)
+    cm.exception.request_id.should_not.be.none