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 collections.abc import Callable 

9from typing import Any 

10 

11from mlos_bench.services.base_fileshare import FileShareService 

12from mlos_bench.services.base_service import Service 

13from mlos_bench.services.types.fileshare_type import SupportsFileShareOps 

14 

15_LOG = logging.getLogger(__name__) 

16 

17 

18class MockFileShareService(FileShareService, SupportsFileShareOps): 

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

20 

21 def __init__( 

22 self, 

23 config: dict[str, Any] | None = None, 

24 global_config: dict[str, Any] | None = None, 

25 parent: Service | None = None, 

26 methods: dict[str, Callable] | list[Callable] | None = None, 

27 ): 

28 super().__init__( 

29 config, 

30 global_config, 

31 parent, 

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

33 ) 

34 self._upload: list[tuple[str, str]] = [] 

35 self._download: list[tuple[str, str]] = [] 

36 

37 def upload( 

38 self, 

39 params: dict, 

40 local_path: str, 

41 remote_path: str, 

42 recursive: bool = True, 

43 ) -> None: 

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

45 

46 def download( 

47 self, 

48 params: dict, 

49 remote_path: str, 

50 local_path: str, 

51 recursive: bool = True, 

52 ) -> None: 

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

54 

55 def get_upload(self) -> list[tuple[str, str]]: 

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

57 return self._upload 

58 

59 def get_download(self) -> list[tuple[str, str]]: 

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

61 return self._download