add test and providers

Author: Polandia94

libcloud/storage/drivers/azure_blobs.py

@@ -19,8 +19,8 @@
 import base64
 import hashlib
 import binascii
-from datetime import datetime, timedelta
 from typing import Literal
+from datetime import datetime, timedelta
 
 from libcloud.utils.py3 import ET, b, httplib, tostring, urlquote, urlencode
 from libcloud.utils.xml import fixxpath

libcloud/storage/drivers/ovh.py

@@ -83,7 +83,7 @@ def __init__(
     def list_regions(self):
         return REGION_TO_HOST_MAP.keys()
 
-    def get_object_cdn_url(self, obj, ex_expiry=S3_CDN_URL_EXPIRY_HOURS):
+    def get_object_cdn_url(self, obj, ex_expiry=S3_CDN_URL_EXPIRY_HOURS, ex_method: str = "GET"):
         # In order to download (private) objects we need to be able to generate a valid CDN URL,
         # hence shamefully just use the working code from the S3StorageDriver.
-        return S3StorageDriver.get_object_cdn_url(self, obj, ex_expiry=ex_expiry)
+        return S3StorageDriver.get_object_cdn_url(self, obj, ex_expiry=ex_expiry, ex_method=ex_method)

libcloud/storage/drivers/scaleway.py

@@ -85,7 +85,7 @@ def __init__(
     def list_regions(self):
         return REGION_TO_HOST_MAP.keys()
 
-    def get_object_cdn_url(self, obj, ex_expiry=S3_CDN_URL_EXPIRY_HOURS):
+    def get_object_cdn_url(self, obj, ex_expiry=S3_CDN_URL_EXPIRY_HOURS, ex_method: str = "GET"):
         # In order to download (private) objects we need to be able to generate a valid CDN URL,
         # hence shamefully just use the working code from the S3StorageDriver.
-        return S3StorageDriver.get_object_cdn_url(self, obj, ex_expiry=ex_expiry)
+        return S3StorageDriver.get_object_cdn_url(self, obj, ex_expiry=ex_expiry, ex_method=ex_method)

libcloud/test/storage/test_aurora.py

@@ -38,6 +38,13 @@ def test_get_object_cdn_url(self):
         with self.assertRaises(LibcloudError):
             self.driver.get_object_cdn_url(obj)
 
+    def test_get_object_cdn_url_put(self):
+        self.mock_response_klass.type = "get_object"
+        obj = self.driver.get_object(container_name="test2", object_name="test")
+
+        with self.assertRaises(LibcloudError):
+            self.driver.get_object_cdn_url(obj)
+
 
 if __name__ == "__main__":
     sys.exit(unittest.main())

libcloud/test/storage/test_azure_blobs.py

@@ -511,6 +511,15 @@ def test_get_object_cdn_url(self):
         self.assertEqual(len(query["sig"]), 1)
         self.assertGreater(len(query["sig"][0]), 0)
 
+    def test_get_object_cdn_url_put(self):
+        obj = self.driver.get_object(container_name="test_container200", object_name="test")
+
+        url = urlparse.urlparse(self.driver.get_object_cdn_url(obj, ex_method="PUT"))
+        query = urlparse.parse_qs(url.query)
+
+        self.assertEqual(len(query["sig"]), 1)
+        self.assertGreater(len(query["sig"][0]), 0)
+
     def test_get_object_container_doesnt_exist(self):
         # This method makes two requests which makes mocking the response a bit
         # trickier

libcloud/test/storage/test_s3.py

@@ -530,6 +530,23 @@ def test_get_object_cdn_url(self):
             with self.assertRaises(NotImplementedError):
                 self.driver.get_object_cdn_url(obj)
 
+    def test_get_object_cdn_url_put(self):
+        self.mock_response_klass.type = "get_object"
+        obj = self.driver.get_object(container_name="test2", object_name="test")
+
+        # cdn urls can only be generated using a V4 connection
+        if issubclass(self.driver.connectionCls, S3SignatureV4Connection):
+            cdn_url = self.driver.get_object_cdn_url(obj, ex_method="PUT", ex_expiry=12)
+            url = urlparse.urlparse(cdn_url)
+            query = urlparse.parse_qs(url.query)
+
+            self.assertEqual(len(query["X-Amz-Signature"]), 1)
+            self.assertGreater(len(query["X-Amz-Signature"][0]), 0)
+            self.assertEqual(query["X-Amz-Expires"], ["43200"])
+        else:
+            with self.assertRaises(NotImplementedError):
+                self.driver.get_object_cdn_url(obj)
+
     def test_get_object_container_doesnt_exist(self):
         # This method makes two requests which makes mocking the response a bit
         # trickier