1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
|
from socket import socket, AF_UNIX, SOCK_DGRAM
from select import select
from time import time
from struct import unpack, calcsize
from util import ppc
from scapy.layers.l2 import Ether
client_uds_socket_name = "client-uds-socket"
vpp_uds_socket_name = "vpp-uds-socket"
VPP_PUNT_HEADER_FMT = '=Ii'
VPP_PUNT_HEADER_SIZE = calcsize(VPP_PUNT_HEADER_FMT)
class VppPuntAction:
PUNT_L2 = 0
PUNT_IP4_ROUTED = 1
PUNT_IP6_ROUTED = 2
class VppUDSPuntSocket(object):
def __init__(self, testcase, port, is_ip4=1, l4_protocol=0x11):
client_path = '%s/%s-%s-%s' % (testcase.tempdir,
client_uds_socket_name,
"4" if is_ip4 else "6", port)
testcase.vapi.punt_socket_register(
port, client_path, is_ip4=is_ip4, l4_protocol=l4_protocol)
self.testcase = testcase
self.uds = socket(AF_UNIX, SOCK_DGRAM)
self.uds.bind(client_path)
self.uds.connect(testcase.punt_socket_path)
def wait_for_packets(self, count, timeout=1):
packets = []
now = time()
deadline = now + timeout
while len(packets) < count and now < deadline:
r, w, e = select([self.uds], [], [self.uds], deadline - now)
if self.uds in r:
x = self.uds.recv(1024 * 1024)
sw_if_index, punt_action = unpack(
VPP_PUNT_HEADER_FMT, x[:VPP_PUNT_HEADER_SIZE])
packets.append({'sw_if_index': sw_if_index,
'punt_action': punt_action,
'packet': x[VPP_PUNT_HEADER_SIZE:]})
if self.uds in e:
raise Exception("select() indicates error on UDS socket")
now = time()
if len(packets) != count:
raise Exception("Unexpected packet count received, got %s packets,"
" expected %s packets" % (len(packets), count))
self.testcase.logger.debug(
"Got %s packets via punt socket" % len(packets))
return packets
def assert_nothing_captured(self, timeout=.25):
packets = []
now = time()
deadline = now + timeout
while now < deadline:
r, w, e = select([self.uds], [], [self.uds], deadline - now)
if self.uds in r:
x = self.uds.recv(1024 * 1024)
packets.append(Ether(x[VPP_PUNT_HEADER_SIZE:]))
if self.uds in e:
raise Exception("select() indicates error on UDS socket")
now = time()
if len(packets) > 0:
self.testcase.logger.error(
ppc("Unexpected packets captured:", packets))
raise Exception("Unexpected packet count received, got %s packets,"
" expected no packets" % len(packets))
|
