You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@libcloud.apache.org by to...@apache.org on 2020/08/30 14:50:15 UTC
[libcloud] 06/10: Flake8 and MyPy compliance for outscale driver
This is an automated email from the ASF dual-hosted git repository.
tomaz pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/libcloud.git
commit 6f78c1aafb8bb48e36ee6f90e5e22d691110ea69
Author: Tio Gobin <ti...@outscale.com>
AuthorDate: Tue Aug 25 11:41:57 2020 +0200
Flake8 and MyPy compliance for outscale driver
---
libcloud/common/osc.py | 25 +++---
libcloud/compute/drivers/outscale.py | 148 +++++++++++++++++++++--------------
2 files changed, 101 insertions(+), 72 deletions(-)
diff --git a/libcloud/common/osc.py b/libcloud/common/osc.py
index a00d155..475f327 100644
--- a/libcloud/common/osc.py
+++ b/libcloud/common/osc.py
@@ -54,7 +54,7 @@ class OSCRequestSigner(object):
class OSCRequestSignerAlgorithmV4(OSCRequestSigner):
@staticmethod
- def sign(key: str, msg: str):
+ def sign(key, msg):
return hmac.new(key, msg.encode("utf-8"), hashlib.sha256).digest()
@staticmethod
@@ -73,7 +73,7 @@ class OSCRequestSignerAlgorithmV4(OSCRequestSigner):
for k, v in sorted(params.items())])
def get_request_headers(self, service_name: str, region: str, action: str,
- data: dict = None):
+ data: str):
date = datetime.utcnow()
host = "{}.{}.outscale.com".format(service_name, region)
headers = {
@@ -97,10 +97,10 @@ class OSCRequestSignerAlgorithmV4(OSCRequestSigner):
return headers
def _get_authorization_v4_header(self, headers: dict,
+ data: str,
dt: datetime,
method: str = 'GET',
- path: str = '/',
- data: dict = None):
+ path: str = '/'):
credentials_scope = self._get_credential_scope(dt=dt)
signed_headers = self._get_signed_headers(headers=headers)
signature = self._get_signature(headers=headers, dt=dt,
@@ -115,23 +115,25 @@ class OSCRequestSignerAlgorithmV4(OSCRequestSigner):
}
def _get_signature(self, headers: dict, dt: datetime,
- method: str, path: str, data: dict):
+ method: str, path: str, data: str):
string_to_sign = self._get_string_to_sign(headers=headers, dt=dt,
method=method, path=path,
data=data)
- signing_key = self._get_key_to_sign_with(self.access_secret, dt)
+ signing_key = self._get_key_to_sign_with(
+ self.access_secret,
+ dt.strftime('%Y%m%d')
+ )
return hmac.new(signing_key, string_to_sign.encode('utf-8'),
hashlib.sha256).hexdigest()
- def _get_key_to_sign_with(self, key: str, dt: datetime):
- dt = dt.strftime('%Y%m%d')
+ def _get_key_to_sign_with(self, key: str, dt: str):
k_date = self.sign(('OSC4' + key).encode('utf-8'), dt)
k_region = self.sign(k_date, self.connection.region_name)
k_service = self.sign(k_region, self.connection.service_name)
return self.sign(k_service, 'osc4_request')
def _get_string_to_sign(self, headers: dict, dt: datetime, method: str,
- path: str, data: dict):
+ path: str, data: str):
canonical_request = self._get_canonical_request(headers=headers,
method=method,
path=path,
@@ -147,8 +149,7 @@ class OSCRequestSignerAlgorithmV4(OSCRequestSigner):
self.connection.service_name,
'osc4_request'])
- def _get_canonical_request(self, headers, method, path, data = "{}"):
- data = data if data else "{}"
+ def _get_canonical_request(self, headers, method, path, data):
return '\n'.join([
method,
path,
@@ -156,4 +157,4 @@ class OSCRequestSignerAlgorithmV4(OSCRequestSigner):
self._get_canonical_headers(headers),
self._get_signed_headers(headers),
hashlib.sha256(data.encode('utf-8')).hexdigest()
- ])
\ No newline at end of file
+ ])
diff --git a/libcloud/compute/drivers/outscale.py b/libcloud/compute/drivers/outscale.py
index fe1b80b..df43b10 100644
--- a/libcloud/compute/drivers/outscale.py
+++ b/libcloud/compute/drivers/outscale.py
@@ -114,8 +114,8 @@ class OutscaleNodeDriver(NodeDriver):
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
-
- if requests.post(endpoint, data=data, headers=headers).status_code == 200:
+ response = requests.post(endpoint, data=data, headers=headers)
+ if response.status_code == 200:
return True
return False
@@ -153,7 +153,8 @@ class OutscaleNodeDriver(NodeDriver):
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
- if requests.post(endpoint, data=data, headers=headers).status_code == 200:
+ response = requests.post(endpoint, data=data, headers=headers)
+ if response.status_code == 200:
return True
return False
@@ -252,7 +253,8 @@ class OutscaleNodeDriver(NodeDriver):
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
- if requests.post(endpoint, data=data, headers=headers).status_code == 200:
+ response = requests.post(endpoint, data=data, headers=headers)
+ if response.status_code == 200:
return True
return False
@@ -290,7 +292,8 @@ class OutscaleNodeDriver(NodeDriver):
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
- if requests.post(endpoint, data=data, headers=headers).status_code == 200:
+ response = requests.post(endpoint, data=data, headers=headers)
+ if response.status_code == 200:
return True
return False
@@ -387,11 +390,14 @@ class OutscaleNodeDriver(NodeDriver):
create the VM.
:type ex_subnet_id: ``str``
- :param ex_user_data: Data or script used to add a specific configuration to the VM. It must be base64-encoded.
+ :param ex_user_data: Data or script used to add a specific
+ configuration to the VM. It must be base64-encoded.
:type ex_user_data: ``str``
- :param ex_vm_initiated_shutdown_behavior: The VM behavior when you stop it. By default or if set to stop, the
- VM stops. If set to restart, the VM stops then automatically restarts. If set to terminate, the VM stops and is terminated.
+ :param ex_vm_initiated_shutdown_behavior: The VM behavior when
+ you stop it. By default or if set to stop, the
+ VM stops. If set to restart, the VM stops then automatically restarts.
+ If set to terminate, the VM stops and is terminated.
create the VM.
:type ex_vm_initiated_shutdown_behavior: ``str``
@@ -434,7 +440,10 @@ class OutscaleNodeDriver(NodeDriver):
if ex_user_data is not None:
data.update({"UserData": ex_user_data})
if ex_vm_initiated_shutdown_behavior is not None:
- data.update({"VmInstantiatedShutdownBehavior": ex_vm_initiated_shutdown_behavior})
+ data.update({
+ "VmInstantiatedShutdownBehavior":
+ ex_vm_initiated_shutdown_behavior
+ })
if ex_vm_type is not None:
data.update({"VmType": ex_vm_type})
if ex_subnet_id is not None:
@@ -445,8 +454,8 @@ class OutscaleNodeDriver(NodeDriver):
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
- vm = requests.post(endpoint, data=data, headers=headers).json()["Vms"][0]
-
+ response = requests.post(endpoint, data=data, headers=headers)
+ vm = response.json()["Vms"][0]
return self._to_node(vm)
def reboot_node(self, node: Node):
@@ -466,7 +475,8 @@ class OutscaleNodeDriver(NodeDriver):
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
- if requests.post(endpoint, data=data, headers=headers).status_code == 200:
+ response = requests.post(endpoint, data=data, headers=headers)
+ if response.status_code == 200:
return False
return False
@@ -482,8 +492,8 @@ class OutscaleNodeDriver(NodeDriver):
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
- vms = requests.post(endpoint, data=ex_data, headers=headers).json()["Vms"]
- return self._to_nodes(vms)
+ vms = requests.post(endpoint, data=ex_data, headers=headers)
+ return self._to_nodes(vms.json()["Vms"])
def destroy_node(self, node: Node):
"""
@@ -501,7 +511,8 @@ class OutscaleNodeDriver(NodeDriver):
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
- if requests.post(endpoint, data=data, headers=headers).status_code == 200:
+ response = requests.post(endpoint, data=data, headers=headers)
+ if response.status_code == 200:
return True
return False
@@ -587,8 +598,8 @@ class OutscaleNodeDriver(NodeDriver):
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
- image = requests.post(endpoint, data=data, headers=headers).json()["Image"]
- return self._to_node_images(image)
+ image = requests.post(endpoint, data=data, headers=headers).json()
+ return self._to_node_image(image["Image"])
def list_images(self, ex_data: str = "{}"):
"""
@@ -602,8 +613,8 @@ class OutscaleNodeDriver(NodeDriver):
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
- images = requests.post(endpoint, data=ex_data, headers=headers).json()["Images"]
- return self._to_node_images(images)
+ images = requests.post(endpoint, data=ex_data, headers=headers).json()
+ return self._to_node_images(images["Images"])
def get_image(self, image_id: str):
"""
@@ -621,8 +632,8 @@ class OutscaleNodeDriver(NodeDriver):
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
- images = requests.post(endpoint, data=data, headers=headers).json()["Images"]
- return self._to_node_image(images)
+ images = requests.post(endpoint, data=data, headers=headers).json()
+ return self._to_node_image(images["Images"])
def delete_image(self, node_image: NodeImage):
"""
@@ -640,7 +651,8 @@ class OutscaleNodeDriver(NodeDriver):
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
- if requests.post(endpoint, data=data, headers=headers).status_code == 200:
+ response = requests.post(endpoint, data=data, headers=headers)
+ if response.status_code == 200:
return True
return False
@@ -648,7 +660,9 @@ class OutscaleNodeDriver(NodeDriver):
return [self._to_key_pair(key_pair) for key_pair in key_pairs]
def _to_key_pair(self, key_pair):
- private_key = key_pair["PrivateKey"] if "PrivateKey" in key_pair else ""
+ private_key = ""
+ if "PrivateKey" in key_pair:
+ private_key = key_pair["PrivateKey"]
return KeyPair(
name=key_pair["KeypairName"],
public_key="",
@@ -704,8 +718,8 @@ class OutscaleNodeDriver(NodeDriver):
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
- key_pairs = requests.post(endpoint, data=ex_data, headers=headers).json()
- return self._to_key_pairs(key_pairs["Keypairs"])
+ key_pairs = requests.post(endpoint, data=ex_data, headers=headers)
+ return self._to_key_pairs(key_pairs.json()["Keypairs"])
def get_key_pair(self, name: str):
"""
@@ -724,14 +738,15 @@ class OutscaleNodeDriver(NodeDriver):
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
- key_pair = requests.post(endpoint, data=data, headers=headers).json()["Keypairs"][0]
- return self._to_key_pair(key_pair)
+ key_pair = requests.post(endpoint, data=data, headers=headers).json()
+ return self._to_key_pair(key_pair["Keypairs"][0])
def delete_key_pair(self, key_pair: KeyPair):
"""
Delete an image.
- :param key_pair: the name of the keypair you want to delete (required)
+ :param key_pair: the name of the keypair
+ you want to delete (required)
:type key_pair: ``KeyPair``
:return: boolean
@@ -743,7 +758,8 @@ class OutscaleNodeDriver(NodeDriver):
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
- if requests.post(endpoint, data=data, headers=headers).status_code == 200:
+ response = requests.post(endpoint, data=data, headers=headers)
+ if response.status_code == 200:
return True
return False
@@ -814,8 +830,8 @@ class OutscaleNodeDriver(NodeDriver):
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
- snapshot = requests.post(endpoint, data=data, headers=headers).json()["Volume"]
- return self._to_snapshot(snapshot)
+ snapshot = requests.post(endpoint, data=data, headers=headers)
+ return self._to_snapshot(snapshot.json()["Volume"])
def list_snapshots(self, ex_data: str = "{}"):
"""
@@ -829,9 +845,8 @@ class OutscaleNodeDriver(NodeDriver):
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
- snapshots = requests.post(endpoint, data=ex_data, headers=headers).json()
-
- return self._to_snapshots(snapshots)
+ snapshots = requests.post(endpoint, data=ex_data, headers=headers)
+ return self._to_snapshots(snapshots.json()["Snapshots"])
def destroy_volume_snapshot(self, snapshot: VolumeSnapshot):
"""
@@ -839,10 +854,10 @@ class OutscaleNodeDriver(NodeDriver):
:param snapshot: the ID of the snapshot
you want to delete (required)
- :type snapshot: ``str``
+ :type snapshot: ``VolumeSnapshot``
:return: request
- :rtype: ``dict``
+ :rtype: ``bool``
"""
action = "DeleteSnapshot"
data = '{"SnapshotId": "' + snapshot.id + '"}'
@@ -850,7 +865,8 @@ class OutscaleNodeDriver(NodeDriver):
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
- if requests.post(endpoint, data=data, headers=headers).status_code == 200:
+ response = requests.post(endpoint, data=data, headers=headers)
+ if response.status_code == 200:
return True
return False
@@ -914,8 +930,8 @@ class OutscaleNodeDriver(NodeDriver):
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
- volume = requests.post(endpoint, data=data, headers=headers).json()["Volume"]
- return self._to_volume(volume)
+ volume = requests.post(endpoint, data=data, headers=headers)
+ return self._to_volume(volume.json()["Volume"])
def list_volumes(self, ex_data: str = "{}"):
"""
@@ -929,8 +945,8 @@ class OutscaleNodeDriver(NodeDriver):
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
- volumes = requests.post(endpoint, data=ex_data, headers=headers).json()["Volumes"]
- return self._to_volumes(volumes)
+ volumes = requests.post(endpoint, data=ex_data, headers=headers)
+ return self._to_volumes(volumes.json()["Volumes"])
def destroy_volume(self, volume: StorageVolume):
"""
@@ -938,10 +954,10 @@ class OutscaleNodeDriver(NodeDriver):
:param volume: the ID of the volume
you want to delete (required)
- :type volume: ``str``
+ :type volume: ``StorageVolume``
:return: request
- :rtype: ``dict``
+ :rtype: ``bool``
"""
action = "DeleteVolume"
data = '{"VolumeId": "' + volume.id + '"}'
@@ -949,21 +965,27 @@ class OutscaleNodeDriver(NodeDriver):
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
- if requests.post(endpoint, data=data, headers=headers).status_code == 200:
+ response = requests.post(endpoint, data=data, headers=headers)
+ if response.status_code == 200:
return True
return False
- def attach_volume(self, node: Node, volume: StorageVolume, device: str = None):
+ def attach_volume(
+ self,
+ node: Node,
+ volume: StorageVolume,
+ device: str = None
+ ):
"""
Attach a volume.
:param node: the ID of the VM you want
to attach the volume to (required)
- :type node: ``str``
+ :type node: ``Node``
:param volume: the ID of the volume
you want to attach (required)
- :type volume: ``str``
+ :type volume: ``StorageVolume``
:param device: the name of the device (required)
:type device: ``str``
@@ -981,7 +1003,8 @@ class OutscaleNodeDriver(NodeDriver):
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
- if requests.post(endpoint, data=data, headers=headers).status_code == 200:
+ response = requests.post(endpoint, data=data, headers=headers)
+ if response.status_code == 200:
return True
return False
@@ -1017,7 +1040,8 @@ class OutscaleNodeDriver(NodeDriver):
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
- if requests.post(endpoint, data=data, headers=headers).status_code == 200:
+ response = requests.post(endpoint, data=data, headers=headers)
+ if response.status_code == 200:
return True
return False
@@ -1029,18 +1053,22 @@ class OutscaleNodeDriver(NodeDriver):
action
)
- def _ex_generate_headers(self, action: str, data: dict):
- return self.signer.get_request_headers(action=action,
- data=data,
- service_name=self.service_name,
- region=self.region)
+ def _ex_generate_headers(self, action: str, data: str):
+ return self.signer.get_request_headers(
+ action=action,
+ data=data,
+ service_name=self.service_name,
+ region=self.region
+ )
def _to_location(self, region):
- return NodeLocation(id="",
- name=region["RegionName"],
- country="",
- driver=self,
- extra=region)
+ return NodeLocation(
+ id="",
+ name=region["RegionName"],
+ country="",
+ driver=self,
+ extra=region
+ )
def _to_locations(self, regions: list):
return [self._to_location(region) for region in regions]
@@ -1053,7 +1081,7 @@ class OutscaleNodeDriver(NodeDriver):
return VolumeSnapshot(
id=snapshot["SnapshotId"],
name=name,
- size=snapshot["Size"],
+ size=snapshot["VolumeSize"],
driver=self,
state=snapshot["State"],
created=None,