-
Notifications
You must be signed in to change notification settings - Fork 0
/
run.py
executable file
·134 lines (98 loc) · 3.43 KB
/
run.py
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
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
#!/usr/bin/env python3
import os
import argparse
import sys
import config
from harness.utils import clean_traffic
from harness.config_virtual import virtual_ctxs
from harness.setup import create_virtual_servers_from_list
from harness.client import start_client
from harness.server import start_server
from harness.xdp_case import (XDPCaseNetwork, XDPCaseBPTR)
def run_bptr(unittest_args):
"""Start a client in an offline mode."""
ctxs = config.remote_server_ctxs
ctxs.get_local_main().xdp_mode = None
for i in range(ctxs.server_count()):
if ctxs.get_remote(i) is None:
print("BPTR mode requires all ContextClient "
"to have a predefined remote context.")
return -1
ctxs.get_local(i).index = i + 1
res = start_client(ctxs, XDPCaseBPTR, unittest_args)
return res
def run_client(unittest_args):
"""Build virtual servers and start a client using network."""
created_servers_procs = []
netns = []
try:
clean_traffic("default")
(created_servers_procs, netns) = \
create_virtual_servers_from_list(virtual_ctxs, None)
for i in range(config.remote_server_ctxs.server_count()):
clean_traffic(config.remote_server_ctxs.get_local(i).iface)
res = start_client(config.remote_server_ctxs,
XDPCaseNetwork, unittest_args)
finally:
for i in created_servers_procs:
try:
i.terminate()
except KeyboardInterrupt:
pass
for i in netns:
i.close()
i.remove()
return res
def run_server():
"""Start a server with configuration from config.py."""
config.local_server_ctx.local.fill_missing()
clean_traffic(config.local_server_ctx.local.iface)
start_server(config.local_server_ctx)
def parse_args():
"""Parse command line arguments."""
parser = argparse.ArgumentParser(description="""
XDP testsuite
For configuration, use the config.py file.
Example usage:
./run.py client test_general.ReturnValuesBasic
""")
test_names = (
"tests",
{
"help": """A list of patters defining tests to run.
The patterns are using the unittest format -
modules, classes and methods separated by dots.""",
"nargs": argparse.REMAINDER,
"default": None,
}
)
type_subparser = parser.add_subparsers(dest="type", required=True)
server_parser = type_subparser.add_parser(
"server", help="Start a standalone response server."
)
client_parser = type_subparser.add_parser(
"client", help="Start testing using a network."
)
client_parser.add_argument(test_names[0], **test_names[1])
bptr_parser = type_subparser.add_parser(
"bptr", help="Start testing using BPF_PROG_TEST_RUN command."
)
bptr_parser.add_argument(test_names[0], **test_names[1])
return parser.parse_args()
def main():
args = parse_args()
res = 0
if os.getuid() != 0:
print("Admin privileges required.")
sys.exit(-1)
if args.type == "client":
unittest_args = {"tests": args.tests}
res = run_client(unittest_args)
elif args.type == "server":
run_server()
elif args.type == "bptr":
unittest_args = {"tests": args.tests}
res = run_bptr(unittest_args)
sys.exit(res)
if __name__ == "__main__":
main()