/openthread-latest/tests/scripts/thread-cert/ |
D | test_ipv6.py | 174 def any_udp_payload(src_port, dst_port, payload, checksum): argument 176 …return bytearray([(src_port >> 8) & 0xff, src_port & 0xff, (dst_port >> 8) & 0xff, dst_port & 0xff, 412 src_port = any_port() 417 udp_header = UDPHeader(src_port, dst_port, payload_length, checksum) 423 self.assertEqual(src_port, struct.unpack("!H", data[0:2])[0]) 430 src_port = any_port() 435 data = struct.pack("!H", src_port) + struct.pack("!H", dst_port) + \ 442 self.assertEqual(src_port, udp_header.src_port) 550 UDPHeader(src_port=19788, dst_port=19788), 678 src_port = any_port() [all …]
|
D | common.py | 97 def src_port(self): member in MessageInfo 100 @src_port.setter 101 def src_port(self, value): member in MessageInfo
|
D | ipv6.py | 482 def __init__(self, src_port, dst_port, payload_length=0, checksum=0): argument 483 self.src_port = src_port 502 data = struct.pack(">H", self.src_port) 511 src_port = struct.unpack(">H", data.read(2))[0] 516 return cls(src_port, dst_port, payload_length, checksum) 1061 if message_info.src_port in self._factories: 1062 factory = self._factories[message_info.src_port] 1088 message_info.src_port = header.src_port
|
D | test_lowpan.py | 2202 src_port = any_src_port() 2205 data_bytes = struct.pack(">H", src_port) + struct.pack(">H", dst_port) 2211 self.assertEqual(src_port, actual_src_port) 2223 src_port = any_src_port() 2226 data_bytes = struct.pack(">H", src_port) + bytearray([struct.pack(">H", dst_port)[1]]) 2233 self.assertEqual(src_port, actual_src_port) 2244 src_port = any_compressable_src_port() 2247 data_bytes = bytearray([struct.pack(">H", src_port)[1]]) + struct.pack(">H", dst_port) 2254 self.assertEqual(src_port, actual_src_port) 2265 src_port = any_nibble_src_port() [all …]
|
D | lowpan.py | 278 src_port = struct.unpack(">H", data.read(2))[0] 282 src_port = struct.unpack(">H", data.read(2))[0] 286 src_port = 0xf000 + ord(data.read(1)) 291 src_port = 0xf0b0 + ((udp_ports_byte >> 4) & 0x0F) 294 return src_port, dst_port 308 src_port, dst_port = self._decompress_udp_ports(udphc, data) 312 header = ipv6.UDPHeader(src_port, dst_port, checksum=checksum)
|
/openthread-latest/tests/toranj/ncp/ |
D | wpan.py | 692 src_port = src[1] 695 src_port = random.randint(49152, 65535) 711 return AsyncSender(self, src_addr, src_port, dst_addr, dst_port, msg, count, mcast_hops) 733 receiver._add_sender(sender.src_addr, sender.src_port, sender.msg, sender.count) 780 def __init__(self, node, src_addr, src_port, dst_addr, dst_port, msg, count, mcast_hops=None): argument 783 self._src_port = src_port 805 src_sock_addr = _create_socket_address(src_addr + '%' + node.interface_name, src_port) 807 src_sock_addr = _create_socket_address(src_addr, src_port) 823 def src_port(self): member in AsyncSender 972 src_port = src_sock_addr[1] [all …]
|
D | test-023-multicast-traffic.py | 90 … all([msg[1][0] != sender.src_addr and msg[1][1] != sender.src_port for msg in lsnr.all_rx_msg]))
|
/openthread-latest/tests/toranj/ |
D | README_NCP.md | 287 >>> sender.src_port 298 # get all received msg by `recver` as list of tuples `(msg, (src_address, src_port))`
|