Coverage for src/km3modules/communication.py: 34%

1# Filename: communication.py 

2# -*- coding: utf-8 -*- 

3# pylint: disable=locally-disabled 

4""" 

5Talk to the ELOG, interact with the Chat server etc. 

6 

7""" 

8 

9import builtins 

10import os 

11import passlib 

12import requests 

13import time 

14 

15import km3pipe as kp 

16 

17 

18class ELOGService(kp.Module): 

19 """Allows communication with the KM3NeT ELOG server. 

20 

21 Parameters 

22 ---------- 

23 username: string (optional) 

24 password: string (required) 

25 url: string (optional) 

26 

27 """ 

28 

29 def configure(self): 

30 self.url = self.get("url", default="https://elog.km3net.de/") 

31 self._password = create_elog_password_hash(self.require("password")) 

32 self._username = self.get("username", default="km3net") 

33 self.expose(self.post_elog, "post_elog") 

34 

35 def post_elog( 

36 self, logbook, subject, message, author, message_type="Comment", files=None 

37 ): 

38 """Post an ELOG entry 

39 

40 Make sure to provide an existing `author`. 

41 

42 Parameters 

43 ---------- 

44 logbook: string (e.g. "individual+Logbooks") 

45 subject: string 

46 message: string 

47 author: string (make sure to provide an existing one!) 

48 message_type: string (default="Comment") 

49 files: list of strings (filenames, optional) 

50 """ 

51 data = { 

52 "exp": logbook, 

53 "cmd": "Submit", 

54 "Subject": subject, 

55 "Author": author, 

56 "Type": message_type, 

57 "unm": self._username, 

58 "upwd": self._password, 

59 "When": int(time.time()), 

60 } 

61 if files is None: 

62 files = [] 

63 files = [ 

64 ("attfile{}".format(idx + 1), (os.path.basename(f), builtins.open(f, "rb"))) 

65 for idx, f in enumerate(files) 

66 ] 

67 files.append(("Text", ("", message))) 

68 try: 

69 r = requests.post( 

70 self.url + "/" + logbook, 

71 data=data, 

72 files=files, 

73 allow_redirects=False, 

74 verify=False, 

75 ) 

76 except requests.RequestException as e: 

77 self.log.error("Cannot reach the ELOG server!\nError: {}".format(e)) 

78 else: 

79 if r.status_code not in [200, 302]: 

80 self.log.error( 

81 "Something went wrong...\n\nHere is what we got:\n{}".format( 

82 r.content.decode("utf-8", "ignore") 

83 ) 

84 ) 

85 else: 

86 self.cprint("ELOG post created successfully.") 

87 

88 for (_, (_, fobj)) in files: 

89 if fobj and hasattr(fobj, "close"): 

90 fobj.close() 

91 

92 

93def create_elog_password_hash(password): 

94 """Create a SHA256 encrypted password for ELOGs.""" 

95 from passlib.hash import sha256_crypt 

96 

97 return sha256_crypt.encrypt(password, salt="", rounds=5000)[4:]