Coverage for dataexcept/exceptions/notification.py: 88%

1# notification.py 

2from ..redaction import redact_url 

3from .base import JobError 

4 

5 

6class NotificationError(JobError): 

7 """Base exception for notification failures.""" 

8 

9 def __init__( 

10 self, 

11 channel: str, 

12 original_exception: Exception | None = None, 

13 message: str | None = None, 

14 ): 

15 self.channel = channel 

16 self.original_exception = original_exception 

17 msg = message or f"Notification via '{channel}' failed" 

18 if message is None and original_exception: 18 ↛ 19line 18 didn't jump to line 19 because the condition on line 18 was never true

19 msg += f": {original_exception}" 

20 super().__init__(msg) 

21 

22 

23class EmailError(NotificationError): 

24 """Raised when sending an email fails.""" 

25 

26 def __init__( 

27 self, 

28 recipient: str, 

29 subject: str, 

30 original_exception: Exception | None = None, 

31 ): 

32 self.recipient = recipient 

33 self.subject = subject 

34 # original_exception is set by NotificationError.__init__ below. 

35 msg = f"Email to '{recipient}' with subject '{subject}' failed" 

36 if original_exception: 36 ↛ 37line 36 didn't jump to line 37 because the condition on line 36 was never true

37 msg += f": {original_exception}" 

38 super().__init__("email", original_exception, message=msg) 

39 

40 

41class WebhookError(NotificationError): 

42 """Raised when a webhook POST fails.""" 

43 

44 # Slack, Discord and others put the secret in the webhook path, so keeping 

45 # the path would defeat the redaction. This also applies to the scrubbing 

46 # of the whole message, which is how a wrapped HTTP exception quoting the 

47 # original URL gets its path dropped too. 

48 _keep_url_path = False 

49 

50 def __init__(self, url: str, original_exception: Exception | None = None): 

51 self.url = redact_url(url, keep_path=False) 

52 # original_exception is set by NotificationError.__init__ below. 

53 msg = f"Webhook to URL '{self.url}' failed" 

54 if original_exception: 

55 msg += f": {original_exception}" 

56 super().__init__("webhook", original_exception, message=msg)