Coverage for mlos_bench/mlos_bench/tests/services/remote/mock/mock_fileshare_service.py: 100%

1# 

2# Copyright (c) Microsoft Corporation. 

3# Licensed under the MIT License. 

4# 

5""" 

6A collection Service functions for mocking file share ops. 

7""" 

8 

9import logging 

10from typing import Any, Callable, Dict, List, Optional, Tuple, Union 

11 

12from mlos_bench.services.base_service import Service 

13from mlos_bench.services.base_fileshare import FileShareService 

14from mlos_bench.services.types.fileshare_type import SupportsFileShareOps 

15 

16_LOG = logging.getLogger(__name__) 

17 

18 

19class MockFileShareService(FileShareService, SupportsFileShareOps): 

20 """ 

21 A collection Service functions for mocking file share ops. 

22 """ 

23 

24 def __init__(self, config: Optional[Dict[str, Any]] = None, 

25 global_config: Optional[Dict[str, Any]] = None, 

26 parent: Optional[Service] = None, 

27 methods: Union[Dict[str, Callable], List[Callable], None] = None): 

28 super().__init__( 

29 config, global_config, parent, 

30 self.merge_methods(methods, [self.upload, self.download]) 

31 ) 

32 self._upload: List[Tuple[str, str]] = [] 

33 self._download: List[Tuple[str, str]] = [] 

34 

35 def upload(self, params: dict, local_path: str, remote_path: str, recursive: bool = True) -> None: 

36 self._upload.append((local_path, remote_path)) 

37 

38 def download(self, params: dict, remote_path: str, local_path: str, recursive: bool = True) -> None: 

39 self._download.append((remote_path, local_path)) 

40 

41 def get_upload(self) -> List[Tuple[str, str]]: 

42 """ 

43 Get the list of files that were uploaded. 

44 """ 

45 return self._upload 

46 

47 def get_download(self) -> List[Tuple[str, str]]: 

48 """ 

49 Get the list of files that were downloaded. 

50 """ 

51 return self._download