val/irctest
1
0
Fork 0
mirror of https://github.com/progval/irctest.git synced 2025-04-07 07:49:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

test_buffering: improve log readability

Browse Source
This commit is contained in:
Valentin Lorentz
2021-07-02 22:25:07 +02:00
parent cd58d14608
commit 42225a68b7
1 changed files with 12 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
irctest/server_tests/test_buffering.py
View File

@ -10,31 +10,31 @@ from irctest.patma import ANYSTR
def _sendWhole(self, line): def _sendWhole(self, line):
print("sending", repr(line.encode())) print("(repr) 1 -> S", repr(line.encode()))
self.clients[1].conn.sendall(line.encode()) self.clients[1].conn.sendall(line.encode())
def _sendCharPerChar(self, line): def _sendCharPerChar(self, line):
print("sending", repr(line.encode())) print("(repr) 1 -> S", repr(line.encode()))
for char in line: for char in line:
self.clients[1].conn.sendall(char.encode()) self.clients[1].conn.sendall(char.encode())
def _sendBytePerByte(self, line): def _sendBytePerByte(self, line):
print("sending", repr(line.encode())) print("(repr) 1 -> S", repr(line.encode()))
for byte in line.encode(): for byte in line.encode():
self.clients[1].conn.sendall(bytes([byte])) self.clients[1].conn.sendall(bytes([byte]))
def _testNoTags(sender_function, colon): def _testNoTags(sender_function, colon):
def f(self): def f(self):
self.connectClient("foo") self.connectClient("nick1")
self.connectClient("bar") self.connectClient("nick2")
overhead = self.get_overhead(1, 2, colon=colon) overhead = self.get_overhead(1, 2, colon=colon)
print(f"overhead is {overhead}") print(f"overhead is {overhead}")
line = f"PRIVMSG bar {colon}" line = f"PRIVMSG nick2 {colon}"
remaining_size = 512 - len(line) - len("\r\n") remaining_size = 512 - len(line) - len("\r\n")
emoji_size = len("😃".encode()) emoji_size = len("😃".encode())
payloads = [ payloads = [
@ -68,7 +68,7 @@ def _testNoTags(sender_function, colon):
continue continue
received_line = self._getLine(2) received_line = self._getLine(2)
print("received", repr(received_line)) print("(repr) S -> 2:", repr(received_line))
try: try:
decoded_line = received_line.decode() decoded_line = received_line.decode()
except UnicodeDecodeError: except UnicodeDecodeError:
@ -82,7 +82,9 @@ def _testNoTags(sender_function, colon):
payload_intact = False payload_intact = False
else: else:
msg = message_parser.parse_message(decoded_line) msg = message_parser.parse_message(decoded_line)
self.assertMessageMatch(msg, command="PRIVMSG", params=["bar", ANYSTR]) self.assertMessageMatch(
msg, command="PRIVMSG", params=["nick2", ANYSTR]
)
payload_intact = msg.params[1] == payload payload_intact = msg.params[1] == payload
if not payload_intact: if not payload_intact:
# truncated # truncated
@ -102,9 +104,9 @@ class BufferingTestCase(cases.BaseServerTestCase):
# show_io = False # show_io = False
def get_overhead(self, client1, client2, colon): def get_overhead(self, client1, client2, colon):
self.sendLine(client1, f"PRIVMSG bar {colon}a\r\n") self.sendLine(client1, f"PRIVMSG nick2 {colon}a\r\n")
line = self._getLine(client2) line = self._getLine(client2)
return len(line) - len(f"PRIVMSG bar {colon}a\r\n") return len(line) - len(f"PRIVMSG nick2 {colon}a\r\n")
def _getLine(self, client) -> bytes: def _getLine(self, client) -> bytes:
line = b"" line = b""