# coding: utf-8 # ------------------------------------------------------------------------- # Copyright (c) Microsoft Corporation. All rights reserved. # Licensed under the MIT License. See License.txt in the project root for # license information. # -------------------------------------------------------------------------- import unittest from azure.common import ( AzureConflictHttpError, AzureMissingResourceHttpError, ) from azure.storage.file import ( FileService, DeleteSnapshot, ) from tests.testcase import ( StorageTestCase, record, LogCaptured, ) # ------------------------------------------------------------------------------ class StorageDirectoryTest(StorageTestCase): def setUp(self): super(StorageDirectoryTest, self).setUp() self.fs = self._create_storage_service(FileService, self.settings) self.share_name = self.get_resource_name('utshare') if not self.is_playback(): self.fs.create_share(self.share_name) def tearDown(self): if not self.is_playback(): try: self.fs.delete_share(self.share_name, delete_snapshots=DeleteSnapshot.Include) except: pass return super(StorageDirectoryTest, self).tearDown() # --Helpers----------------------------------------------------------------- # --Test cases for directories ---------------------------------------------- @record def test_create_directories(self): # Arrange # Act created = self.fs.create_directory(self.share_name, 'dir1') # Assert self.assertTrue(created) @record def test_create_directories_with_metadata(self): # Arrange metadata = {'hello': 'world', 'number': '42'} # Act self.fs.create_directory(self.share_name, 'dir1', metadata=metadata) # Assert md = self.fs.get_directory_metadata(self.share_name, 'dir1') self.assertDictEqual(md, metadata) @record def test_create_directories_fail_on_exist(self): # Arrange # Act created = self.fs.create_directory(self.share_name, 'dir1') with self.assertRaises(AzureConflictHttpError): self.fs.create_directory(self.share_name, 'dir1', fail_on_exist=True) # Assert self.assertTrue(created) @record def test_create_directory_with_already_existing_directory(self): # Arrange # Act created1 = self.fs.create_directory(self.share_name, 'dir1') created2 = self.fs.create_directory(self.share_name, 'dir1') # Assert self.assertTrue(created1) self.assertFalse(created2) @record def test_get_directory_properties(self): # Arrange self.fs.create_directory(self.share_name, 'dir1') # Act props = self.fs.get_directory_properties(self.share_name, 'dir1') # Assert self.assertIsNotNone(props) self.assertIsNotNone(props.properties.etag) self.assertIsNotNone(props.properties.last_modified) @record def test_get_directory_properties_with_snapshot(self): # Arrange metadata = {"test1": "foo", "test2": "bar"} self.fs.create_directory(self.share_name, 'dir1', metadata) snapshot1 = self.fs.snapshot_share(self.share_name) metadata2 = {"test100": "foo100", "test200": "bar200"} self.fs.set_directory_metadata(self.share_name, 'dir1', metadata2) # Act props = self.fs.get_directory_properties(self.share_name, 'dir1', snapshot=snapshot1.snapshot) # Assert self.assertIsNotNone(props) self.assertIsNotNone(props.properties.etag) self.assertIsNotNone(props.properties.last_modified) self.assertDictEqual(metadata, props.metadata) @record def test_get_directory_metadata_with_snapshot(self): # Arrange metadata = {"test1": "foo", "test2": "bar"} self.fs.create_directory(self.share_name, 'dir1', metadata) snapshot1 = self.fs.snapshot_share(self.share_name) metadata2 = {"test100": "foo100", "test200": "bar200"} self.fs.set_directory_metadata(self.share_name, 'dir1', metadata2) # Act snapshot_metadata = self.fs.get_directory_metadata(self.share_name, 'dir1', snapshot=snapshot1.snapshot) # Assert self.assertIsNotNone(snapshot_metadata) self.assertDictEqual(metadata, snapshot_metadata) @record def test_get_directory_properties_with_non_existing_directory(self): # Arrange # Act with self.assertRaises(AzureMissingResourceHttpError): self.fs.get_directory_properties(self.share_name, 'dir1') # Assert @record def test_directory_exists(self): # Arrange self.fs.create_directory(self.share_name, 'dir1') # Act exists = self.fs.exists(self.share_name, 'dir1') # Assert self.assertTrue(exists) @record def test_directory_not_exists(self): # Arrange # Act with LogCaptured(self) as log_captured: exists = self.fs.exists(self.share_name, 'missing') log_as_str = log_captured.getvalue() self.assertTrue('ERROR' not in log_as_str) # Assert self.assertFalse(exists) @record def test_directory_parent_not_exists(self): # Arrange # Act with LogCaptured(self) as log_captured: exists = self.fs.exists(self.share_name, 'missing1/missing2') log_as_str = log_captured.getvalue() self.assertTrue('ERROR' not in log_as_str) # Assert self.assertFalse(exists) @record def test_directory_exists_with_snapshot(self): # Arrange self.fs.create_directory(self.share_name, 'dir1') snapshot = self.fs.snapshot_share(self.share_name) self.fs.delete_directory(self.share_name, 'dir1') # Act exists = self.fs.exists(self.share_name, 'dir1', snapshot=snapshot.snapshot) # Assert self.assertTrue(exists) @record def test_directory_not_exists_with_snapshot(self): # Arrange snapshot = self.fs.snapshot_share(self.share_name) self.fs.create_directory(self.share_name, 'dir1') # Act exists = self.fs.exists(self.share_name, 'dir1', snapshot=snapshot.snapshot) # Assert self.assertFalse(exists) @record def test_get_set_directory_metadata(self): # Arrange metadata = {'hello': 'world', 'number': '43'} self.fs.create_directory(self.share_name, 'dir1') # Act self.fs.set_directory_metadata(self.share_name, 'dir1', metadata) md = self.fs.get_directory_metadata(self.share_name, 'dir1') # Assert self.assertDictEqual(md, metadata) @record def test_delete_directory_with_existing_share(self): # Arrange self.fs.create_directory(self.share_name, 'dir1') # Act deleted = self.fs.delete_directory(self.share_name, 'dir1') # Assert self.assertTrue(deleted) with self.assertRaises(AzureMissingResourceHttpError): self.fs.get_directory_properties(self.share_name, 'dir1') @record def test_delete_directory_with_existing_directory_fail_not_exist(self): # Arrange self.fs.create_directory(self.share_name, 'dir1') # Act deleted = self.fs.delete_directory(self.share_name, 'dir1') # Assert self.assertTrue(deleted) with self.assertRaises(AzureMissingResourceHttpError): self.fs.get_directory_properties(self.share_name, 'dir1') @record def test_delete_directory_with_non_existing_directory(self): # Arrange # Act with LogCaptured(self) as log_captured: deleted = self.fs.delete_directory(self.share_name, 'dir1', False) log_as_str = log_captured.getvalue() self.assertTrue('ERROR' not in log_as_str) # Assert self.assertFalse(deleted) @record def test_delete_directory_with_non_existing_directory_fail_not_exist(self): # Arrange # Act with LogCaptured(self) as log_captured: with self.assertRaises(AzureMissingResourceHttpError): self.fs.delete_directory(self.share_name, 'dir1', True) log_as_str = log_captured.getvalue() self.assertTrue('ERROR' in log_as_str) # Assert @record def test_get_directory_properties_server_encryption(self): # Arrange self.fs.create_directory(self.share_name, 'dir1') # Act props = self.fs.get_directory_properties(self.share_name, 'dir1') # Assert self.assertIsNotNone(props) self.assertIsNotNone(props.properties.etag) self.assertIsNotNone(props.properties.last_modified) if self.is_file_encryption_enabled(): self.assertTrue(props.properties.server_encrypted) else: self.assertFalse(props.properties.server_encrypted) # ------------------------------------------------------------------------------ if __name__ == '__main__': unittest.main()