Coverage for selenible/modules/webhook.py: 26%

1import json 

2 

3import requests 

4import yaml 

5 

6webhook_schema = yaml.safe_load(""" 

7type: object 

8properties: 

9 url: {type: string} 

10 query: {type: object} 

11 body: {type: object} 

12 method: {type: string} 

13 cookies: {type: object} 

14 headers: {type: object} 

15required: [url, body] 

16""") 

17 

18 

19def Base_webhook(self, params): 

20 """ 

21 - name: notify webhook 

22 webhook: 

23 url: https://host/path 

24 query: 

25 id: "xyz" 

26 body: 

27 text: "hello, world" 

28 """ 

29 url = params.get("url") 

30 query = params.get("query", {}) 

31 body = params.get("body", {}) 

32 method = params.get("method", "post") 

33 cookies = params.get("cookies", {}) 

34 headers = params.get("headers", {"content-type": "application/json"}) 

35 timeout = params.get("timeout", None) 

36 sess = requests.Session() 

37 for name, value in cookies.items(): 

38 sess.cookies.set(name, value) 

39 resp = sess.request( 

40 method, 

41 url, 

42 params=query, 

43 headers=headers, 

44 timeout=timeout, 

45 data=json.dumps(body, ensure_ascii=False), 

46 ) 

47 return resp.json()