Coverage for manila/tests/integrated/integrated_helpers.py: 54%

1# Copyright 2011 Justin Santa Barbara 

2# All Rights Reserved. 

3# 

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

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

6# a copy of the License at 

7# 

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

9# 

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

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

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

13# License for the specific language governing permissions and limitations 

14# under the License. 

15 

16""" 

17Provides common functionality for integrated unit tests 

18""" 

19 

20import random 

21import string 

22 

23from oslo_log import log 

24 

25from manila import service 

26from manila import test # For the flags 

27from manila.tests.integrated.api import client 

28 

29from oslo_config import cfg 

30from oslo_utils import uuidutils 

31 

32CONF = cfg.CONF 

33LOG = log.getLogger(__name__) 

34 

35 

36def generate_random_alphanumeric(length): 

37 """Creates a random alphanumeric string of specified length.""" 

38 return ''.join(random.choice(string.ascii_uppercase + string.digits) 

39 for _x in range(length)) 

40 

41 

42def generate_random_numeric(length): 

43 """Creates a random numeric string of specified length.""" 

44 return ''.join(random.choice(string.digits) 

45 for _x in range(length)) 

46 

47 

48def generate_new_element(items, prefix, numeric=False): 

49 """Creates a random string with prefix, that is not in 'items' list.""" 

50 while True: 

51 if numeric: 

52 candidate = prefix + generate_random_numeric(8) 

53 else: 

54 candidate = prefix + generate_random_alphanumeric(8) 

55 if candidate not in items: 

56 return candidate 

57 LOG.debug("Random collision on %s.", candidate) 

58 

59 

60class _IntegratedTestBase(test.TestCase): 

61 def setUp(self): 

62 super(_IntegratedTestBase, self).setUp() 

63 

64 f = self._get_flags() 

65 self.flags(**f) 

66 

67 # set up services 

68 self.share = self.start_service('share') 

69 self.scheduler = self.start_service('scheduler') 

70 

71 self._start_api_service() 

72 

73 self.api = client.TestOpenStackClient('fake', 'fake', self.endpoint) 

74 

75 def tearDown(self): 

76 self.osapi.stop() 

77 super(_IntegratedTestBase, self).tearDown() 

78 

79 def _start_api_service(self): 

80 self.osapi = service.WSGIService("osapi_share") 

81 self.osapi.start() 

82 self.endpoint = 'http://%s:%s/v2' % (self.osapi.host, 

83 self.osapi.port) 

84 LOG.info("Manila API started at %s", self.endpoint) 

85 

86 def _get_flags(self): 

87 """An opportunity to setup flags, before the services are started.""" 

88 f = {} 

89 

90 # Ensure tests only listen on localhost 

91 f['osapi_share_listen'] = '127.0.0.1' 

92 

93 # Auto-assign ports to allow concurrent tests 

94 f['osapi_share_listen_port'] = 0 

95 

96 return f 

97 

98 def get_unused_server_name(self): 

99 servers = self.api.get_servers() 

100 server_names = [server['name'] for server in servers] 

101 return generate_new_element(server_names, 'server') 

102 

103 def get_invalid_image(self): 

104 return uuidutils.generate_uuid() 

105 

106 def _build_minimal_create_server_request(self): 

107 server = {} 

108 

109 image = self.api.get_images()[0] 

110 LOG.debug("Image: %s.", image) 

111 

112 if 'imageRef' in image: 

113 image_href = image['imageRef'] 

114 else: 

115 image_href = image['id'] 

116 image_href = 'http://fake.server/%s' % image_href 

117 

118 # We now have a valid imageId 

119 server['imageRef'] = image_href 

120 

121 # Set a valid flavorId 

122 flavor = self.api.get_flavors()[0] 

123 LOG.debug("Using flavor: %s.", flavor) 

124 server['flavorRef'] = 'http://fake.server/%s' % flavor['id'] 

125 

126 # Set a valid server name 

127 server_name = self.get_unused_server_name() 

128 server['name'] = server_name 

129 return server