diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 6cdc4191..0b0085fb 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -15,11 +15,13 @@ Azure ----- - Fix ``collectstatic --clear`` (`#1403`_) +- Add ``mode`` kwarg to ``.url()`` to support creation of signed URLs for upload (`#1414`_) .. _#1399: https://github.com/jschneier/django-storages/pull/1399 .. _#1381: https://github.com/jschneier/django-storages/pull/1381 .. _#1402: https://github.com/jschneier/django-storages/pull/1402 .. _#1403: https://github.com/jschneier/django-storages/pull/1403 +.. _#1414: https://github.com/jschneier/django-storages/pull/1414 1.14.3 (2024-05-04) diff --git a/storages/backends/azure_storage.py b/storages/backends/azure_storage.py index 69b02137..8d8c5184 100644 --- a/storages/backends/azure_storage.py +++ b/storages/backends/azure_storage.py @@ -309,9 +309,10 @@ def _expire_at(self, expire): # azure expects time in UTC return datetime.utcnow() + timedelta(seconds=expire) - def url(self, name, expire=None, parameters=None): + def url(self, name, expire=None, parameters=None, mode="r"): name = self._get_valid_path(name) params = parameters or {} + permission = BlobSasPermissions.from_string(mode) if expire is None: expire = self.expiration_secs @@ -326,7 +327,7 @@ def url(self, name, expire=None, parameters=None): name, account_key=self.account_key, user_delegation_key=user_delegation_key, - permission=BlobSasPermissions(read=True), + permission=permission, expiry=expiry, **params, ) diff --git a/tests/test_azure.py b/tests/test_azure.py index cad33203..799605de 100644 --- a/tests/test_azure.py +++ b/tests/test_azure.py @@ -164,7 +164,9 @@ def test_url_expire(self, generate_blob_sas_mocked): fixed_time = make_aware( datetime.datetime(2016, 11, 6, 4), datetime.timezone.utc ) + with mock.patch("storages.backends.azure_storage.datetime") as d_mocked: + # Implicit read permission d_mocked.utcnow.return_value = fixed_time self.assertEqual( self.storage.url("some blob", 100), @@ -179,6 +181,26 @@ def test_url_expire(self, generate_blob_sas_mocked): permission=mock.ANY, expiry=fixed_time + timedelta(seconds=100), ) + called_args, called_kwargs = generate_blob_sas_mocked.call_args + self.assertEqual(str(called_kwargs["permission"]), "r") + + # Explicit write permission + d_mocked.utcnow.return_value = fixed_time + self.assertEqual( + self.storage.url("some blob", expire=100, mode="w"), + "https://ret_foo.blob.core.windows.net/test/some%20blob", + ) + generate_blob_sas_mocked.assert_called_with( + self.account_name, + self.container_name, + "some blob", + account_key=self.account_key, + user_delegation_key=None, + permission=mock.ANY, + expiry=fixed_time + timedelta(seconds=100), + ) + called_args, called_kwargs = generate_blob_sas_mocked.call_args + self.assertEqual(str(called_kwargs["permission"]), "w") @mock.patch("storages.backends.azure_storage.generate_blob_sas") def test_url_expire_user_delegation_key(self, generate_blob_sas_mocked):