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
|
import pytest
from pr2test.context_manager import make_test_matrix
from pr2test.marks import require_root
from pr2test.tools import interface_exists
pytestmark = [require_root()]
tnl_matrix = make_test_matrix(
targets=['local', 'netns'],
types=['gre', 'ipip', 'sit'],
dbs=['sqlite3/:memory:', 'postgres/pr2test'],
)
def _test_tunnel_endpoints(context, state):
ifname = context.new_ifname
ipaddr_local1 = context.new_ipaddr
ipaddr_local2 = context.new_ipaddr
ipaddr_remote = context.new_ipaddr
kind = context.kind
(
context.ndb.interfaces.create(
**{
'ifname': ifname,
'state': state,
'kind': kind,
f'{kind}_local': ipaddr_local1,
f'{kind}_remote': ipaddr_remote,
}
).commit()
)
def match(ifname, ipaddr):
return (
lambda x: x.get_nested('IFLA_LINKINFO', 'IFLA_INFO_KIND') == kind
and x.get_attr('IFLA_IFNAME') == ifname
and x.get_nested(
'IFLA_LINKINFO',
'IFLA_INFO_DATA',
'IFLA_%s_LOCAL' % kind.upper(),
)
== ipaddr
)
assert interface_exists(context.netns, match(ifname, ipaddr_local1))
(
context.ndb.interfaces[ifname]
.set(f'{kind}_local', ipaddr_local2)
.commit()
)
assert interface_exists(context.netns, match(ifname, ipaddr_local2))
@pytest.mark.parametrize('context', tnl_matrix, indirect=True)
def test_tunnel_endpoints_down(context):
return _test_tunnel_endpoints(context, 'down')
@pytest.mark.parametrize('context', tnl_matrix, indirect=True)
def test_tunnel_endpoints_up(context):
return _test_tunnel_endpoints(context, 'up')
|
