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

1"""The remote_logger integration: ship HA system_log_event to an OTLP collector or syslog server.""" 

2 

3from __future__ import annotations 

4 

5import asyncio 

6import contextlib 

7import logging 

8from typing import TYPE_CHECKING 

9 

10from .const import ( 

11 BACKEND_SYSLOG, 

12 CONF_BACKEND, 

13 DOMAIN, 

14 EVENT_SYSTEM_LOG, 

15) 

16from .otel.exporter import OtlpLogExporter 

17from .syslog.exporter import SyslogExporter 

18 

19REF_CANCEL_LISTENER = "cancel_listener" 

20REF_FLUSH_TASK = "flush_task" 

21REF_EXPORTER = "exporter" 

22 

23if TYPE_CHECKING: 

24 from homeassistant.config_entries import ConfigEntry 

25 from homeassistant.core import HomeAssistant 

26 

27_LOGGER = logging.getLogger(__name__) 

28 

29 

30async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: 

31 """Set up remote logs from a config entry.""" 

32 backend = entry.data.get(CONF_BACKEND) 

33 

34 exporter: OtlpLogExporter | SyslogExporter 

35 if backend == BACKEND_SYSLOG: 

36 exporter = SyslogExporter(hass, entry) 

37 label: str = exporter.endpoint_desc 

38 else: 

39 exporter = OtlpLogExporter(hass, entry) 

40 label = exporter.endpoint_url 

41 

42 cancel_listener = hass.bus.async_listen(EVENT_SYSTEM_LOG, exporter.handle_event) 

43 flush_task: asyncio.Task[None] = asyncio.create_task(exporter.flush_loop()) 

44 

45 hass.data.setdefault(DOMAIN, {}) 

46 hass.data[DOMAIN][entry.entry_id] = { 

47 REF_CANCEL_LISTENER: cancel_listener, 

48 REF_FLUSH_TASK: flush_task, 

49 REF_EXPORTER: exporter, 

50 } 

51 

52 _LOGGER.info("remote_logger: listening for system_log_event, exporting %s to %s", backend, label) 

53 return True 

54 

55 

56async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: 

57 """Unload remote_logger config entry.""" 

58 data = hass.data[DOMAIN].pop(entry.entry_id, None) 

59 if data is None: 

60 return True 

61 

62 if data.get(REF_CANCEL_LISTENER): 

63 data[REF_CANCEL_LISTENER]() 

64 del data[REF_CANCEL_LISTENER] 

65 

66 if data.get(REF_FLUSH_TASK): 

67 data[REF_FLUSH_TASK].cancel() 

68 with contextlib.suppress(asyncio.CancelledError): 

69 await data[REF_FLUSH_TASK] 

70 del data[REF_FLUSH_TASK] 

71 

72 if data.get(REF_EXPORTER): 

73 await data[REF_EXPORTER].flush() 

74 await data[REF_EXPORTER].close() 

75 del data[REF_EXPORTER] 

76 

77 _LOGGER.info("remote_logger: unloaded, flushed remaining logs") 

78 return True