Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
:return: IPv4 packet.
:rtype: bytes
"""
ip_header = b"\x45" # Version | Header Length
ip_header += b"\x00" # "Differentiated Services Field"
ip_header += struct.pack(">H", IP_HEADER_LEN + len(payload)) # Length
ip_header += b"\x00\x01" # ID Field
ip_header += b"\x40\x00" # Flags, Fragment Offset
ip_header += b"\x40" # Time to live
ip_header += protocol
ip_header += b"\x00\x00" # Header checksum (fill in zeros in order to compute checksum)
ip_header += src_ip
ip_header += dst_ip
checksum = struct.pack(">H", helpers.ipv4_checksum(ip_header))
ip_header = ip_header[:10] + checksum + ip_header[12:]
return ip_header + payload
def test_udp_broadcast_client(self):
"""
Given: A SocketConnection 'udp' object with udp_broadcast set, and a UDP server.
When: Calling SocketConnection.open(), .send(), .recv(), and .close()
Then: send() returns length of payload.
and: Sent and received data is as expected.
"""
try:
broadcast_addr = six.next(get_local_non_loopback_ipv4_addresses_info())["broadcast"]
except StopIteration:
assert False, TEST_ERR_NO_NON_LOOPBACK_IPV4
data_to_send = helpers.str_to_bytes(
'"Never drink because you need it, for this is rational drinking, and the '
"way to death and hell. But drink because you do not need it, for this is "
'irrational drinking, and the ancient health of the world."'
)
# Given
server = MiniTestServer(proto="udp", host="")
server.bind()
t = threading.Thread(target=server.serve_once)
t.daemon = True
t.start()
# noinspection PyDeprecation
uut = SocketConnection(
host=broadcast_addr,
def css_class(self):
return helpers.test_step_info[self.type]["css_class"]
def html_log_line(self):
return helpers.format_log_msg(
msg_type="test_case", description=self.description, timestamp=self.timestamp, format_type="html"
)
def html_log_line(self):
return helpers.format_log_msg(
msg_type=self.type,
description=self.description,
data=self.data,
timestamp=self.timestamp,
truncated=self.truncated,
format_type="html",
)
def text_render(self):
s = helpers.format_log_msg(
msg_type="test_case", description=self.description, timestamp=self.timestamp, format_type="terminal"
)
return s
def text_render(self):
return helpers.format_log_msg(
msg_type=self.type,
description=self.description,
data=self.data,
timestamp=self.timestamp,
truncated=self.truncated,
format_type="terminal",
)
def request_multiple_blocks(context):
r = Request("unit-test-request")
r.push(primitives.Byte(value=1, name="string block"))
r.push(primitives.String(value="The perfection of art is to conceal art.", name="unit-test-byte"))
context.uut = r
)
session._restart_target = self._mock_restart_target()
s_initialize("test-msg-a")
s_string("test-str-value")
s_static("\r\n")
s_initialize("test-msg-b")
s_string("test-str-value")
s_static("\r\n")
session.connect(s_get("test-msg-a"))
session.connect(s_get("test-msg-a"), s_get("test-msg-b"))
# When
session.fuzz_single_case(s_get("test-msg-a").num_mutations() + 1)
# Then
t.join(THREAD_WAIT_TIMEOUT)
self.assertFalse(t.is_alive())
self.assertEqual(1, self.restarts)
fuzz_loggers=[], # log to nothing
check_data_received_each_request=True,
keep_web_open=False,
)
session._restart_target = self._mock_restart_target()
s_initialize("test-msg-a")
s_string("test-str-value")
s_static("\r\n")
s_initialize("test-msg-b")
s_string("test-str-value")
s_static("\r\n")
session.connect(s_get("test-msg-a"))
session.connect(s_get("test-msg-a"), s_get("test-msg-b"))
# When
session.fuzz_single_case(s_get("test-msg-a").num_mutations() + 1)
# Then
t.join(THREAD_WAIT_TIMEOUT)
self.assertFalse(t.is_alive())
self.assertEqual(1, self.restarts)