Coverage for manila/tests/scheduler/filters/test_share_replication.py: 100%

1# All Rights Reserved. 

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""" 

16Tests for the ShareReplicationFilter. 

17""" 

18import ddt 

19 

20from oslo_context import context 

21 

22from manila.scheduler.filters import share_replication 

23from manila import test 

24from manila.tests.scheduler import fakes 

25 

26 

27@ddt.ddt 

28class ShareReplicationFilterTestCase(test.TestCase): 

29 """Test case for ShareReplicationFilter.""" 

30 

31 def setUp(self): 

32 super(ShareReplicationFilterTestCase, self).setUp() 

33 self.filter = share_replication.ShareReplicationFilter() 

34 self.debug_log = self.mock_object(share_replication.LOG, 'debug') 

35 

36 @staticmethod 

37 def _create_replica_request(replication_domain='kashyyyk', 

38 replication_type='dr', 

39 active_replica_host=fakes.FAKE_HOST_STRING_1, 

40 all_replica_hosts=fakes.FAKE_HOST_STRING_1, 

41 is_admin=False): 

42 ctxt = context.RequestContext('fake', 'fake', is_admin=is_admin) 

43 return { 

44 'context': ctxt, 

45 'request_spec': { 

46 'active_replica_host': active_replica_host, 

47 'all_replica_hosts': all_replica_hosts, 

48 }, 

49 'resource_type': { 

50 'extra_specs': { 

51 'replication_type': replication_type, 

52 }, 

53 }, 

54 'replication_domain': replication_domain, 

55 } 

56 

57 @ddt.data('tatooine', '') 

58 def test_share_replication_filter_fails_incompatible_domain(self, domain): 

59 request = self._create_replica_request() 

60 

61 host = fakes.FakeHostState('host1', 

62 { 

63 'replication_domain': domain, 

64 }) 

65 

66 self.assertFalse(self.filter.host_passes(host, request)) 

67 self.assertTrue(self.debug_log.called) 

68 

69 def test_share_replication_filter_fails_no_replication_domain(self): 

70 request = self._create_replica_request() 

71 

72 host = fakes.FakeHostState('host1', 

73 { 

74 'replication_domain': None, 

75 }) 

76 

77 self.assertFalse(self.filter.host_passes(host, request)) 

78 self.assertTrue(self.debug_log.called) 

79 

80 def test_share_replication_filter_fails_host_has_replicas(self): 

81 all_replica_hosts = ','.join(['host1', fakes.FAKE_HOST_STRING_1]) 

82 request = self._create_replica_request( 

83 all_replica_hosts=all_replica_hosts) 

84 

85 host = fakes.FakeHostState('host1', 

86 { 

87 'replication_domain': 'kashyyyk', 

88 }) 

89 self.assertFalse(self.filter.host_passes(host, request)) 

90 self.assertTrue(self.debug_log.called) 

91 

92 def test_share_replication_filter_passes_no_replication_type(self): 

93 request = self._create_replica_request(replication_type=None) 

94 

95 host = fakes.FakeHostState('host1', 

96 { 

97 'replication_domain': 'tatooine', 

98 }) 

99 

100 self.assertTrue(self.filter.host_passes(host, request)) 

101 

102 def test_share_replication_filter_passes_no_active_replica_host(self): 

103 request = self._create_replica_request(active_replica_host=None) 

104 

105 host = fakes.FakeHostState('host1', 

106 { 

107 'replication_domain': 'tatooine', 

108 }) 

109 

110 self.assertTrue(self.filter.host_passes(host, request)) 

111 

112 def test_share_replication_filter_passes_happy_day(self): 

113 all_replica_hosts = ','.join(['host1', fakes.FAKE_HOST_STRING_1]) 

114 request = self._create_replica_request( 

115 all_replica_hosts=all_replica_hosts) 

116 

117 host = fakes.FakeHostState('host2', 

118 { 

119 'replication_domain': 'kashyyyk', 

120 }) 

121 

122 self.assertTrue(self.filter.host_passes(host, request)) 

123 

124 def test_share_replication_filter_empty(self): 

125 request = {} 

126 

127 host = fakes.FakeHostState('host1', 

128 { 

129 'replication_domain': 'naboo', 

130 }) 

131 

132 self.assertTrue(self.filter.host_passes(host, request))