mirror of
https://github.com/Tautulli/Tautulli.git
synced 2025-07-07 05:31:15 -07:00
52 lines
1.1 KiB
Python
52 lines
1.1 KiB
Python
"""self-explanatory."""
|
|
|
|
from cheroot import makefile
|
|
|
|
|
|
__metaclass__ = type
|
|
|
|
|
|
class MockSocket:
|
|
"""Mocks a socket."""
|
|
|
|
def __init__(self):
|
|
"""Initialize."""
|
|
self.messages = []
|
|
|
|
def recv_into(self, buf):
|
|
"""Simulate recv_into for Python 3."""
|
|
if not self.messages:
|
|
return 0
|
|
msg = self.messages.pop(0)
|
|
for index, byte in enumerate(msg):
|
|
buf[index] = byte
|
|
return len(msg)
|
|
|
|
def recv(self, size):
|
|
"""Simulate recv for Python 2."""
|
|
try:
|
|
return self.messages.pop(0)
|
|
except IndexError:
|
|
return ''
|
|
|
|
def send(self, val):
|
|
"""Simulate a send."""
|
|
return len(val)
|
|
|
|
|
|
def test_bytes_read():
|
|
"""Reader should capture bytes read."""
|
|
sock = MockSocket()
|
|
sock.messages.append(b'foo')
|
|
rfile = makefile.MakeFile(sock, 'r')
|
|
rfile.read()
|
|
assert rfile.bytes_read == 3
|
|
|
|
|
|
def test_bytes_written():
|
|
"""Writer should capture bytes writtten."""
|
|
sock = MockSocket()
|
|
sock.messages.append(b'foo')
|
|
wfile = makefile.MakeFile(sock, 'w')
|
|
wfile.write(b'bar')
|
|
assert wfile.bytes_written == 3
|