Coverage for manila/tests/fake_utils.py: 92%

1# Copyright (c) 2011 Citrix Systems, Inc. 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"); you may 

4# not use this file except in compliance with the License. You may obtain 

5# a copy of the License at 

6# 

7# http://www.apache.org/licenses/LICENSE-2.0 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 

11# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 

12# License for the specific language governing permissions and limitations 

13# under the License. 

14 

15"""This modules stubs out functions in manila.utils.""" 

16 

17import re 

18from unittest import mock 

19 

20from eventlet import greenthread 

21from oslo_log import log 

22 

23from manila import exception 

24from manila import utils 

25 

26LOG = log.getLogger(__name__) 

27 

28_fake_execute_repliers = [] 

29_fake_execute_log = [] 

30 

31 

32def fake_execute_get_log(): 

33 return _fake_execute_log 

34 

35 

36def fake_execute_clear_log(): 

37 global _fake_execute_log 

38 _fake_execute_log = [] 

39 

40 

41def fake_execute_set_repliers(repliers): 

42 """Allows the client to configure replies to commands.""" 

43 global _fake_execute_repliers 

44 _fake_execute_repliers = repliers 

45 

46 

47def fake_execute_default_reply_handler(*ignore_args, **ignore_kwargs): 

48 """A reply handler for commands that haven't been added to the reply list. 

49 

50 Returns empty strings for stdout and stderr. 

51 

52 """ 

53 return '', '' 

54 

55 

56def fake_execute(*cmd_parts, **kwargs): 

57 """This function stubs out execute. 

58 

59 It optionally executes a preconfigued function to return expected data. 

60 

61 """ 

62 process_input = kwargs.get('process_input', None) 

63 check_exit_code = kwargs.get('check_exit_code', 0) 

64 delay_on_retry = kwargs.get('delay_on_retry', True) 

65 attempts = kwargs.get('attempts', 1) 

66 run_as_root = kwargs.get('run_as_root', False) 

67 cmd_str = ' '.join(str(part) for part in cmd_parts) 

68 

69 LOG.debug("Faking execution of cmd (subprocess): %s", cmd_str) 

70 _fake_execute_log.append(cmd_str) 

71 

72 reply_handler = fake_execute_default_reply_handler 

73 

74 for fake_replier in _fake_execute_repliers: 

75 if re.match(fake_replier[0], cmd_str): 75 ↛ 74line 75 didn't jump to line 74 because the condition on line 75 was always true

76 reply_handler = fake_replier[1] 

77 LOG.debug('Faked command matched %s', fake_replier[0]) 

78 break 

79 

80 if isinstance(reply_handler, str): 80 ↛ 82line 80 didn't jump to line 82 because the condition on line 80 was never true

81 # If the reply handler is a string, return it as stdout 

82 reply = reply_handler, '' 

83 else: 

84 try: 

85 # Alternative is a function, so call it 

86 reply = reply_handler(cmd_parts, 

87 process_input=process_input, 

88 delay_on_retry=delay_on_retry, 

89 attempts=attempts, 

90 run_as_root=run_as_root, 

91 check_exit_code=check_exit_code) 

92 except exception.ProcessExecutionError as e: 

93 LOG.debug('Faked command raised an exception %s', e) 

94 raise 

95 

96 stdout = reply[0] 

97 stderr = reply[1] 

98 LOG.debug("Reply to faked command is stdout='%(stdout)s' " 

99 "stderr='%(stderr)s'.", {"stdout": stdout, "stderr": stderr}) 

100 

101 # Replicate the sleep call in the real function 

102 greenthread.sleep(0) 

103 return reply 

104 

105 

106def stub_out_utils_execute(testcase): 

107 fake_execute_set_repliers([]) 

108 fake_execute_clear_log() 

109 testcase.mock_object(utils, 'execute', fake_execute) 

110 

111 

112def get_fake_lock_context(): 

113 context_manager_mock = mock.Mock() 

114 setattr(context_manager_mock, '__enter__', mock.Mock()) 

115 setattr(context_manager_mock, '__exit__', mock.Mock()) 

116 return context_manager_mock