(message_type, message)
| 9 | |
| 10 | |
| 11 | def send_to_webhook(message_type, message): |
| 12 | args = get_args() |
| 13 | |
| 14 | if not args.webhooks: |
| 15 | # what are you even doing here... |
| 16 | return |
| 17 | |
| 18 | data = { |
| 19 | 'type': message_type, |
| 20 | 'message': message |
| 21 | } |
| 22 | |
| 23 | for w in args.webhooks: |
| 24 | try: |
| 25 | requests.post(w, json=data, timeout=(None, 1)) |
| 26 | except requests.exceptions.ReadTimeout: |
| 27 | log.debug('Response timeout on webhook endpoint %s', w) |
| 28 | except requests.exceptions.RequestException as e: |
| 29 | log.debug(e) |
| 30 | |
| 31 | |
| 32 | def wh_updater(args, q): |
no test coverage detected