Coverage for rfpy/api/endpoints/webhooks.py: 96%

1""" 

2Manage Webhook Subscriptions - register external URLs to receive notification of system events. 

3""" 

4 

5from typing import List 

6 

7from rfpy.auth import perms 

8from rfpy.web import serial 

9from rfpy.suxint import http 

10from rfpy.model.notify import WebhookSubscription 

11from rfpy.model.humans import Organisation 

12 

13from .. import fetch, validate 

14 

15 

16@http 

17def post_webhook(session, user, webhook_doc: serial.NewWebhook): 

18 """ 

19 Save a new Webhook subscribing to the event_type given in the body JSON document. 

20 

21 The will sent a POST request to the provided `url` with a JSON encoded body for each event 

22 of the given `event_type`. 

23 """ 

24 target_org = user.organisation 

25 if webhook_doc.org_id != user.org_id: 

26 target_org = fetch.organisation(session, webhook_doc.org_id) 

27 validate.check(user, perms.MANAGE_ORGANISATION, target_org=target_org) 

28 whs = WebhookSubscription(**webhook_doc.model_dump()) 

29 session.add(whs) 

30 

31 

32@http 

33def put_webhook(session, user, webhook_doc: serial.NewWebhook): 

34 """ 

35 Update the `remote_url` or `http_header` for the Webhook with the given event_type / org_id. 

36 

37 delivery_status and retries are reset to default values 

38 """ 

39 target_org = user.organisation 

40 if webhook_doc.org_id != user.org_id: 

41 target_org = fetch.organisation(session, webhook_doc.org_id) 

42 validate.check(user, perms.MANAGE_ORGANISATION, target_org=target_org) 

43 whd = webhook_doc 

44 wh: WebhookSubscription = session.get( 

45 WebhookSubscription, (whd.org_id, whd.event_type) 

46 ) 

47 if wh is None: 

48 raise ValueError( 

49 f"No Subscription found for org {whd.org_id} and {whd.event_type}" 

50 ) 

51 wh.remote_url = str(webhook_doc.remote_url) 

52 wh.http_header = webhook_doc.http_header 

53 wh.reset_status() 

54 

55 

56@http 

57def get_webhooks(session, user, q_org_id) -> List[serial.Webhook]: 

58 """ 

59 Fetch an array of Webhook objects for the current users' organisation, or, if given, the 

60 organisation indicated by query param `orgId`. 

61 """ 

62 target_org = user.organisation 

63 if q_org_id is not None: 

64 target_org = fetch.organisation(session, q_org_id) 

65 validate.check(user, perms.MANAGE_ORGANISATION, target_org=target_org) 

66 return [ 

67 serial.Webhook.model_validate(wh) for wh in target_org.webhook_subscriptions 

68 ] 

69 

70 

71@http 

72def delete_webhook(session, user, webhook_doc: serial.NewWebhook): 

73 """ 

74 Delete the webhook for the given `event_type` - `org_id` combination 

75 """ 

76 target_org = user.organisation 

77 if webhook_doc.org_id != user.org_id: 

78 target_org = session.get(Organisation, webhook_doc.org_id) 

79 if target_org is None: 

80 raise ValueError(f"Organisation {webhook_doc.org_id} not found") 

81 validate.check(user, perms.MANAGE_ORGANISATION, target_org=target_org) 

82 whd = webhook_doc 

83 whs = session.get(WebhookSubscription, (whd.org_id, whd.event_type)) 

84 if whs is None: 

85 raise ValueError( 

86 f"No Subscription found for org {whd.org_id} and {whd.event_type}" 

87 ) 

88 session.delete(whs) 

89 

90 

91@http 

92def get_webhook_events(session, user) -> List[str]: 

93 """ 

94 Fetch an array of event types which accept webhook subscriptions 

95 """ 

96 from rfpy.jobs.events.action import webhook_evt_types 

97 

98 return webhook_evt_types()