Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def tests_format_message():
for message in message_1, message_2:
source, msg, result = message
msg = struct.pack('>%iB' % len(msg), *msg)
assert format_message(*source)[0] == msg
def test_read_message_wrong_address():
msg, stat = format_message(b'test', [])
with raises(ValueError, match="doesn't start with a '/'") as e:
address, tags, values, size = read_message(msg)
[1.2345, 1.2345, 10000000000.],
list(range(500)),
]
print("#" * 80)
print("format/parse test")
for i, pattern in enumerate(patterns):
print("*" * 100)
print(f"pattern: {i}")
n = 0
timeout = time() + DURATION
while time() < timeout:
n += 1
p, s = format_message(b'/address', pattern)
size = len(p) / 1000
print(
f"formated message {n} times ({n / DURATION}/s) "
f"({n * size / DURATION:.2f}MB/s)"
)
n = 0
timeout = time() + DURATION
while time() < timeout:
n += 1
read_message(p)
print(
f"parsed message {n} times ({n / DURATION}/s) "
osc.bind(b'/count', count, sock=sock)
for i, pattern in enumerate(patterns):
for safer in (False, True):
timeout = time() + DURATION
sent = 0
received = 0
while time() < timeout:
send_message(
b'/count', pattern, address, port, sock=SOCK, safer=safer
)
sent += 1
sleep(10e-9)
size = len(format_message(b'/count', pattern)[0]) / 1000.
print(
f"{i}: safe: {safer}\t",
f"sent:\t{sent}\t({sent / DURATION}/s)\t({sent * size / DURATION:.2f}MB/s)\t" # noqa
f"received:\t{received}\t({received / DURATION}/s)\t({received * size / DURATION:.2f}MB/s)\t" # noqa
f"loss {((sent - received) / sent) * 100}%"
)
if family == 'unix':
os.unlink(address)
osc.stop_all()
def test_format_wrong_types():
with raises(TypeError):
format_message(b'/test', values=[u'test'])
def test_format_encoding():
s = u'éééààà'
with raises(TypeError):
read_message(format_message('/test', [s])[0])
assert read_message(format_message('/test', [s], encoding='utf8')[0])[2][0] == s.encode('utf8') # noqa
assert read_message(format_message('/test', [s], encoding='utf8')[0], encoding='utf8')[2][0] == s # noqa
with raises(UnicodeEncodeError):
format_message('/test', [s], encoding='ascii') # noqa
with raises(UnicodeDecodeError):
read_message(format_message('/test', [s], encoding='utf8')[0], encoding='ascii') # noqa
assert read_message(
format_message('/test', [s], encoding='utf8')[0],
encoding='ascii', encoding_errors='ignore'
)[2][0] == ''
assert read_message(
def tests_format_message_null_terminated_address():
for message in message_1, message_2:
source, msg, result = message
source = source[0] + b'\0', source[1]
msg = struct.pack('>%iB' % len(msg), *msg)
assert format_message(*source)[0] == msg
def test_format_encoding():
s = u'éééààà'
with raises(TypeError):
read_message(format_message('/test', [s])[0])
assert read_message(format_message('/test', [s], encoding='utf8')[0])[2][0] == s.encode('utf8') # noqa
assert read_message(format_message('/test', [s], encoding='utf8')[0], encoding='utf8')[2][0] == s # noqa
with raises(UnicodeEncodeError):
format_message('/test', [s], encoding='ascii') # noqa
with raises(UnicodeDecodeError):
read_message(format_message('/test', [s], encoding='utf8')[0], encoding='ascii') # noqa
assert read_message(
format_message('/test', [s], encoding='utf8')[0],
encoding='ascii', encoding_errors='ignore'
)[2][0] == ''
assert read_message(
format_message('/test', [s], encoding='utf8')[0],
encoding='ascii', encoding_errors='replace'
)[2][0] == u'������������'
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
send_message(b'/test', [], '192.168.0.1', 8000, sock=sock, safer=True)
# unix sockets work on linux and osx, and over unix platforms,
# but not windows
sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
send_message(b'/some/address', [1, 2, 3], b'/tmp/sock')
"""
if platform != 'win32' and sock.family == socket.AF_UNIX:
address = ip_address
else:
address = (ip_address, port)
message, stats = format_message(
osc_address, values, encoding=encoding,
encoding_errors=encoding_errors
)
sock.sendto(message, address)
if safer:
sleep(10e-9)
return stats