import unittest try: import azure except ImportError: raise unittest.SkipTest("azure package not installed, skipping tests") import datetime from datetime import timedelta from unittest import mock import django from azure.storage.blob import BlobProperties from django.core.exceptions import SuspiciousOperation from django.core.files.base import ContentFile from django.test import TestCase from django.test import override_settings from django.utils.timezone import make_aware from storages.backends import azure_storage class AzureStorageTest(TestCase): def setUp(self, *args): self.storage = azure_storage.AzureStorage() self.storage._client = mock.MagicMock() self.storage.overwrite_files = True self.account_name = "test" self.account_key = "key" self.container_name = "test" self.storage.azure_account = self.account_name self.storage.account_key = self.account_key self.storage.azure_container = self.container_name def test_get_valid_path(self): self.assertEqual( self.storage._get_valid_path("path/to/somewhere"), "path/to/somewhere" ) self.assertEqual( self.storage._get_valid_path("path/to/../somewhere"), "path/somewhere" ) self.assertEqual(self.storage._get_valid_path("path/to/../"), "path") self.assertEqual(self.storage._get_valid_path("path\\to\\..\\"), "path") self.assertEqual(self.storage._get_valid_path("path/name/"), "path/name") self.assertEqual( self.storage._get_valid_path("path\\to\\somewhere"), "path/to/somewhere" ) self.assertEqual(self.storage._get_valid_path("some/$/path"), "some/$/path") self.assertEqual(self.storage._get_valid_path("/$/path"), "$/path") self.assertEqual(self.storage._get_valid_path("path/$/"), "path/$") self.assertEqual( self.storage._get_valid_path("path/$/$/$/path"), "path/$/$/$/path" ) self.assertEqual(self.storage._get_valid_path("some///path"), "some/path") self.assertEqual(self.storage._get_valid_path("some//path"), "some/path") self.assertEqual(self.storage._get_valid_path("some\\\\path"), "some/path") self.assertEqual(self.storage._get_valid_path("a" * 1024), "a" * 1024) self.assertEqual(self.storage._get_valid_path("a/a" * 256), "a/a" * 256) self.assertRaises(ValueError, self.storage._get_valid_path, "") self.assertRaises(ValueError, self.storage._get_valid_path, "/") self.assertRaises(ValueError, self.storage._get_valid_path, "/../") self.assertRaises(ValueError, self.storage._get_valid_path, "..") self.assertRaises(ValueError, self.storage._get_valid_path, "///") self.assertRaises(ValueError, self.storage._get_valid_path, "a" * 1025) self.assertRaises(ValueError, self.storage._get_valid_path, "a/a" * 257) def test_get_valid_path_idempotency(self): self.assertEqual(self.storage._get_valid_path("//$//a//$//"), "$/a/$") self.assertEqual( self.storage._get_valid_path(self.storage._get_valid_path("//$//a//$//")), self.storage._get_valid_path("//$//a//$//"), ) some_path = "some path/some long name & then some.txt" self.assertEqual(self.storage._get_valid_path(some_path), some_path) self.assertEqual( self.storage._get_valid_path(self.storage._get_valid_path(some_path)), self.storage._get_valid_path(some_path), ) def test_get_available_name(self): self.storage.overwrite_files = False client_mock = mock.MagicMock() client_mock.exists.side_effect = [True, False] self.storage._client.get_blob_client.return_value = client_mock name = self.storage.get_available_name("foo.txt") self.assertTrue(name.startswith("foo_")) self.assertTrue(name.endswith(".txt")) self.assertTrue(len(name) > len("foo.txt")) self.assertEqual(client_mock.exists.call_count, 2) def test_get_available_name_first(self): self.storage.overwrite_files = False client_mock = mock.MagicMock() client_mock.exists.return_value = False self.storage._client.get_blob_client.return_value = client_mock self.assertEqual( self.storage.get_available_name("foo bar baz.txt"), "foo bar baz.txt" ) self.assertEqual(client_mock.exists.call_count, 1) def test_get_available_name_max_len(self): self.storage.overwrite_files = False # if you wonder why this is, file-system # storage will raise when file name is too long as well, # the form should validate this client_mock = mock.MagicMock() client_mock.exists.side_effect = [True, False] self.storage._client.get_blob_client.return_value = client_mock self.assertRaises(ValueError, self.storage.get_available_name, "a" * 1025) name = self.storage.get_available_name( "a" * 1000, max_length=100 ) # max_len == 1024 self.assertEqual(len(name), 100) self.assertTrue("_" in name) self.assertEqual(client_mock.exists.call_count, 2) def test_get_available_invalid(self): self.storage.overwrite_files = False self.storage._client.exists.return_value = False if django.VERSION[:2] == (3, 0): # Django 2.2.21 added this security fix: # https://docs.djangoproject.com/en/3.2/releases/2.2.21/#cve-2021-31542-potential-directory-traversal-via-uploaded-files # It raises SuspiciousOperation before we get to our ValueError. # The fix wasn't applied to 3.0 (no longer in support), but was applied to # 3.1 & 3.2. self.assertRaises(ValueError, self.storage.get_available_name, "") self.assertRaises(ValueError, self.storage.get_available_name, "/") self.assertRaises(ValueError, self.storage.get_available_name, ".") self.assertRaises(ValueError, self.storage.get_available_name, "///") else: self.assertRaises(SuspiciousOperation, self.storage.get_available_name, "") self.assertRaises(SuspiciousOperation, self.storage.get_available_name, "/") self.assertRaises(SuspiciousOperation, self.storage.get_available_name, ".") self.assertRaises( SuspiciousOperation, self.storage.get_available_name, "///" ) self.assertRaises(ValueError, self.storage.get_available_name, "...") def test_url(self): blob_mock = mock.MagicMock() blob_mock.url = "https://ret_foo.blob.core.windows.net/test/some%20blob" self.storage._client.get_blob_client.return_value = blob_mock self.assertEqual(self.storage.url("some blob"), blob_mock.url) self.storage._client.get_blob_client.assert_called_once_with("some blob") def test_url_unsafe_chars(self): blob_mock = mock.MagicMock() blob_mock.url = "https://ret_foo.blob.core.windows.net/test/some%20blob" self.storage._client.get_blob_client.return_value = blob_mock self.assertEqual( self.storage.url("foo;?:@=&\"<>#%{}|^~[]`bar/~!*()'"), blob_mock.url ) self.storage._client.get_blob_client.assert_called_once_with( "foo;?:@=&\"<>#%{}|^~[]`bar/~!*()'" ) @mock.patch("storages.backends.azure_storage.generate_blob_sas") def test_url_expire(self, generate_blob_sas_mocked): generate_blob_sas_mocked.return_value = "foo_token" blob_mock = mock.MagicMock() blob_mock.url = "https://ret_foo.blob.core.windows.net/test/some%20blob" self.storage._client.get_blob_client.return_value = blob_mock self.storage.account_name = self.account_name 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), "https://ret_foo.blob.core.windows.net/test/some%20blob", ) generate_blob_sas_mocked.assert_called_once_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"]), "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") def test_url_custom_domain(self): self.storage.custom_domain = "foo_domain" blob_mock = mock.MagicMock() blob_mock.url = "https://ret_foo.blob.core.windows.net/test/foo_name" self.storage._client.get_blob_client.return_value = blob_mock url = self.storage.url("foo_name") self.assertEqual(url, "https://foo_domain/test/foo_name") @mock.patch("storages.backends.azure_storage.generate_blob_sas") def test_url_expire_user_delegation_key(self, generate_blob_sas_mocked): generate_blob_sas_mocked.return_value = "foo_token" blob_mock = mock.MagicMock() blob_mock.url = "https://ret_foo.blob.core.windows.net/test/some%20blob" self.storage._client.get_blob_client.return_value = blob_mock self.storage.account_name = self.account_name service_client = mock.MagicMock() self.storage._service_client = service_client self.storage.token_credential = "token_credential" fixed_time = make_aware( datetime.datetime(2016, 11, 6, 4), datetime.timezone.utc ) with mock.patch("storages.backends.azure_storage.datetime") as d_mocked: d_mocked.utcnow.return_value = fixed_time service_client.get_user_delegation_key.return_value = "user delegation key" self.assertEqual( self.storage.url("some blob", 100), "https://ret_foo.blob.core.windows.net/test/some%20blob", ) generate_blob_sas_mocked.assert_called_once_with( self.account_name, self.container_name, "some blob", account_key=self.account_key, user_delegation_key="user delegation key", permission=mock.ANY, expiry=fixed_time + timedelta(seconds=100), ) def test_container_client_default_params(self): storage = azure_storage.AzureStorage() storage.account_name = self.account_name with mock.patch( "storages.backends.azure_storage.BlobServiceClient", autospec=True ) as bsc_mocked: client_mock = mock.MagicMock() bsc_mocked.return_value.get_container_client.return_value = client_mock self.assertEqual(storage.client, client_mock) bsc_mocked.assert_called_once_with( "https://test.blob.core.windows.net", credential=None ) def test_container_client_params_account_key(self): storage = azure_storage.AzureStorage() storage.account_name = "foo_name" storage.custom_domain = "foo_domain" storage.account_key = "foo_key" with mock.patch( "storages.backends.azure_storage.BlobServiceClient", autospec=True ) as bsc_mocked: client_mock = mock.MagicMock() bsc_mocked.return_value.get_container_client.return_value = client_mock self.assertEqual(storage.client, client_mock) bsc_mocked.assert_called_once_with( "https://foo_name.blob.core.windows.net", credential={"account_name": "foo_name", "account_key": "foo_key"}, ) def test_container_client_params_sas_token(self): storage = azure_storage.AzureStorage() storage.account_name = "foo_name" storage.azure_ssl = False storage.sas_token = "foo_token" with mock.patch( "storages.backends.azure_storage.BlobServiceClient", autospec=True ) as bsc_mocked: client_mock = mock.MagicMock() bsc_mocked.return_value.get_container_client.return_value = client_mock self.assertEqual(storage.client, client_mock) bsc_mocked.assert_called_once_with( "http://foo_name.blob.core.windows.net", credential="foo_token" ) def test_container_client_params_token_credential(self): storage = azure_storage.AzureStorage() storage.account_name = self.account_name storage.token_credential = "foo_cred" with mock.patch( "storages.backends.azure_storage.BlobServiceClient", autospec=True ) as bsc_mocked: client_mock = mock.MagicMock() bsc_mocked.return_value.get_container_client.return_value = client_mock self.assertEqual(storage.client, client_mock) bsc_mocked.assert_called_once_with( "https://test.blob.core.windows.net", credential="foo_cred" ) def test_connection_string_can_have_missing(self): storage = azure_storage.AzureStorage( connection_string="AccountKey=abc;Foobar=xyz;" ) self.assertEqual(storage.account_key, "abc") self.assertIsNone(storage.account_name) def test_container_client_params_connection_string(self): storage = azure_storage.AzureStorage() storage.account_name = self.account_name storage.connection_string = "foo_conn" with mock.patch( "storages.backends.azure_storage.BlobServiceClient.from_connection_string", spec=azure_storage.BlobServiceClient.from_connection_string, ) as bsc_mocked: client_mock = mock.MagicMock() bsc_mocked.return_value.get_container_client.return_value = client_mock self.assertEqual(storage.client, client_mock) bsc_mocked.assert_called_once_with("foo_conn") # From boto3 def test_storage_save(self): """ Test saving a file """ name = "test storage save.txt" content = ContentFile("new content") with mock.patch("storages.backends.azure_storage.ContentSettings") as c_mocked: c_mocked.return_value = "content_settings_foo" self.assertEqual(self.storage.save(name, content), name) self.storage._client.upload_blob.assert_called_once_with( name, content.file, content_settings="content_settings_foo", max_concurrency=2, timeout=20, overwrite=True, ) c_mocked.assert_called_once_with( content_type="text/plain", content_encoding=None, cache_control=None ) def test_storage_open_write(self): """ Test opening a file in write mode """ name = "test_open_for_writïng.txt" content = "new content" file = self.storage.open(name, "w") file.write(content) written_file = file.file file.close() self.storage._client.upload_blob.assert_called_once_with( name, written_file, content_settings=mock.ANY, max_concurrency=2, timeout=20, overwrite=True, ) def test_storage_exists(self): blob_name = "blob" client_mock = mock.MagicMock() self.storage._client.get_blob_client.return_value = client_mock self.assertTrue(self.storage.exists(blob_name)) self.assertEqual(client_mock.exists.call_count, 1) def test_delete_blob(self): self.storage.delete("name") self.storage._client.delete_blob.assert_called_once_with("name", timeout=20) def test_storage_listdir_base(self): file_names = ["some/path/1.txt", "2.txt", "other/path/3.txt", "4.txt"] result = [] for p in file_names: obj = mock.MagicMock() obj.name = p result.append(obj) self.storage._client.list_blobs.return_value = iter(result) dirs, files = self.storage.listdir("") self.storage._client.list_blobs.assert_called_with( name_starts_with="", timeout=20 ) self.assertEqual(len(dirs), 0) self.assertEqual(len(files), 4) for filename in ["2.txt", "4.txt", "other/path/3.txt", "some/path/1.txt"]: self.assertTrue( filename in files, """ "{}" not in file list "{}".""".format(filename, files), ) def test_size_of_file(self): props = BlobProperties() props.size = 12 client_mock = mock.MagicMock() client_mock.get_blob_properties.return_value = props self.storage._client.get_blob_client.return_value = client_mock self.assertEqual(12, self.storage.size("name")) def test_override_settings(self): with override_settings(AZURE_CONTAINER="foo1"): storage = azure_storage.AzureStorage() self.assertEqual(storage.azure_container, "foo1") with override_settings(AZURE_CONTAINER="foo2"): storage = azure_storage.AzureStorage() self.assertEqual(storage.azure_container, "foo2") def test_override_class_variable(self): class MyStorage1(azure_storage.AzureStorage): azure_container = "foo1" storage = MyStorage1() self.assertEqual(storage.azure_container, "foo1") class MyStorage2(azure_storage.AzureStorage): azure_container = "foo2" storage = MyStorage2() self.assertEqual(storage.azure_container, "foo2") def test_override_init_argument(self): storage = azure_storage.AzureStorage(azure_container="foo1") self.assertEqual(storage.azure_container, "foo1") storage = azure_storage.AzureStorage(azure_container="foo2") self.assertEqual(storage.azure_container, "foo2") @mock.patch("storages.backends.azure_storage.BlobServiceClient", autospec=True) def test_client_settings(self, bsc): with override_settings(AZURE_CLIENT_OPTIONS={"api_version": "1.3"}): storage = azure_storage.AzureStorage(account_name="test") client_mock = mock.MagicMock() bsc.return_value.get_container_client.return_value = client_mock self.assertEqual(storage.client, client_mock) bsc.assert_called_once_with( "https://test.blob.core.windows.net", credential=None, api_version="1.3" )