Prints the tested values that caused a test failure to ease the
troubleshooting of unreliable tests.
Also, clean-up unused variables in multiple testcases.
Signed-off-by: Francis Deslauriers <francis.deslauriers@efficios.com>
Signed-off-by: Jérémie Galarneau <jeremie.galarneau@efficios.com>
+import argparse
+import pprint
+
+from collections import defaultdict
NSEC_PER_SEC = 1000000000
NSEC_PER_SEC = 1000000000
def __init__(self, trace, pid):
self.trace = trace
self.pid = pid
def __init__(self, trace, pid):
self.trace = trace
self.pid = pid
+
+ # This dictionnary holds the results of each testcases of a test.
+ # Its layout is the following:
+ # self.expect={
+ # 'event_name_1': {'check_1': 0, 'check_2: 1},
+ # 'event_name_2': {'check_1': 1}
+ # }
+ # Each test classes checks the payload of different events. Each of
+ # those checks are stored in a event_name specific dictionnary in this
+ # data structure.
+ self.expect = defaultdict(lambda : defaultdict(int))
+
+ # This dictionnary holds the value recorded in the trace that are
+ # tested. Its content is use to print the values that caused a test to
+ # fail.
+ self.recorded_values = {}
def ns_to_hour_nsec(self, ns):
d = time.localtime(ns/NSEC_PER_SEC)
def ns_to_hour_nsec(self, ns):
d = time.localtime(ns/NSEC_PER_SEC)
func(self, event)
ret = 0
func(self, event)
ret = 0
- for i in self.expect.keys():
- if self.expect[i] == 0:
- print("%s not validated" % i)
- ret = 1
+ # For each event of the test case, check all entries for failed
+ for event_name, event_results in self.expect.items():
+ for val in event_results.keys():
+ if self.expect[event_name][val] == 0:
+ print("%s not validated" % val)
+ print("Values of the local variables of this test:")
+ # using pprint for pretty printing the dictionnary
+ pprint.pprint(self.recorded_values[event_name])
+ ret = 1
class Test1(TraceParser):
def __init__(self, trace, pid):
super().__init__(trace, pid)
class Test1(TraceParser):
def __init__(self, trace, pid):
super().__init__(trace, pid)
- self.expect["select_in_fd0"] = 0
- self.expect["select_in_fd1023"] = 0
- self.expect["select_out_fd0"] = 0
- self.expect["select_out_fd1023"] = 0
- self.expect["poll_in_nfds1"] = 0
- self.expect["poll_out_nfds1"] = 0
- self.expect["epoll_ctl_in_add"] = 0
- self.expect["epoll_ctl_out_ok"] = 0
- self.expect["epoll_wait_in_ok"] = 0
- self.expect["epoll_wait_out_fd0"] = 0
+ self.expect["select_entry"]["select_in_fd0"] = 0
+ self.expect["select_entry"]["select_in_fd1023"] = 0
+ self.expect["select_exit"]["select_out_fd0"] = 0
+ self.expect["select_exit"]["select_out_fd1023"] = 0
+ self.expect["poll_entry"]["poll_in_nfds1"] = 0
+ self.expect["poll_exit"]["poll_out_nfds1"] = 0
+ self.expect["epoll_ctl_entry"]["epoll_ctl_in_add"] = 0
+ self.expect["epoll_ctl_exit"]["epoll_ctl_out_ok"] = 0
+ self.expect["epoll_wait_entry"]["epoll_wait_in_ok"] = 0
+ self.expect["epoll_wait_exit"]["epoll_wait_out_fd0"] = 0
def select_entry(self, event):
def select_entry(self, event):
- timestamp = event.timestamp
- cpu_id = event["cpu_id"]
n = event["n"]
overflow = event["overflow"]
n = event["n"]
overflow = event["overflow"]
- tvp = event["tvp"]
- _readfds_length = event["_readfds_length"]
- readfds = event["readfds"]
- _writefds_length = event["_writefds_length"]
- writefds = event["writefds"]
- _exceptfds_length = event["_exceptfds_length"]
- exceptfds = event["exceptfds"]
+ readfd_0 = event["readfds"][0]
# check that the FD 0 is actually set in the readfds
# check that the FD 0 is actually set in the readfds
- if n == 1 and readfds[0] == 1:
- self.expect["select_in_fd0"] = 1
+ if n == 1 and readfd_0 == 1:
+ self.expect["select_entry"]["select_in_fd0"] = 1
+ readfd_127 = event["readfds"][127]
+ writefd_127 = event["writefds"][127]
+ exceptfd_127 = event["exceptfds"][127]
+
# check that the FD 1023 is actually set in the readfds
# check that the FD 1023 is actually set in the readfds
- if readfds[127] == 0x40 and writefds[127] == 0 and \
- exceptfds[127] == 0 and overflow == 0:
- self.expect["select_in_fd1023"] = 1
+ if readfd_127 == 0x40 and writefd_127 == 0 and \
+ exceptfd_127 == 0 and overflow == 0:
+ self.expect["select_entry"]["select_in_fd1023"] = 1
+
+ # Save values of local variables to print in case of test failure
+ self.recorded_values["select_entry"] = locals()
def select_exit(self, event):
def select_exit(self, event):
- timestamp = event.timestamp
- cpu_id = event["cpu_id"]
- overflow = event["overflow"]
+ overflow = event["overflow"]
_readfds_length = event["_readfds_length"]
_readfds_length = event["_readfds_length"]
- readfds = event["readfds"]
- _writefds_length = event["_writefds_length"]
- writefds = event["writefds"]
- _exceptfds_length = event["_exceptfds_length"]
- exceptfds = event["exceptfds"]
if ret == 1:
# check that the FD 0 is actually set in the readfds
if ret == 1:
# check that the FD 0 is actually set in the readfds
- if readfds[0] == 1:
- self.expect["select_out_fd0"] = 1
+ readfd_0 = event["readfds"][0]
+
+ if readfd_0 == 1:
+ self.expect["select_exit"]["select_out_fd0"] = 1
# check that the FD 1023 is actually set in the readfds
# check that the FD 1023 is actually set in the readfds
- if _readfds_length == 128 and readfds[127] == 0x40 and \
- writefds[127] == 0 and exceptfds[127] == 0 and tvp == 0:
- self.expect["select_out_fd1023"] = 1
+ if _readfds_length == 128:
+ readfd_127 = event["readfds"][127]
+ writefd_127 = event["writefds"][127]
+ exceptfd_127 = event["exceptfds"][127]
+ if readfd_127 == 0x40 and writefd_127 == 0 and \
+ exceptfd_127 == 0 and tvp == 0:
+ self.expect["select_exit"]["select_out_fd1023"] = 1
+
+ # Save values of local variables to print in case of test failure
+ self.recorded_values["select_exit"] = locals()
def poll_entry(self, event):
def poll_entry(self, event):
- timestamp = event.timestamp
- cpu_id = event["cpu_id"]
nfds = event["nfds"]
fds_length = event["fds_length"]
overflow = event["overflow"]
nfds = event["nfds"]
fds_length = event["fds_length"]
overflow = event["overflow"]
# check that only one FD is set, that it has the POLLIN flag and that
# the raw value matches the events bit field.
# check that only one FD is set, that it has the POLLIN flag and that
# the raw value matches the events bit field.
- if nfds == 1 and fds_length == 1 and fds[0]["raw_events"] == 0x3 \
- and fds[0]["events"]["POLLIN"] == 1 and \
- fds[0]["events"]["padding"] == 0:
- self.expect["poll_in_nfds1"] = 1
+ if nfds == 1 and fds_length == 1:
+ fd_0 = event["fds"][0]
+ if fd_0["raw_events"] == 0x3 and fd_0["events"]["POLLIN"] == 1 and \
+ fd_0["events"]["padding"] == 0:
+ self.expect["poll_entry"]["poll_in_nfds1"] = 1
+
+ # Save values of local variables to print in case of test failure
+ self.recorded_values["poll_entry"] = locals()
def poll_exit(self, event):
def poll_exit(self, event):
- timestamp = event.timestamp
- cpu_id = event["cpu_id"]
fds_length = event["fds_length"]
fds_length = event["fds_length"]
# check that only one FD is set, that it has the POLLIN flag and that
# the raw value matches the events bit field.
# check that only one FD is set, that it has the POLLIN flag and that
# the raw value matches the events bit field.
- if ret == 1 and fds_length == 1 and fds[0]["raw_events"] == 0x1 \
- and fds[0]["events"]["POLLIN"] == 1 and \
- fds[0]["events"]["padding"] == 0:
- self.expect["poll_out_nfds1"] = 1
+ if ret == 1 and fds_length == 1:
+ fd_0 = event["fds"][0]
+ if fd_0["raw_events"] == 0x1 and fd_0["events"]["POLLIN"] == 1 and \
+ fd_0["events"]["padding"] == 0:
+ self.expect["poll_exit"]["poll_out_nfds1"] = 1
+
+ # Save values of local variables to print in case of test failure
+ self.recorded_values["poll_exit"] = locals()
def epoll_ctl_entry(self, event):
def epoll_ctl_entry(self, event):
- timestamp = event.timestamp
- cpu_id = event["cpu_id"]
epfd = event["epfd"]
op_enum = event["op_enum"]
fd = event["fd"]
epfd = event["epfd"]
op_enum = event["op_enum"]
fd = event["fd"]
_event["data_union"]["fd"] == 0 and \
_event["events"]["EPOLLIN"] == 1 and \
_event["events"]["EPOLLPRI"] == 1:
_event["data_union"]["fd"] == 0 and \
_event["events"]["EPOLLIN"] == 1 and \
_event["events"]["EPOLLPRI"] == 1:
- self.expect["epoll_ctl_in_add"] = 1
+ self.expect["epoll_ctl_entry"]["epoll_ctl_in_add"] = 1
+
+ # Save values of local variables to print in case of test failure
+ self.recorded_values["epoll_ctl_entry"] = locals()
def epoll_ctl_exit(self, event):
def epoll_ctl_exit(self, event):
- timestamp = event.timestamp
- cpu_id = event["cpu_id"]
ret = event["ret"]
if ret == 0:
ret = event["ret"]
if ret == 0:
- self.expect["epoll_ctl_out_ok"] = 1
+ self.expect["epoll_ctl_exit"]["epoll_ctl_out_ok"] = 1
+
+ # Save values of local variables to print in case of test failure
+ self.recorded_values["epoll_ctl_exit"] = locals()
def epoll_wait_entry(self, event):
def epoll_wait_entry(self, event):
- timestamp = event.timestamp
- cpu_id = event["cpu_id"]
epfd = event["epfd"]
maxevents = event["maxevents"]
timeout = event["timeout"]
if epfd == 3 and maxevents == 1 and timeout == -1:
epfd = event["epfd"]
maxevents = event["maxevents"]
timeout = event["timeout"]
if epfd == 3 and maxevents == 1 and timeout == -1:
- self.expect["epoll_wait_in_ok"] = 1
+ self.expect["epoll_wait_entry"]["epoll_wait_in_ok"] = 1
+
+ # Save values of local variables to print in case of test failure
+ self.recorded_values["epoll_wait_entry"] = locals()
def epoll_wait_exit(self, event):
def epoll_wait_exit(self, event):
- timestamp = event.timestamp
- cpu_id = event["cpu_id"]
ret = event["ret"]
fds_length = event["fds_length"]
overflow = event["overflow"]
ret = event["ret"]
fds_length = event["fds_length"]
overflow = event["overflow"]
# check that FD 0 returned with EPOLLIN and the right data.fd
# check that FD 0 returned with EPOLLIN and the right data.fd
- if ret == 1 and fds_length == 1 and overflow == 0 and \
- fds[0]["data_union"]["fd"] == 0 and \
- fds[0]["events"]["EPOLLIN"] == 1:
- self.expect["epoll_wait_out_fd0"] = 1
+ if ret == 1 and fds_length == 1:
+ fd_0 = event["fds"][0]
+ if overflow == 0 and fd_0["data_union"]["fd"] == 0 and \
+ fd_0["events"]["EPOLLIN"] == 1:
+ self.expect["epoll_wait_exit"]["epoll_wait_out_fd0"] = 1
+
+ # Save values of local variables to print in case of test failure
+ self.recorded_values["epoll_wait_exit"] = locals()
class Test2(TraceParser):
def __init__(self, trace, pid):
super().__init__(trace, pid)
class Test2(TraceParser):
def __init__(self, trace, pid):
super().__init__(trace, pid)
- self.expect["select_timeout_in_fd0"] = 0
- self.expect["select_timeout_in_fd1023"] = 0
- self.expect["select_timeout_out"] = 0
- self.expect["poll_timeout_in"] = 0
- self.expect["poll_timeout_out"] = 0
- self.expect["epoll_ctl_timeout_in_add"] = 0
- self.expect["epoll_ctl_timeout_out_ok"] = 0
- self.expect["epoll_wait_timeout_in"] = 0
- self.expect["epoll_wait_timeout_out"] = 0
+ self.expect["select_entry"]["select_timeout_in_fd0"] = 0
+ self.expect["select_entry"]["select_timeout_in_fd1023"] = 0
+ self.expect["select_exit"]["select_timeout_out"] = 0
+ self.expect["poll_entry"]["poll_timeout_in"] = 0
+ self.expect["poll_exit"]["poll_timeout_out"] = 0
+ self.expect["epoll_ctl_entry"]["epoll_ctl_timeout_in_add"] = 0
+ self.expect["epoll_ctl_exit"]["epoll_ctl_timeout_out_ok"] = 0
+ self.expect["epoll_wait_entry"]["epoll_wait_timeout_in"] = 0
+ self.expect["epoll_wait_exit"]["epoll_wait_timeout_out"] = 0
def select_entry(self, event):
def select_entry(self, event):
- timestamp = event.timestamp
- cpu_id = event["cpu_id"]
- overflow = event["overflow"]
- _readfds_length = event["_readfds_length"]
- readfds = event["readfds"]
- _writefds_length = event["_writefds_length"]
- writefds = event["writefds"]
- _exceptfds_length = event["_exceptfds_length"]
- exceptfds = event["exceptfds"]
- self.expect["select_timeout_in_fd0"] = 1
+ self.expect["select_entry"]["select_timeout_in_fd0"] = 1
- if readfds[127] == 0x40 and writefds[127] == 0 and \
- exceptfds[127] == 0 and tvp != 0:
- self.expect["select_timeout_in_fd1023"] = 1
+ readfd_127 = event["readfds"][127]
+ writefd_127 = event["writefds"][127]
+ exceptfd_127 = event["exceptfds"][127]
+
+ if readfd_127 == 0x40 and writefd_127 == 0 and \
+ exceptfd_127 == 0 and tvp != 0:
+ self.expect["select_entry"]["select_timeout_in_fd1023"] = 1
+
+ # Save values of local variables to print in case of test failure
+ self.recorded_values["select_entry"] = locals()
def select_exit(self, event):
def select_exit(self, event):
- timestamp = event.timestamp
- cpu_id = event["cpu_id"]
- overflow = event["overflow"]
- _readfds_length = event["_readfds_length"]
- readfds = event["readfds"]
- _writefds_length = event["_writefds_length"]
- writefds = event["writefds"]
- _exceptfds_length = event["_exceptfds_length"]
- exceptfds = event["exceptfds"]
if ret == 0 and tvp != 0:
if ret == 0 and tvp != 0:
- self.expect["select_timeout_out"] = 1
+ self.expect["select_exit"]["select_timeout_out"] = 1
+
+ # Save values of local variables to print in case of test failure
+ self.recorded_values["select_exit"] = locals()
def poll_entry(self, event):
def poll_entry(self, event):
- timestamp = event.timestamp
- cpu_id = event["cpu_id"]
nfds = event["nfds"]
fds_length = event["fds_length"]
nfds = event["nfds"]
fds_length = event["fds_length"]
- overflow = event["overflow"]
- fds = event["fds"]
# check that we wait on FD 0 for POLLIN and that the raw_events
# field matches the value of POLLIN
# check that we wait on FD 0 for POLLIN and that the raw_events
# field matches the value of POLLIN
- if nfds == 1 and fds_length == 1 and fds[0]["raw_events"] == 0x3 \
- and fds[0]["events"]["POLLIN"] == 1 and \
- fds[0]["events"]["padding"] == 0:
- self.expect["poll_timeout_in"] = 1
+ if nfds == 1 and fds_length == 1:
+ fd_0 = event["fds"][0]
+ if fd_0["raw_events"] == 0x3 and \
+ fd_0["events"]["POLLIN"] == 1 and \
+ fd_0["events"]["padding"] == 0:
+ self.expect["poll_entry"]["poll_timeout_in"] = 1
+
+ # Save values of local variables to print in case of test failure
+ self.recorded_values["poll_entry"] = locals()
def poll_exit(self, event):
def poll_exit(self, event):
- timestamp = event.timestamp
- cpu_id = event["cpu_id"]
ret = event["ret"]
nfds = event["nfds"]
fds_length = event["fds_length"]
ret = event["ret"]
nfds = event["nfds"]
fds_length = event["fds_length"]
if ret == 0 and nfds == 1 and fds_length == 0:
if ret == 0 and nfds == 1 and fds_length == 0:
- self.expect["poll_timeout_out"] = 1
+ self.expect["poll_exit"]["poll_timeout_out"] = 1
+
+ # Save values of local variables to print in case of test failure
+ self.recorded_values["poll_exit"] = locals()
def epoll_ctl_entry(self, event):
def epoll_ctl_entry(self, event):
- timestamp = event.timestamp
- cpu_id = event["cpu_id"]
- epfd = event["epfd"]
op_enum = event["op_enum"]
op_enum = event["op_enum"]
_event = event["event"]
# make sure we see a EPOLLIN|EPOLLPRI
if op_enum == "EPOLL_CTL_ADD" and \
_event["events"]["EPOLLIN"] == 1 and \
_event["events"]["EPOLLPRI"] == 1:
_event = event["event"]
# make sure we see a EPOLLIN|EPOLLPRI
if op_enum == "EPOLL_CTL_ADD" and \
_event["events"]["EPOLLIN"] == 1 and \
_event["events"]["EPOLLPRI"] == 1:
- self.expect["epoll_ctl_timeout_in_add"] = 1
+ self.expect["epoll_ctl_entry"]["epoll_ctl_timeout_in_add"] = 1
+
+ # Save values of local variables to print in case of test failure
+ self.recorded_values["epoll_ctl_entry"] = locals()
def epoll_ctl_exit(self, event):
def epoll_ctl_exit(self, event):
- timestamp = event.timestamp
- cpu_id = event["cpu_id"]
ret = event["ret"]
if ret == 0:
ret = event["ret"]
if ret == 0:
- self.expect["epoll_ctl_timeout_out_ok"] = 1
+ self.expect["epoll_ctl_exit"]["epoll_ctl_timeout_out_ok"] = 1
+
+ # Save values of local variables to print in case of test failure
+ self.recorded_values["epoll_ctl_exit"] = locals()
def epoll_wait_entry(self, event):
def epoll_wait_entry(self, event):
- timestamp = event.timestamp
- cpu_id = event["cpu_id"]
- epfd = event["epfd"]
maxevents = event["maxevents"]
timeout = event["timeout"]
if maxevents == 1 and timeout == 1:
maxevents = event["maxevents"]
timeout = event["timeout"]
if maxevents == 1 and timeout == 1:
- self.expect["epoll_wait_timeout_in"] = 1
+ self.expect["epoll_wait_entry"]["epoll_wait_timeout_in"] = 1
+
+ # Save values of local variables to print in case of test failure
+ self.recorded_values["epoll_wait_entry"] = locals()
def epoll_wait_exit(self, event):
def epoll_wait_exit(self, event):
- timestamp = event.timestamp
- cpu_id = event["cpu_id"]
ret = event["ret"]
fds_length = event["fds_length"]
overflow = event["overflow"]
ret = event["ret"]
fds_length = event["fds_length"]
overflow = event["overflow"]
if ret == 0 and fds_length == 0 and overflow == 0:
if ret == 0 and fds_length == 0 and overflow == 0:
- self.expect["epoll_wait_timeout_out"] = 1
+ self.expect["epoll_wait_exit"]["epoll_wait_timeout_out"] = 1
+
+ # Save values of local variables to print in case of test failure
+ self.recorded_values["epoll_wait_exit"] = locals()
class Test3(TraceParser):
def __init__(self, trace, pid):
super().__init__(trace, pid)
class Test3(TraceParser):
def __init__(self, trace, pid):
super().__init__(trace, pid)
- self.expect["select_invalid_fd_in"] = 0
- self.expect["select_invalid_fd_out"] = 0
+ self.expect["select_entry"]["select_invalid_fd_in"] = 0
+ self.expect["select_exit"]["select_invalid_fd_out"] = 0
def select_entry(self, event):
def select_entry(self, event):
- timestamp = event.timestamp
- cpu_id = event["cpu_id"]
n = event["n"]
overflow = event["overflow"]
n = event["n"]
overflow = event["overflow"]
- tvp = event["tvp"]
- _readfds_length = event["_readfds_length"]
- readfds = event["readfds"]
- _writefds_length = event["_writefds_length"]
- writefds = event["writefds"]
- _exceptfds_length = event["_exceptfds_length"]
- exceptfds = event["exceptfds"]
if n > 0 and overflow == 0:
if n > 0 and overflow == 0:
- self.expect["select_invalid_fd_in"] = 1
+ self.expect["select_entry"]["select_invalid_fd_in"] = 1
+
+ # Save values of local variables to print in case of test failure
+ self.recorded_values["select_entry"] = locals()
def select_exit(self, event):
def select_exit(self, event):
- timestamp = event.timestamp
- cpu_id = event["cpu_id"]
ret = event["ret"]
overflow = event["overflow"]
ret = event["ret"]
overflow = event["overflow"]
_readfds_length = event["_readfds_length"]
_readfds_length = event["_readfds_length"]
- readfds = event["readfds"]
- _writefds_length = event["_writefds_length"]
- writefds = event["writefds"]
- _exceptfds_length = event["_exceptfds_length"]
- exceptfds = event["exceptfds"]
# make sure the event has a ret field equal to -EBADF
if ret == -9 and overflow == 0 and _readfds_length == 0:
# make sure the event has a ret field equal to -EBADF
if ret == -9 and overflow == 0 and _readfds_length == 0:
- self.expect["select_invalid_fd_out"] = 1
+ self.expect["select_exit"]["select_invalid_fd_out"] = 1
+
+ # Save values of local variables to print in case of test failure
+ self.recorded_values["select_exit"] = locals()
class Test4(TraceParser):
def __init__(self, trace, pid):
super().__init__(trace, pid)
class Test4(TraceParser):
def __init__(self, trace, pid):
super().__init__(trace, pid)
- self.expect["big_poll_in"] = 0
- self.expect["big_poll_out"] = 0
+ self.expect["poll_entry"]["big_poll_in"] = 0
+ self.expect["poll_exit"]["big_poll_out"] = 0
def poll_entry(self, event):
def poll_entry(self, event):
- timestamp = event.timestamp
- cpu_id = event["cpu_id"]
nfds = event["nfds"]
fds_length = event["fds_length"]
overflow = event["overflow"]
nfds = event["nfds"]
fds_length = event["fds_length"]
overflow = event["overflow"]
# test of big list of FDs and the behaviour of the overflow
# test of big list of FDs and the behaviour of the overflow
- if nfds == 2047 and fds_length == 512 and overflow == 1 and \
- fds[0]["raw_events"] == 0x3 \
- and fds[0]["events"]["POLLIN"] == 1 and \
- fds[0]["events"]["padding"] == 0 and \
- fds[511]["events"]["POLLIN"] == 1 and \
- fds[511]["events"]["POLLPRI"] == 1:
- self.expect["big_poll_in"] = 1
+ if nfds == 2047 and fds_length == 512 and overflow == 1:
+ fd_0 = event["fds"][0]
+ fd_511 = event["fds"][511]
+ if fd_0["raw_events"] == 0x3 and fd_0["events"]["POLLIN"] == 1 and \
+ fd_0["events"]["padding"] == 0 and \
+ fd_511["events"]["POLLIN"] == 1 and \
+ fd_511["events"]["POLLPRI"] == 1:
+ self.expect["poll_entry"]["big_poll_in"] = 1
+
+ # Save values of local variables to print in case of test failure
+ self.recorded_values["poll_entry"] = locals()
def poll_exit(self, event):
def poll_exit(self, event):
- timestamp = event.timestamp
- cpu_id = event["cpu_id"]
ret = event["ret"]
nfds = event["nfds"]
fds_length = event["fds_length"]
overflow = event["overflow"]
ret = event["ret"]
nfds = event["nfds"]
fds_length = event["fds_length"]
overflow = event["overflow"]
# test of big list of FDs and the behaviour of the overflow
# test of big list of FDs and the behaviour of the overflow
- if ret == 2047 and nfds == 2047 and fds_length == 512 and \
- overflow == 1 and fds[0]["events"]["POLLIN"] == 1 and \
- fds[511]["events"]["POLLIN"] == 1:
- self.expect["big_poll_out"] = 1
+ if ret == 2047 and nfds == 2047 and fds_length == 512 and overflow == 1:
+ fd_0 = event["fds"][0]
+ fd_511 = event["fds"][511]
+ if fd_0["events"]["POLLIN"] == 1 and fd_511["events"]["POLLIN"] == 1:
+ self.expect["poll_exit"]["big_poll_out"] = 1
+ # Save values of local variables to print in case of test failure
+ self.recorded_values["poll_exit"] = locals()
class Test5(TraceParser):
def __init__(self, trace, pid):
super().__init__(trace, pid)
class Test5(TraceParser):
def __init__(self, trace, pid):
super().__init__(trace, pid)
- self.expect["poll_overflow_in"] = 0
- self.expect["poll_overflow_out"] = 0
+ self.expect["poll_entry"]["poll_overflow_in"] = 0
+ self.expect["poll_exit"]["poll_overflow_out"] = 0
def poll_entry(self, event):
def poll_entry(self, event):
- timestamp = event.timestamp
- cpu_id = event["cpu_id"]
nfds = event["nfds"]
fds_length = event["fds_length"]
overflow = event["overflow"]
nfds = event["nfds"]
fds_length = event["fds_length"]
overflow = event["overflow"]
# test that event in valid even though the target buffer is too small
# and the program segfaults
# test that event in valid even though the target buffer is too small
# and the program segfaults
- if nfds == 100 and fds_length == 100 and overflow == 0 and \
- fds[0]["events"]["POLLIN"] == 1:
- self.expect["poll_overflow_in"] = 1
+ if nfds == 100 and fds_length == 100 and overflow == 0:
+ fd_0 = event["fds"][0]
+ if fd_0["events"]["POLLIN"] == 1:
+ self.expect["poll_entry"]["poll_overflow_in"] = 1
+
+ # Save values of local variables to print in case of test failure
+ self.recorded_values["poll_entry"] = locals()
def poll_exit(self, event):
def poll_exit(self, event):
- timestamp = event.timestamp
- cpu_id = event["cpu_id"]
- ret = event["ret"]
- fds_length = event["fds_length"]
overflow = event["overflow"]
overflow = event["overflow"]
# test that event in valid even though the target buffer is too small
# and the program segfaults
if nfds == 100 and overflow == 0:
# test that event in valid even though the target buffer is too small
# and the program segfaults
if nfds == 100 and overflow == 0:
- self.expect["poll_overflow_out"] = 1
+ self.expect["poll_exit"]["poll_overflow_out"] = 1
+
+ # Save values of local variables to print in case of test failure
+ self.recorded_values["poll_exit"] = locals()
class Test6(TraceParser):
def __init__(self, trace, pid):
super().__init__(trace, pid)
class Test6(TraceParser):
def __init__(self, trace, pid):
super().__init__(trace, pid)
- self.expect["pselect_invalid_in"] = 0
- self.expect["pselect_invalid_out"] = 0
+ self.expect["select_entry"]["pselect_invalid_in"] = 0
+ self.expect["select_exit"]["pselect_invalid_out"] = 0
def select_entry(self, event):
def select_entry(self, event):
- timestamp = event.timestamp
- cpu_id = event["cpu_id"]
n = event["n"]
overflow = event["overflow"]
n = event["n"]
overflow = event["overflow"]
_readfds_length = event["_readfds_length"]
_readfds_length = event["_readfds_length"]
- readfds = event["readfds"]
- _writefds_length = event["_writefds_length"]
- writefds = event["writefds"]
- _exceptfds_length = event["_exceptfds_length"]
- exceptfds = event["exceptfds"]
# test that event in valid even though the target buffer pointer is
# invalid and the program segfaults
if n == 1 and overflow == 0 and _readfds_length == 0:
# test that event in valid even though the target buffer pointer is
# invalid and the program segfaults
if n == 1 and overflow == 0 and _readfds_length == 0:
- self.expect["pselect_invalid_in"] = 1
+ self.expect["select_entry"]["pselect_invalid_in"] = 1
+
+ # Save values of local variables to print in case of test failure
+ self.recorded_values["select_entry"] = locals()
def select_exit(self, event):
def select_exit(self, event):
- timestamp = event.timestamp
- cpu_id = event["cpu_id"]
ret = event["ret"]
overflow = event["overflow"]
ret = event["ret"]
overflow = event["overflow"]
_readfds_length = event["_readfds_length"]
_readfds_length = event["_readfds_length"]
- readfds = event["readfds"]
- _writefds_length = event["_writefds_length"]
- writefds = event["writefds"]
- _exceptfds_length = event["_exceptfds_length"]
- exceptfds = event["exceptfds"]
# test that event in valid even though the target buffer pointer is
# invalid and the program segfaults
if ret == -14 and overflow == 0 and _readfds_length == 0:
# test that event in valid even though the target buffer pointer is
# invalid and the program segfaults
if ret == -14 and overflow == 0 and _readfds_length == 0:
- self.expect["pselect_invalid_out"] = 1
+ self.expect["select_exit"]["pselect_invalid_out"] = 1
+
+ # Save values of local variables to print in case of test failure
+ self.recorded_values["select_exit"] = locals()
class Test7(TraceParser):
def __init__(self, trace, pid):
super().__init__(trace, pid)
class Test7(TraceParser):
def __init__(self, trace, pid):
super().__init__(trace, pid)
- self.expect["poll_max_in"] = 0
- self.expect["poll_max_out"] = 0
+ self.expect["poll_entry"]["poll_max_in"] = 0
+ self.expect["poll_exit"]["poll_max_out"] = 0
def poll_entry(self, event):
def poll_entry(self, event):
- timestamp = event.timestamp
- cpu_id = event["cpu_id"]
- fds_length = event["fds_length"]
overflow = event["overflow"]
overflow = event["overflow"]
# check the proper working of INT_MAX maxevent value
if nfds == 4294967295 and overflow == 1:
# check the proper working of INT_MAX maxevent value
if nfds == 4294967295 and overflow == 1:
- self.expect["poll_max_in"] = 1
+ self.expect["poll_entry"]["poll_max_in"] = 1
+
+ # Save values of local variables to print in case of test failure
+ self.recorded_values["poll_entry"] = locals()
+
def poll_exit(self, event):
def poll_exit(self, event):
- timestamp = event.timestamp
- cpu_id = event["cpu_id"]
ret = event["ret"]
nfds = event["nfds"]
ret = event["ret"]
nfds = event["nfds"]
- fds_length = event["fds_length"]
overflow = event["overflow"]
overflow = event["overflow"]
# check the proper working of UINT_MAX maxevent value
if ret == -22 and nfds == 4294967295 and overflow == 0:
# check the proper working of UINT_MAX maxevent value
if ret == -22 and nfds == 4294967295 and overflow == 0:
- self.expect["poll_max_out"] = 1
+ self.expect["poll_exit"]["poll_max_out"] = 1
+
+ # Save values of local variables to print in case of test failure
+ self.recorded_values["poll_exit"] = locals()
class Test8(TraceParser):
def __init__(self, trace, pid):
super().__init__(trace, pid)
class Test8(TraceParser):
def __init__(self, trace, pid):
super().__init__(trace, pid)
- self.expect["epoll_wait_invalid_in"] = 0
- self.expect["epoll_wait_invalid_out"] = 0
+ self.expect["epoll_wait_entry"]["epoll_wait_invalid_in"] = 0
+ self.expect["epoll_wait_exit"]["epoll_wait_invalid_out"] = 0
def epoll_wait_entry(self, event):
def epoll_wait_entry(self, event):
- timestamp = event.timestamp
- cpu_id = event["cpu_id"]
epfd = event["epfd"]
maxevents = event["maxevents"]
timeout = event["timeout"]
epfd = event["epfd"]
maxevents = event["maxevents"]
timeout = event["timeout"]
# test that event in valid even though the target buffer pointer is
# invalid and the program segfaults
if epfd == 3 and maxevents == 1 and timeout == -1:
# test that event in valid even though the target buffer pointer is
# invalid and the program segfaults
if epfd == 3 and maxevents == 1 and timeout == -1:
- self.expect["epoll_wait_invalid_in"] = 1
+ self.expect["epoll_wait_entry"]["epoll_wait_invalid_in"] = 1
+
+ # Save values of local variables to print in case of test failure
+ self.recorded_values["epoll_wait_entry"] = locals()
def epoll_wait_exit(self, event):
def epoll_wait_exit(self, event):
- timestamp = event.timestamp
- cpu_id = event["cpu_id"]
ret = event["ret"]
fds_length = event["fds_length"]
overflow = event["overflow"]
ret = event["ret"]
fds_length = event["fds_length"]
overflow = event["overflow"]
# test that event in valid even though the target buffer pointer is
# invalid and the program segfaults
if ret == -14 and fds_length == 0 and overflow == 0:
# test that event in valid even though the target buffer pointer is
# invalid and the program segfaults
if ret == -14 and fds_length == 0 and overflow == 0:
- self.expect["epoll_wait_invalid_out"] = 1
+ self.expect["epoll_wait_exit"]["epoll_wait_invalid_out"] = 1
+
+ # Save values of local variables to print in case of test failure
+ self.recorded_values["epoll_wait_exit"] = locals()
class Test9(TraceParser):
def __init__(self, trace, pid):
super().__init__(trace, pid)
class Test9(TraceParser):
def __init__(self, trace, pid):
super().__init__(trace, pid)
- self.expect["epoll_wait_max_in"] = 0
- self.expect["epoll_wait_max_out"] = 0
+ self.expect["epoll_wait_entry"]["epoll_wait_max_in"] = 0
+ self.expect["epoll_wait_exit"]["epoll_wait_max_out"] = 0
def epoll_wait_entry(self, event):
def epoll_wait_entry(self, event):
- timestamp = event.timestamp
- cpu_id = event["cpu_id"]
epfd = event["epfd"]
maxevents = event["maxevents"]
timeout = event["timeout"]
# check the proper working of INT_MAX maxevent value
if epfd == 3 and maxevents == 2147483647 and timeout == -1:
epfd = event["epfd"]
maxevents = event["maxevents"]
timeout = event["timeout"]
# check the proper working of INT_MAX maxevent value
if epfd == 3 and maxevents == 2147483647 and timeout == -1:
- self.expect["epoll_wait_max_in"] = 1
+ self.expect["epoll_wait_entry"]["epoll_wait_max_in"] = 1
+
+ # Save values of local variables to print in case of test failure
+ self.recorded_values["epoll_wait_entry"] = locals()
def epoll_wait_exit(self, event):
def epoll_wait_exit(self, event):
- timestamp = event.timestamp
- cpu_id = event["cpu_id"]
ret = event["ret"]
fds_length = event["fds_length"]
overflow = event["overflow"]
ret = event["ret"]
fds_length = event["fds_length"]
overflow = event["overflow"]
# check the proper working of INT_MAX maxevent value
if ret == -22 and fds_length == 0 and overflow == 0:
# check the proper working of INT_MAX maxevent value
if ret == -22 and fds_length == 0 and overflow == 0:
- self.expect["epoll_wait_max_out"] = 1
+ self.expect["epoll_wait_exit"]["epoll_wait_max_out"] = 1
+
+ # Save values of local variables to print in case of test failure
+ self.recorded_values["epoll_wait_exit"] = locals()
if __name__ == "__main__":
if __name__ == "__main__":
projects / lttng-tools.git / commitdiff
