Coverage for manila/tests/scheduler/drivers/test_base.py: 100%

1# Copyright 2010 United States Government as represented by the 

2# Administrator of the National Aeronautics and Space Administration. 

3# All Rights Reserved. 

4# 

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

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

7# a copy of the License at 

8# 

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

10# 

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

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

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

14# License for the specific language governing permissions and limitations 

15# under the License. 

16""" 

17Tests For Base Scheduler 

18""" 

19 

20from unittest import mock 

21 

22from oslo_config import cfg 

23from oslo_utils import timeutils 

24 

25from manila import context 

26from manila import db 

27from manila.scheduler.drivers import base 

28from manila import test 

29from manila import utils 

30 

31CONF = cfg.CONF 

32 

33 

34class SchedulerTestCase(test.TestCase): 

35 """Test case for base scheduler driver class.""" 

36 

37 # So we can subclass this test and re-use tests if we need. 

38 driver_cls = base.Scheduler 

39 

40 def setUp(self): 

41 super(SchedulerTestCase, self).setUp() 

42 self.driver = self.driver_cls() 

43 self.context = context.RequestContext('fake_user', 'fake_project') 

44 self.topic = 'fake_topic' 

45 

46 def test_update_service_capabilities(self): 

47 service_name = 'fake_service' 

48 host = 'fake_host' 

49 capabilities = {'fake_capability': 'fake_value'} 

50 timestamp = 1111 

51 with mock.patch.object(self.driver.host_manager, 

52 'update_service_capabilities', mock.Mock()): 

53 self.driver.update_service_capabilities( 

54 service_name, host, capabilities, timestamp) 

55 (self.driver.host_manager.update_service_capabilities. 

56 assert_called_once_with(service_name, host, 

57 capabilities, timestamp)) 

58 

59 def test_hosts_up(self): 

60 service1 = {'host': 'host1'} 

61 service2 = {'host': 'host2'} 

62 services = [service1, service2] 

63 

64 def fake_service_is_up(*args, **kwargs): 

65 if args[0]['host'] == 'host1': 

66 return False 

67 return True 

68 

69 with mock.patch.object(db, 'service_get_all_by_topic', 

70 mock.Mock(return_value=services)): 

71 with mock.patch.object(utils, 'service_is_up', 

72 mock.Mock(side_effect=fake_service_is_up)): 

73 result = self.driver.hosts_up(self.context, self.topic) 

74 self.assertEqual(['host2'], result) 

75 db.service_get_all_by_topic.assert_called_once_with( 

76 self.context, self.topic) 

77 

78 

79class SchedulerDriverBaseTestCase(SchedulerTestCase): 

80 """Test cases for base scheduler driver class methods. 

81 

82 These can't fail if the driver is changed. 

83 """ 

84 

85 def test_unimplemented_schedule(self): 

86 fake_args = (1, 2, 3) 

87 fake_kwargs = {'cat': 'meow'} 

88 

89 self.assertRaises(NotImplementedError, self.driver.schedule, 

90 self.context, self.topic, 'schedule_something', 

91 *fake_args, **fake_kwargs) 

92 

93 

94class SchedulerDriverModuleTestCase(test.TestCase): 

95 """Test case for scheduler driver module methods.""" 

96 

97 def setUp(self): 

98 super(SchedulerDriverModuleTestCase, self).setUp() 

99 self.context = context.RequestContext('fake_user', 'fake_project') 

100 

101 @mock.patch.object(db, 'share_update', mock.Mock()) 

102 def test_share_host_update_db(self): 

103 with mock.patch.object(timeutils, 'utcnow', 

104 mock.Mock(return_value='fake-now')): 

105 base.share_update_db(self.context, 31337, 'fake_host') 

106 db.share_update.assert_called_once_with( 

107 self.context, 31337, 

108 {'host': 'fake_host', 'scheduled_at': 'fake-now'})