Coverage for extras/management/commands/webhook_receiver.py: 0%

1import json 

2import sys 

3from http.server import BaseHTTPRequestHandler, HTTPServer 

4 

5from django.core.management.base import BaseCommand 

6 

7request_counter = 1 

8 

9 

10class WebhookHandler(BaseHTTPRequestHandler): 

11 show_headers = True 

12 

13 def __getattr__(self, item): 

14 

15 # Return the same method for any type of HTTP request (GET, POST, etc.) 

16 if item.startswith('do_'): 

17 return self.do_ANY 

18 

19 raise AttributeError 

20 

21 def log_message(self, format_str, *args): 

22 global request_counter 

23 

24 print("[{}] {} {} {}".format( 

25 request_counter, 

26 self.date_time_string(), 

27 self.address_string(), 

28 format_str % args 

29 )) 

30 

31 def do_ANY(self): 

32 global request_counter 

33 

34 # Send a 200 response regardless of the request content 

35 self.send_response(200) 

36 self.end_headers() 

37 self.wfile.write(b'Webhook received!\n') 

38 

39 # Print the request headers 

40 if self.show_headers: 

41 for k, v in self.headers.items(): 

42 print(f'{k}: {v}') 

43 print() 

44 

45 # Print the request body (if any) 

46 content_length = self.headers.get('Content-Length') 

47 if content_length is not None: 

48 body = self.rfile.read(int(content_length)).decode('utf-8') 

49 if self.headers.get('Content-Type') == 'application/json': 

50 body = json.loads(body) 

51 print(json.dumps(body, indent=4)) 

52 else: 

53 print('(No body)') 

54 

55 print(f'Completed request #{request_counter}') 

56 print('------------') 

57 

58 request_counter += 1 

59 

60 

61class Command(BaseCommand): 

62 help = "Start a simple listener to display received HTTP requests" 

63 

64 default_port = 9000 

65 

66 def add_arguments(self, parser): 

67 parser.add_argument( 

68 '--port', type=int, default=self.default_port, 

69 help="Optional port number (default: {})".format(self.default_port) 

70 ) 

71 parser.add_argument( 

72 "--no-headers", action='store_true', dest='no_headers', 

73 help="Hide HTTP request headers" 

74 ) 

75 

76 def handle(self, *args, **options): 

77 port = options['port'] 

78 quit_command = 'CTRL-BREAK' if sys.platform == 'win32' else 'CONTROL-C' 

79 

80 WebhookHandler.show_headers = not options['no_headers'] 

81 

82 self.stdout.write('Listening on port http://localhost:{}. Stop with {}.'.format(port, quit_command)) 

83 httpd = HTTPServer(('localhost', port), WebhookHandler) 

84 

85 try: 

86 httpd.serve_forever() 

87 except KeyboardInterrupt: 

88 self.stdout.write("\nExiting...")