test_azure_2_sas.py 15 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327
  1. from azure_storage.methods import extract_account_name, sas_prep, write_sas
  2. from azure_storage.azure_sas import AzureContainerSAS, AzureSAS, cli, container_sas, file_sas, folder_sas
  3. from unittest.mock import patch
  4. import argparse
  5. import pathlib
  6. import pytest
  7. import os
  8. @pytest.fixture(name='variables', scope='module')
  9. def setup():
  10. class Variables:
  11. def __init__(self):
  12. # Extract the account name and connection string from the system keyring prior to running tests
  13. self.passphrase = 'AzureStorage'
  14. self.account_name = extract_account_name(passphrase=self.passphrase)
  15. self.container_name = '00000container'
  16. self.test_path = os.path.abspath(os.path.dirname(__file__))
  17. self.file_path = os.path.join(self.test_path, 'files')
  18. self.output_file = os.path.join(self.file_path, 'sas_urls.txt')
  19. return Variables()
  20. def delete_output_file(output_file):
  21. os.remove(output_file)
  22. assert not os.path.isfile(output_file)
  23. def read_contents(output_file):
  24. contents = open(output_file, 'r').readlines()
  25. return contents
  26. def test_sas_prep(variables):
  27. variables.container_name, variables.connection_string, variables.account_key, \
  28. variables.blob_service_client, variables.container_client = \
  29. sas_prep(container_name=variables.container_name,
  30. passphrase=variables.passphrase,
  31. account_name=variables.account_name)
  32. assert variables.connection_string.startswith('DefaultEndpointsProtocol')
  33. @pytest.mark.parametrize('file_name',
  34. ['file_1.txt',
  35. 'nested_folder/nested_file_2.txt',
  36. 'nested_folder/nested_folder_2/nested_folder_test_1.txt'])
  37. def test_file_sas(variables, file_name):
  38. variables.sas_urls = AzureSAS.file_sas(container_client=variables.container_client,
  39. account_name=variables.account_name,
  40. container_name=variables.container_name,
  41. object_name=file_name,
  42. account_key=variables.account_key,
  43. expiry=10,
  44. sas_urls=dict())
  45. assert variables.sas_urls[os.path.basename(file_name)]\
  46. .startswith(f'https://{variables.account_name}.blob.core.windows.net/{variables.container_name}/{file_name}?')
  47. def test_file_sas_invalid_expiry(variables):
  48. with pytest.raises(SystemExit):
  49. sas_file = AzureSAS(object_name='file_1.txt',
  50. container_name=variables.container_name,
  51. output_file=variables.output_file,
  52. account_name=variables.account_name,
  53. passphrase=variables.passphrase,
  54. expiry=700,
  55. verbosity='info',
  56. category='file')
  57. sas_file.main()
  58. def test_file_sas_invalid_category(variables):
  59. with pytest.raises(SystemExit):
  60. sas_file = AzureSAS(object_name='file_1.txt',
  61. container_name=variables.container_name,
  62. output_file=variables.output_file,
  63. account_name=variables.account_name,
  64. passphrase=variables.passphrase,
  65. expiry=7,
  66. verbosity='info',
  67. category='container')
  68. sas_file.main()
  69. def test_file_sas_invalid_path(variables):
  70. output_file = os.path.join('/invalid', 'sas_urls.txt')
  71. if not os.environ.get('CIRCLECI'):
  72. with pytest.raises(SystemExit):
  73. sas_file = AzureSAS(object_name='file_1.txt',
  74. container_name=variables.container_name,
  75. output_file=output_file,
  76. account_name=variables.account_name,
  77. passphrase=variables.passphrase,
  78. expiry=7,
  79. verbosity='info',
  80. category='file')
  81. sas_file.main()
  82. def test_file_sas_tilde(variables):
  83. path_obj = pathlib.Path(variables.file_path)
  84. path = f'~{os.sep}{path_obj.relative_to(pathlib.Path.home())}'
  85. sas_file = AzureSAS(object_name='file_1.txt',
  86. container_name=variables.container_name,
  87. output_file=os.path.join(path, 'sas_urls.txt'),
  88. account_name=variables.account_name,
  89. passphrase=variables.passphrase,
  90. expiry=7,
  91. verbosity='info',
  92. category='file')
  93. sas_file.main()
  94. contents = read_contents(output_file=variables.output_file)
  95. assert contents[0] \
  96. .startswith(f'https://{variables.account_name}.blob.core.windows.net/{variables.container_name}/')
  97. delete_output_file(output_file=variables.output_file)
  98. @pytest.mark.parametrize('file_name',
  99. ['file_3.txt',
  100. 'nested_folder/nested_file_1.txt',
  101. 'nested_folder_3/nested_folder_2/nested_folder_test_1.txt'])
  102. def test_file_sas_invalid(variables, file_name):
  103. with pytest.raises(SystemExit):
  104. AzureSAS.file_sas(container_client=variables.container_client,
  105. account_name=variables.account_name,
  106. container_name=variables.container_name,
  107. object_name=file_name,
  108. account_key=variables.account_key,
  109. expiry=10,
  110. sas_urls=dict())
  111. def test_sas_urls_output_exists(variables):
  112. write_sas(output_file=variables.output_file,
  113. sas_urls=variables.sas_urls)
  114. assert os.path.isfile(variables.output_file)
  115. def test_sas_urls_output_contents(variables):
  116. contents = read_contents(output_file=variables.output_file)
  117. assert contents[0]\
  118. .startswith(f'https://{variables.account_name}.blob.core.windows.net/{variables.container_name}/')
  119. @pytest.mark.parametrize('output_file,expiry',
  120. [('sas_urls.txt', 0),
  121. ('sas_urls.txt', 500),
  122. ('', 1),
  123. ('folder/', 1)])
  124. @patch('argparse.ArgumentParser.parse_args')
  125. def test_file_sas_integration_invalid(mock_args, output_file, expiry, variables):
  126. file_name = 'file_1.txt'
  127. output_file = os.path.join(variables.file_path, output_file)
  128. with pytest.raises(SystemExit):
  129. mock_args.return_value = argparse.Namespace(passphrase=variables.passphrase,
  130. account_name=variables.account_name,
  131. container_name=variables.container_name,
  132. verbosity='info',
  133. file=file_name,
  134. output_file=output_file,
  135. expiry=expiry)
  136. arguments = cli()
  137. file_sas(args=arguments)
  138. @patch('argparse.ArgumentParser.parse_args')
  139. def test_file_sas_integration(mock_args, variables):
  140. delete_output_file(output_file=variables.output_file)
  141. file_name = 'file_1.txt'
  142. mock_args.return_value = argparse.Namespace(passphrase=variables.passphrase,
  143. account_name=variables.account_name,
  144. container_name=variables.container_name,
  145. verbosity='info',
  146. file=file_name,
  147. output_file=variables.output_file,
  148. expiry=1)
  149. arguments = cli()
  150. file_sas(args=arguments)
  151. contents = read_contents(output_file=variables.output_file)
  152. assert contents[0] \
  153. .startswith(f'https://{variables.account_name}.blob.core.windows.net/{variables.container_name}/')
  154. delete_output_file(output_file=variables.output_file)
  155. @pytest.mark.parametrize('folder_name,expected_dictionary_length',
  156. [('nested_folder', 3),
  157. ('nested_folder/nested_folder_2/', 1)])
  158. def test_folder_sas(variables, folder_name, expected_dictionary_length):
  159. variables.sas_urls = AzureSAS.folder_sas(container_client=variables.container_client,
  160. account_name=variables.account_name,
  161. container_name=variables.container_name,
  162. object_name=folder_name,
  163. account_key=variables.account_key,
  164. expiry=10,
  165. sas_urls=dict())
  166. assert len(variables.sas_urls) == expected_dictionary_length
  167. @pytest.mark.parametrize('folder_name',
  168. ['nested3',
  169. '',
  170. 'nested_folder/nested_folder_2/nested_folder_test_1.txt'
  171. 'nested_folder/nested_folder_1/',
  172. 'nested_folder/nested_folder_2/nested_folder_3'])
  173. def test_folder_sas_invalid(variables, folder_name):
  174. with pytest.raises(SystemExit):
  175. variables.sas_urls = AzureSAS.folder_sas(container_client=variables.container_client,
  176. account_name=variables.account_name,
  177. container_name=variables.container_name,
  178. object_name=folder_name,
  179. account_key=variables.account_key,
  180. expiry=10,
  181. sas_urls=dict())
  182. @patch('argparse.ArgumentParser.parse_args')
  183. def test_folder_sas_integration(mock_args, variables):
  184. folder_name = 'nested_folder'
  185. mock_args.return_value = argparse.Namespace(passphrase=variables.passphrase,
  186. account_name=variables.account_name,
  187. container_name=variables.container_name,
  188. verbosity='info',
  189. folder=folder_name,
  190. output_file=variables.output_file,
  191. expiry=1)
  192. arguments = cli()
  193. folder_sas(args=arguments)
  194. contents = read_contents(output_file=variables.output_file)
  195. assert contents[0] \
  196. .startswith(f'https://{variables.account_name}.blob.core.windows.net/{variables.container_name}/')
  197. def test_folder_sas_urls_integration_output_length(variables):
  198. contents = read_contents(output_file=variables.output_file)
  199. assert len(contents) == 3
  200. delete_output_file(output_file=variables.output_file)
  201. def test_container_sas(variables):
  202. variables.sas_urls = AzureContainerSAS.container_sas(container_client=variables.container_client,
  203. account_name=variables.account_name,
  204. container_name=variables.container_name,
  205. account_key=variables.account_key,
  206. expiry=1,
  207. sas_urls=dict())
  208. assert len(variables.sas_urls) == 10
  209. def test_container_sas_tilde(variables):
  210. path_obj = pathlib.Path(variables.file_path)
  211. path = f'~{os.sep}{path_obj.relative_to(pathlib.Path.home())}'
  212. sas = AzureContainerSAS(container_name=variables.container_name,
  213. output_file=os.path.join(path, 'sas_urls.txt'),
  214. account_name=variables.account_name,
  215. passphrase=variables.passphrase,
  216. expiry=1,
  217. verbosity='info')
  218. sas.main()
  219. contents = read_contents(output_file=variables.output_file)
  220. assert contents[0] \
  221. .startswith(f'https://{variables.account_name}.blob.core.windows.net/{variables.container_name}/')
  222. delete_output_file(output_file=variables.output_file)
  223. def test_container_invalid_name(variables):
  224. with pytest.raises(SystemExit):
  225. sas = AzureContainerSAS(container_name='000000000container',
  226. output_file=variables.output_file,
  227. account_name=variables.account_name,
  228. passphrase=variables.passphrase,
  229. expiry=1,
  230. verbosity='info')
  231. sas.main()
  232. def test_container_invalid_path(variables):
  233. output_file = os.path.join('/invalid', 'sas_urls.txt')
  234. if not os.environ.get('CIRCLECI'):
  235. with pytest.raises(SystemExit):
  236. sas = AzureContainerSAS(container_name=variables.container_name,
  237. output_file=output_file,
  238. account_name=variables.account_name,
  239. passphrase=variables.passphrase,
  240. expiry=1,
  241. verbosity='info')
  242. sas.main()
  243. def test_container_folder_provided(variables):
  244. with pytest.raises(SystemExit):
  245. sas = AzureContainerSAS(container_name=variables.container_name,
  246. output_file=variables.file_path,
  247. account_name=variables.account_name,
  248. passphrase=variables.passphrase,
  249. expiry=1,
  250. verbosity='info')
  251. sas.main()
  252. def test_container_invalid_expiry(variables):
  253. with pytest.raises(SystemExit):
  254. sas = AzureContainerSAS(container_name=variables.container_name,
  255. output_file=variables.output_file,
  256. account_name=variables.account_name,
  257. passphrase=variables.passphrase,
  258. expiry=500,
  259. verbosity='info')
  260. sas.main()
  261. @patch('argparse.ArgumentParser.parse_args')
  262. def test_container_sas_integration(mock_args, variables):
  263. mock_args.return_value = argparse.Namespace(passphrase=variables.passphrase,
  264. account_name=variables.account_name,
  265. container_name=variables.container_name,
  266. verbosity='info',
  267. output_file=variables.output_file,
  268. expiry=1)
  269. arguments = cli()
  270. container_sas(args=arguments)
  271. contents = read_contents(output_file=variables.output_file)
  272. assert len(contents) == 10
  273. def test_delete_output_file(variables):
  274. delete_output_file(output_file=variables.output_file)