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
|
import errno
import time
import pytest
from pr2test.marks import require_root
from pyroute2 import L2tp, NetlinkError
pytestmark = [require_root()]
@pytest.fixture
def l2ctx(context):
try:
context.l2tp = L2tp()
except NetlinkError as e:
if e.code == errno.ENOENT:
pytest.skip('L2TP netlink API not available')
raise
context.local_ip = context.get_ipaddr(r=0) # <- 0.255
gateway = context.get_ipaddr(r=0) # <- 0.254 (gateway can not be 0.255)
context.remote_ip = context.get_ipaddr(r=1)
context.l2tpeth0 = context.new_ifname
context.ndb.interfaces[context.default_interface.ifname].add_ip(
f'{context.local_ip}/24'
).commit()
context.ndb.routes.create(
dst=str(context.ipnets[1].network), dst_len=24, gateway=gateway
).commit()
yield context
try:
context.l2tp.delete_session(tunnel_id=2324, session_id=3435)
except Exception:
pass
try:
context.l2tp.delete_tunnel(tunnel_id=2324)
except Exception:
pass
context.l2tp.close()
@pytest.mark.xfail(reason='flaky test, only to collect failure logs')
def test_complete(l2ctx):
# 1. create tunnel
l2ctx.l2tp.create_tunnel(
tunnel_id=2324,
peer_tunnel_id=2425,
remote=l2ctx.remote_ip,
local=l2ctx.local_ip,
udp_dport=32000,
udp_sport=32000,
encap="udp",
)
tunnel = l2ctx.l2tp.get_tunnel(tunnel_id=2324)
assert tunnel.get_attr("L2TP_ATTR_CONN_ID") == 2324
assert tunnel.get_attr("L2TP_ATTR_PEER_CONN_ID") == 2425
assert tunnel.get_attr("L2TP_ATTR_IP_DADDR") == l2ctx.remote_ip
assert tunnel.get_attr("L2TP_ATTR_IP_SADDR") == l2ctx.local_ip
assert tunnel.get_attr("L2TP_ATTR_UDP_DPORT") == 32000
assert tunnel.get_attr("L2TP_ATTR_UDP_SPORT") == 32000
assert tunnel.get_attr("L2TP_ATTR_ENCAP_TYPE") == 0 # 0 == UDP
assert tunnel.get_attr("L2TP_ATTR_DEBUG") == 0
# 2. create session
l2ctx.l2tp.create_session(
tunnel_id=2324,
session_id=3435,
peer_session_id=3536,
ifname=l2ctx.l2tpeth0,
)
session = l2ctx.l2tp.get_session(tunnel_id=2324, session_id=3435)
assert session.get_attr("L2TP_ATTR_SESSION_ID") == 3435
assert session.get_attr("L2TP_ATTR_PEER_SESSION_ID") == 3536
assert session.get_attr("L2TP_ATTR_DEBUG") == 0
# setting up DEBUG -> 95, operation not supported; review the test
# # 3. modify session
# l2ctx.l2tp.modify_session(tunnel_id=2324, session_id=3435, debug=True)
# session = l2ctx.l2tp.get_session(tunnel_id=2324, session_id=3435)
# assert session[0].get_attr("L2TP_ATTR_DEBUG") == 1
# setting up DEBUG -> 95, operation not supported; review the test
# # 4. modify tunnel
# l2ctx.l2tp.modify_tunnel(tunnel_id=2324, debug=True)
# tunnel = l2ctx.l2tp.get_tunnel(tunnel_id=2324)
# assert tunnel[0].get_attr("L2TP_ATTR_DEBUG") == 1
# 5. destroy session
l2ctx.l2tp.delete_session(tunnel_id=2324, session_id=3435)
for _ in range(5):
try:
l2ctx.l2tp.get_session(tunnel_id=2324, session_id=3435)
except NetlinkError:
time.sleep(0.1)
continue
break
else:
raise Exception('could not remove L2TP session')
# 6. destroy tunnel
l2ctx.l2tp.delete_tunnel(tunnel_id=2324)
for _ in range(5):
try:
l2ctx.l2tp.get_tunnel(tunnel_id=2324)
except NetlinkError:
time.sleep(0.1)
continue
break
else:
raise Exception('could not remove L2TP tunnel')
|
