Coverage for custom_components/remote_logger/config_flow.py: 100%

1"""Config flow for the remote_logger integration.""" 

2 

3from __future__ import annotations 

4 

5import logging 

6from typing import Any 

7 

8from homeassistant.config_entries import ConfigFlow, ConfigFlowResult 

9from homeassistant.helpers.aiohttp_client import async_get_clientsession 

10 

11from .const import ( 

12 BACKEND_OTEL, 

13 BACKEND_SYSLOG, 

14 CONF_BACKEND, 

15 CONF_ENCODING, 

16 CONF_HOST, 

17 CONF_PORT, 

18 CONF_PROTOCOL, 

19 CONF_RESOURCE_ATTRIBUTES, 

20 CONF_USE_TLS, 

21 DOMAIN, 

22) 

23from .otel.const import OTEL_DATA_SCHEMA, OTLP_LOGS_PATH 

24from .otel.exporter import parse_resource_attributes 

25from .otel.exporter import validate as otel_validate 

26from .syslog.const import SYSLOG_DATA_SCHEMA 

27from .syslog.exporter import validate as syslog_validate 

28 

29_LOGGER = logging.getLogger(__name__) 

30 

31 

32def _build_endpoint_url(host: str, port: int, use_tls: bool) -> str: 

33 """Build the full OTLP endpoint URL.""" 

34 scheme = "https" if use_tls else "http" 

35 return f"{scheme}://{host}:{port}{OTLP_LOGS_PATH}" 

36 

37 

38class OtelLogsConfigFlow(ConfigFlow, domain=DOMAIN): 

39 """Handle a config flow for OpenTelemetry Log Exporter.""" 

40 

41 VERSION = 2 

42 

43 async def async_step_user( 

44 self, 

45 user_input: dict[str, Any] | None = None, # noqa: ARG002 

46 ) -> ConfigFlowResult: 

47 """Show menu to choose backend type.""" 

48 return self.async_show_menu( 

49 step_id="user", 

50 menu_options=["otel", "syslog"], 

51 ) 

52 

53 async def async_step_otel(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult: 

54 """Handle OpenTelemetry OTLP configuration.""" 

55 errors: dict[str, str] = {} 

56 

57 if user_input is not None: 

58 host = user_input[CONF_HOST] 

59 port = user_input[CONF_PORT] 

60 use_tls = user_input[CONF_USE_TLS] 

61 url = _build_endpoint_url(host, port, use_tls) 

62 

63 # Validate connectivity 

64 session = async_get_clientsession(self.hass, verify_ssl=use_tls) 

65 errors = await otel_validate(session, url, user_input[CONF_ENCODING]) 

66 # Validate resource attributes format 

67 if not errors: 

68 raw_attrs = user_input.get(CONF_RESOURCE_ATTRIBUTES, "") 

69 if raw_attrs.strip(): 

70 try: 

71 parse_resource_attributes(raw_attrs) 

72 except ValueError: 

73 errors[CONF_RESOURCE_ATTRIBUTES] = "invalid_attributes" 

74 

75 if not errors: 

76 await self.async_set_unique_id(f"{DOMAIN}_{BACKEND_OTEL}") 

77 self._abort_if_unique_id_configured() 

78 return self.async_create_entry( 

79 title=f"OTLP @ {host}:{port}", 

80 data={**user_input, CONF_BACKEND: BACKEND_OTEL}, 

81 ) 

82 

83 return self.async_show_form( 

84 step_id="otel", 

85 data_schema=OTEL_DATA_SCHEMA, 

86 errors=errors, 

87 ) 

88 

89 async def async_step_syslog(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult: 

90 """Handle Syslog RFC 5424 configuration.""" 

91 errors: dict[str, str] = {} 

92 

93 if user_input is not None: 

94 host = user_input[CONF_HOST] 

95 port = user_input[CONF_PORT] 

96 protocol = user_input[CONF_PROTOCOL] 

97 use_tls = user_input.get(CONF_USE_TLS, False) 

98 

99 # Validate connectivity 

100 error = await syslog_validate(self.hass, host, port, protocol, use_tls) 

101 if error: 

102 errors["base"] = error 

103 

104 if not errors: 

105 await self.async_set_unique_id(f"{DOMAIN}_{BACKEND_SYSLOG}") 

106 self._abort_if_unique_id_configured() 

107 return self.async_create_entry( 

108 title=f"Syslog @ {host}:{port} ({protocol.upper()})", 

109 data={**user_input, CONF_BACKEND: BACKEND_SYSLOG}, 

110 ) 

111 

112 return self.async_show_form( 

113 step_id="syslog", 

114 data_schema=SYSLOG_DATA_SCHEMA, 

115 errors=errors, 

116 )