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"""A collection Service functions for mocking file share ops.""" 

6 

7import logging 

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

9 

10from mlos_bench.services.base_fileshare import FileShareService 

11from mlos_bench.services.base_service import Service 

12from mlos_bench.services.types.fileshare_type import SupportsFileShareOps 

13 

14_LOG = logging.getLogger(__name__) 

15 

16 

17class MockFileShareService(FileShareService, SupportsFileShareOps): 

18 """A collection Service functions for mocking file share ops.""" 

19 

20 def __init__( 

21 self, 

22 config: Optional[Dict[str, Any]] = None, 

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

24 parent: Optional[Service] = None, 

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

26 ): 

27 super().__init__( 

28 config, 

29 global_config, 

30 parent, 

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

32 ) 

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

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

35 

36 def upload( 

37 self, 

38 params: dict, 

39 local_path: str, 

40 remote_path: str, 

41 recursive: bool = True, 

42 ) -> None: 

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

44 

45 def download( 

46 self, 

47 params: dict, 

48 remote_path: str, 

49 local_path: str, 

50 recursive: bool = True, 

51 ) -> None: 

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

53 

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

55 """Get the list of files that were uploaded.""" 

56 return self._upload 

57 

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

59 """Get the list of files that were downloaded.""" 

60 return self._download