Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 10 additions & 2 deletions libcloud/storage/drivers/azure_blobs.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,15 @@
from typing import Literal
from datetime import datetime, timedelta

from libcloud.utils.py3 import ET, b, httplib, tostring, urlquote, urlencode
from libcloud.utils.py3 import (
ET,
b,
httplib,
tostring,
urlquote,
urlencode,
urlunquote,
)
from libcloud.utils.xml import fixxpath
from libcloud.utils.files import read_in_chunks
from libcloud.common.azure import AzureConnection, AzureActiveDirectoryConnection
Expand Down Expand Up @@ -654,7 +662,7 @@ def get_object_cdn_url(
params["sp"],
params["st"],
params["se"],
"/blob/{}{}".format(self.key, object_path),
"/blob/{}{}".format(self.key, urlunquote(object_path)),
"", # signedIdentifier
"", # signedIP
params["spr"],
Expand Down
14 changes: 14 additions & 0 deletions libcloud/test/storage/test_azure_blobs.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
import json
import tempfile
from io import BytesIO
from unittest.mock import patch

from libcloud.test import generate_random_data # pylint: disable-msg=E0611
from libcloud.test import unittest
Expand Down Expand Up @@ -520,6 +521,19 @@ def test_get_object_cdn_url_put(self):
self.assertEqual(len(query["sig"]), 1)
self.assertGreater(len(query["sig"][0]), 0)

@patch("libcloud.storage.drivers.azure_blobs.hmac.new")
def test_get_object_cdn_url_with_spaces(self, mock_hmac_new):
mock_hmac_new.return_value.digest.return_value = b"signature"
container = Container(name="test_container200", extra={}, driver=self.driver)
obj = Object("file name.txt", 0, None, {}, {}, container, self.driver)

url = self.driver.get_object_cdn_url(obj)
string_to_sign = mock_hmac_new.call_args.args[1].decode("utf-8")
canonical_resource = "/blob/{}/test_container200/file name.txt".format(self.driver.key)

self.assertIn(canonical_resource, string_to_sign)
self.assertIn("file%20name.txt", url)

def test_get_object_container_doesnt_exist(self):
# This method makes two requests which makes mocking the response a bit
# trickier
Expand Down