Coverage for watcher/common/metal_helper/base.py: 100%

1# Copyright 2023 Cloudbase Solutions 

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 

16import abc 

17 

18from watcher.common import exception 

19from watcher.common.metal_helper import constants as metal_constants 

20 

21 

22class BaseMetalNode(abc.ABC): 

23 hv_up_when_powered_off = False 

24 

25 def __init__(self, nova_node=None): 

26 self._nova_node = nova_node 

27 

28 def get_hypervisor_node(self): 

29 if not self._nova_node: 

30 raise exception.Invalid(message="No associated hypervisor.") 

31 return self._nova_node 

32 

33 def get_hypervisor_hostname(self): 

34 return self.get_hypervisor_node().hypervisor_hostname 

35 

36 @abc.abstractmethod 

37 def get_power_state(self): 

38 # TODO(lpetrut): document the following methods 

39 pass 

40 

41 @abc.abstractmethod 

42 def get_id(self): 

43 """Return the node id provided by the bare metal service.""" 

44 pass 

45 

46 @abc.abstractmethod 

47 def power_on(self): 

48 pass 

49 

50 @abc.abstractmethod 

51 def power_off(self): 

52 pass 

53 

54 def set_power_state(self, state): 

55 state = metal_constants.PowerState(state) 

56 if state == metal_constants.PowerState.ON: 

57 self.power_on() 

58 elif state == metal_constants.PowerState.OFF: 

59 self.power_off() 

60 else: 

61 raise exception.UnsupportedActionType( 

62 "Cannot set power state: %s" % state) 

63 

64 

65class BaseMetalHelper(abc.ABC): 

66 def __init__(self, osc): 

67 self._osc = osc 

68 

69 @property 

70 def nova_client(self): 

71 if not getattr(self, "_nova_client", None): 

72 self._nova_client = self._osc.nova() 

73 return self._nova_client 

74 

75 @abc.abstractmethod 

76 def list_compute_nodes(self): 

77 pass 

78 

79 @abc.abstractmethod 

80 def get_node(self, node_id): 

81 pass