Coverage for relppy/server.py: 82%

1import socketserver 

2from .protocol import Message, relp_ua 

3from logging import getLogger 

4 

5_log = getLogger(__name__) 

6 

7 

8class RelpStreamHandler(socketserver.StreamRequestHandler): 

9 """ 

10 RELP server 

11 """ 

12 autoack = True 

13 

14 def finish(self): 

15 if not self.wfile.closed: 

16 _log.debug("send server close") 

17 self.wfile.write(Message(0, b"serverclose").pack()) 

18 super().finish() 

19 

20 def _ack(self, txn: int, msg: str, add: str | None = None): 

21 if txn == 0: 21 ↛ 22line 21 didn't jump to line 22 because the condition on line 21 was never true

22 _log.debug("no ack sent: msg=%s", msg) 

23 return 

24 if add: 

25 msg1 = msg + "\n" + add 

26 else: 

27 msg1 = msg 

28 self.wfile.write(Message(txn, b"rsp", msg1.encode("utf-8")).pack()) 

29 _log.debug("ack(%s) sent: %s", msg, txn) 

30 

31 def do_any(self, msg: Message) -> str | None: 

32 raise NotImplementedError(f"do_any {msg}") 

33 

34 def do_close(self, msg: Message) -> None: 

35 _log.info("close %s", msg) 

36 self._ack(msg.txnr, "200 OK", "bye") 

37 raise EOFError("got close") 

38 

39 def do_syslog(self, msg: Message) -> str | None: 

40 _log.info("syslog %s", msg) 

41 raise NotImplementedError(f"do_syslog {msg}") 

42 

43 def do_open(self, msg: Message) -> str: 

44 _log.info("open %s", msg) 

45 self.client_nego: dict[str, list[str]] = {} 

46 for i in msg.data.splitlines(): 

47 if b"=" in i: 

48 k, v = i.split(b"=", 1) 

49 self.client_nego[k.decode()] = v.decode().split(",") 

50 _log.info("client negotiation: %s", self.client_nego) 

51 ignore = {"do_any", "do_open", "do_close"} 

52 command_set = {x.removeprefix("do_") for x in dir(self) if x.startswith("do_") and x not in ignore} 

53 client_commands = set(self.client_nego.get("commands", [])) 

54 commands = ",".join(command_set & client_commands) 

55 return f"relp_version=1\nrelp_software={relp_ua}\ncommands={commands}" 

56 

57 def handle(self): 

58 while True: 

59 try: 

60 msg = self._readmsg() 

61 self._execmsg(msg) 

62 except EOFError: 

63 _log.info("closed?") 

64 self.finish() 

65 break 

66 

67 def _readmsg(self) -> Message: 

68 l0 = self.rfile.readline() 

69 if len(l0) == 0: 69 ↛ 70line 69 didn't jump to line 70 because the condition on line 69 was never true

70 raise EOFError("closed") 

71 l1 = l0.split(b" ", 3) 

72 txnr = int(l1[0]) 

73 command = l1[1] 

74 if l1[2] == b"0\n": 

75 datalen = 0 

76 data = b"" 

77 else: 

78 datalen = int(l1[2]) 

79 data = l1[3] 

80 if len(data) != datalen+1: 

81 data += self.rfile.read(datalen-len(data)+1) 

82 if data[-1] != ord(b"\n"): 82 ↛ 83line 82 didn't jump to line 83 because the condition on line 82 was never true

83 _log.warning("invalid message tail: %s", data) 

84 else: 

85 data = data[:-1] 

86 return Message(txnr, command, data) 

87 

88 def _execmsg(self, msg: Message): 

89 _log.debug("all recv: %s", msg) 

90 fname = f"do_{msg.command.decode('ascii')}" 

91 try: 

92 if hasattr(self, fname): 92 ↛ 95line 92 didn't jump to line 95 because the condition on line 92 was always true

93 ackadd = getattr(self, fname)(msg) 

94 else: 

95 ackadd = self.do_any(msg) 

96 except EOFError: 

97 raise 

98 except Exception as exc: 

99 _log.exception("caught error: msg=%s", msg) 

100 if self.autoack: 

101 self._ack(msg.txnr, f"500 {exc}") 

102 raise 

103 else: 

104 if self.autoack: 104 ↛ exitline 104 didn't return from function '_execmsg' because the condition on line 104 was always true

105 self._ack(msg.txnr, "200 OK", ackadd)