ms/artifact-manager/tests/utils_test.py

   1 import os
   2 import shutil
   3 import zipfile
   4 from unittest.mock import patch
   5
   6 import manager.utils
   7 from manager.utils import FileRepository, Repository, RepositoryStrategy
   8
   9
  10 class MockZipFile(zipfile.ZipFile):
  11     def __init__(self, *args, **kwargs):
  12         pass
  13
  14     def extractall(self, path: str) -> None:
  15         pass
  16
  17     def write(self, *arg, **kwargs) -> None:
  18         pass
  19
  20
  21 def test_fetch_proper_repository():
  22     repo: Repository = RepositoryStrategy.get_reporitory()
  23     assert repo.__class__ is FileRepository
  24
  25
  26 def test_blueprint_upload():
  27     repo: Repository = RepositoryStrategy.get_reporitory()
  28     # fmt: off
  29     with patch.object(manager.utils, "is_zipfile", return_value=True) as mock_is_zip, \
  30             patch.object(os, "makedirs", return_value=None) as mock_mkdirs, \
  31             patch.object(manager.utils, 'ZipFile', return_value=MockZipFile()
  32         ):
  33         repo.upload_blueprint(b"abcd", "test_cba", "1.0.a")
  34         mock_is_zip.assert_called_once()
  35         mock_mkdirs.assert_called_once_with('/tmp/test_cba/1.0.a', mode=0o744)
  36     # fmt: on
  37
  38
  39 def test_blueprint_download():
  40     repo: Repository = RepositoryStrategy.get_reporitory()
  41     mock_path = [
  42         ("test_cba", ["1.0.a"], []),
  43         ("test_cba/1.0.a", [], ["file.txt"]),
  44     ]
  45     # fmt: off
  46     with patch.object(os, "walk", return_value=mock_path) as mock_walk, \
  47             patch.object(manager.utils, 'ZipFile', return_value=MockZipFile()), \
  48             patch.object(os.path, 'exists', return_value=True
  49         ):
  50         repo.download_blueprint("test_cba", "1.0.a")
  51         mock_walk.assert_called_once_with('/tmp/test_cba/1.0.a')
  52     # fmt: on
  53
  54
  55 def test_remove_blueprint():
  56     repo: Repository = RepositoryStrategy.get_reporitory()
  57     with patch.object(shutil, "rmtree", return_value=None) as mock_rmtree:
  58         repo.remove_blueprint("cba", "1.0a")
  59         mock_rmtree.assert_called_once()