Coverage for custom_components/autoarm/hass_api.py: 93%

1import logging 

2from functools import partial 

3from typing import TYPE_CHECKING, Any, cast 

4 

5from homeassistant.components.alarm_control_panel.const import AlarmControlPanelState 

6from homeassistant.exceptions import ConditionError, ConditionErrorContainer 

7from homeassistant.helpers import condition as condition_helper 

8from homeassistant.helpers import issue_registry as ir 

9from homeassistant.helpers.template import Template 

10 

11from .const import DOMAIN, ConditionVariables 

12 

13if TYPE_CHECKING: 

14 from collections.abc import Callable 

15 

16 from homeassistant.core import Context, HomeAssistant 

17 from homeassistant.helpers.typing import ConfigType, TemplateVarsType 

18 

19 

20_LOGGER = logging.getLogger(__name__) 

21 

22 

23class HomeAssistantAPI: 

24 def __init__(self, hass: HomeAssistant | None = None) -> None: 

25 self._hass = hass 

26 

27 def raise_issue( 

28 self, 

29 issue_id: str, 

30 issue_key: str, 

31 issue_map: dict[str, str], 

32 severity: ir.IssueSeverity = ir.IssueSeverity.WARNING, 

33 learn_more_url: str = "https://autoarm.rhizomatics.org.uk", 

34 is_fixable: bool = False, 

35 ) -> None: 

36 if not self._hass: 

37 return 

38 ir.async_create_issue( 

39 self._hass, 

40 DOMAIN, 

41 issue_id, 

42 translation_key=issue_key, 

43 translation_placeholders=issue_map, 

44 severity=severity, 

45 learn_more_url=learn_more_url, 

46 is_fixable=is_fixable, 

47 ) 

48 

49 async def build_condition( 

50 self, condition_config: list[ConfigType], strict: bool = False, validate: bool = False, name: str = DOMAIN 

51 ) -> Callable[[TemplateVarsType], bool] | None: 

52 if self._hass is None: 

53 raise ValueError("HomeAssistant not available") 

54 capturing_logger: ConditionErrorLoggingAdaptor = ConditionErrorLoggingAdaptor(_LOGGER) 

55 condition_variables: ConditionVariables = ConditionVariables(None, None, False, AlarmControlPanelState.PENDING, {}) 

56 cond_list: list[ConfigType] 

57 try: 

58 if validate: 

59 cond_list = cast( 

60 "list[ConfigType]", await condition_helper.async_validate_conditions_config(self._hass, condition_config) 

61 ) 

62 else: 

63 cond_list = condition_config 

64 except Exception: 

65 _LOGGER.exception("AUTOARM Condition validation failed") 

66 raise 

67 try: 

68 if strict: 

69 force_strict_template_mode(cond_list, undo=False) 

70 

71 test: Callable[[TemplateVarsType], bool] = await condition_helper.async_conditions_from_config( 

72 self._hass, cond_list, cast("logging.Logger", capturing_logger), name 

73 ) 

74 if test is None: 

75 raise ValueError(f"Invalid condition {condition_config}") 

76 test({DOMAIN: condition_variables.as_dict()}) 

77 if strict and capturing_logger.condition_errors: 

78 for exception in capturing_logger.condition_errors: 

79 _LOGGER.warning("AUTOARM Invalid condition %s:%s", condition_config, exception) 

80 raise capturing_logger.condition_errors[0] 

81 return test 

82 except Exception: 

83 _LOGGER.exception("AUTOARM Condition eval failed") 

84 raise 

85 finally: 

86 if strict: 

87 force_strict_template_mode(condition_config, undo=True) 

88 

89 def evaluate_condition( 

90 self, 

91 checker: Callable[[TemplateVarsType], bool], 

92 condition_variables: ConditionVariables | None = None, 

93 ) -> bool | None: 

94 if self._hass is None: 

95 raise ValueError("HomeAssistant not available") 

96 try: 

97 return checker({DOMAIN: condition_variables.as_dict()} if condition_variables else None) 

98 except Exception as e: 

99 _LOGGER.error("AUTOARM Condition eval failed: %s", e) 

100 raise 

101 

102 def fire_event(self, event_name: str, event_data: dict[str, Any] | None = None, context: Context | None = None) -> None: 

103 if self._hass is not None: 

104 _LOGGER.debug("AUTOARM Firing %s event: %s", event_name, event_data) 

105 self._hass.bus.async_fire(f"{DOMAIN}_{event_name}", event_data, context=context) 

106 

107 

108class ConditionErrorLoggingAdaptor(logging.LoggerAdapter["logging.Logger"]): 

109 def __init__(self, *args: Any, **kwargs: Any) -> None: 

110 super().__init__(*args, **kwargs) 

111 self.condition_errors: list[ConditionError] = [] 

112 

113 def capture(self, args: Any) -> None: 

114 if args and isinstance(args, (list, tuple)): 

115 for arg in args: 

116 if isinstance(arg, ConditionErrorContainer): 

117 self.condition_errors.extend(arg.errors) 

118 elif isinstance(arg, ConditionError): 

119 self.condition_errors.append(arg) 

120 

121 def error(self, msg: Any, *args: object, **kwargs: Any) -> None: 

122 self.capture(args) 

123 self.logger.error(msg, args, kwargs) 

124 

125 def warning(self, msg: Any, *args: Any, **kwargs: Any) -> None: 

126 self.capture(args) 

127 self.logger.warning(msg, args, kwargs) 

128 

129 

130def force_strict_template_mode(conditions: list[ConfigType], undo: bool = False) -> None: 

131 class TemplateWrapper: 

132 def __init__(self, obj: Template) -> None: 

133 self._obj = obj 

134 

135 def __getattr__(self, name: str) -> Any: 

136 if name == "async_render_to_info": 

137 return partial(self._obj.async_render_to_info, strict=True) 

138 return getattr(self._obj, name) 

139 

140 def __setattr__(self, name: str, value: Any) -> None: 

141 super().__setattr__(name, value) 

142 

143 def wrap_template(cond: ConfigType, undo: bool) -> ConfigType: 

144 for key, val in cond.items(): 

145 if not undo and isinstance(val, Template) and hasattr(val, "_env"): 

146 cond[key] = TemplateWrapper(val) 

147 elif undo and isinstance(val, TemplateWrapper): 

148 cond[key] = val._obj 

149 elif isinstance(val, dict): 

150 wrap_template(val, undo) 

151 return cond 

152 

153 if conditions is not None: 

154 conditions = [wrap_template(condition, undo) for condition in conditions]