Coverage for custom_components/supernotify/diagnostics.py: 100%

1"""Diagnostics support for the Supernotify integration. 

2 

3Exposes the same operational state already surfaced piecemeal through the various 

4`supernotify.enquire_*` admin services (notify.py) - recipients, scenario/delivery routing, 

5snoozes, the last notification sent, and the sent/failure counters - through Home Assistant's 

6standard Settings > Devices & Services > Supernotify > Download diagnostics flow, so a bug 

7report doesn't require walking the user through calling half a dozen services first. 

8""" 

9 

10from __future__ import annotations 

11 

12from functools import partial 

13from typing import TYPE_CHECKING, Any 

14 

15from homeassistant.core import HomeAssistant 

16from homeassistant.helpers import issue_registry as ir 

17from homeassistant.helpers.redact import REDACTED, async_redact_data, partial_redact 

18 

19if TYPE_CHECKING: 

20 from . import SupernotifyConfigEntry 

21 

22_partial_redact = partial(partial_redact, unmasked_prefix=2, unmasked_suffix=1) 

23 

24# Person.as_dict() (people.py) is the source of every identifying field below - reached both 

25# directly (enquire_recipients) and indirectly (a snooze or the last notification can reference 

26# a recipient by the same fields). async_redact_data walks every nested dict/list, so listing 

27# the field names here is enough regardless of where they turn up in the returned structure. 

28# email/phone_number keep a couple of characters visible, as Target.as_dict() does, so a bug 

29# report stays identifiable without exposing the whole address/number. 

30TO_REDACT = { 

31 "email": _partial_redact, 

32 "phone_number": _partial_redact, 

33 "user_id": lambda _value: REDACTED, 

34 "alias": lambda _value: REDACTED, 

35} 

36 

37 

38async def async_get_config_entry_diagnostics(hass: HomeAssistant, entry: SupernotifyConfigEntry) -> dict[str, Any]: 

39 """Return diagnostics for a config entry.""" 

40 service = entry.runtime_data 

41 last_notification = service.last_notification 

42 

43 data: dict[str, Any] = { 

44 "entry": { 

45 "version": entry.version, 

46 "minor_version": entry.minor_version, 

47 "data": dict(entry.data), 

48 "options": dict(entry.options), 

49 }, 

50 "counters": { 

51 "sent": service.sent, 

52 "failures": service.failures, 

53 }, 

54 "housekeeping": service.housekeeping, 

55 "recipients": service.enquire_recipients(), 

56 "scenarios": service.enquire_scenarios(), 

57 "deliveries_by_scenario": service.enquire_deliveries_by_scenario(), 

58 "implicit_deliveries": service.enquire_implicit_deliveries(), 

59 "snoozes": service.enquire_snoozes(), 

60 "last_notification": last_notification.contents(diagnostics=True) if last_notification else None, 

61 "archive_enabled": service.context.archive.enabled, 

62 "media_storage_configured": bool(service.context.media_storage.media_path), 

63 # configuration problems, each naming what's wrong and, where there's a choice, what's configured 

64 "issues": [ 

65 { 

66 "issue_id": issue.issue_id, 

67 "translation_key": issue.translation_key, 

68 "placeholders": issue.translation_placeholders, 

69 "severity": issue.severity, 

70 } 

71 for (domain, _issue_id), issue in ir.async_get(hass).issues.items() 

72 if domain == entry.domain and issue.active 

73 ], 

74 } 

75 

76 return async_redact_data(data, TO_REDACT)